wowlikon/LibraryAPI
1
0
Fork 0
mirror of https://github.com/wowlikon/LibraryAPI.git synced 2025-12-11 21:30:46 +00:00
Code Issues Packages Projects Releases Wiki Activity

Global refactoring of the project to use poetry and implement tests,

Browse Source 
fixing bugs, changing the handling of dto and db models, preparing to
add new functionality
This commit is contained in:
wowlikon
2025-06-24 13:30:35 +03:00
parent 51a6ba75c0
commit 6658d773bf
58 changed files with 2521 additions and 1008 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
.env Normal file
View File

@@ -0,0 +1,4 @@
POSTGRES_USER = "postgres"
POSTGRES_PASSWORD = "password"
POSTGRES_DB = "mydatabase"
POSTGRES_SERVER = "db"

3
.env.example
View File

@@ -1,3 +0,0 @@
POSTGRES_USER=postgres
POSTGRES_PASSWORD=password
POSTGRES_DB=mydatabase

143
.gitignore vendored
View File

@@ -1,4 +1,141 @@
!.env.example
.env
# Byte-compiled / optimized / DLL files
__pycache__/
*.py[cod]
*$py.class
# C extensions
*.so
# Distribution / packaging
.Python
build/
develop-eggs/
dist/
downloads/
eggs/
.eggs/
lib/
lib64/
parts/
sdist/
var/
wheels/
pip-wheel-metadata/
share/python-wheels/
*.egg-info/
.installed.cfg
*.egg
MANIFEST
# PyInstaller
# Usually these files are written by a python script from a template
# before PyInstaller builds the exe, so as to inject date/other infos into it.
*.manifest
*.spec
# Installer logs
pip-log.txt
pip-delete-this-directory.txt
# Unit test / coverage reports
htmlcov/
.tox/
.nox/
.coverage
.coverage.*
.cache
nosetests.xml
coverage.xml
*.cover
*.py,cover
.hypothesis/
.pytest_cache/
# Translations
*.mo
*.pot
# Django stuff:
*.log
local_settings.py
db.sqlite3
db.sqlite3-journal
# Flask stuff:
instance/
.webassets-cache
# Scrapy stuff:
.scrapy
# Sphinx documentation
docs/_build/
# PyBuilder
target/
# Jupyter Notebook
.ipynb_checkpoints
# IPython
profile_default/
ipython_config.py
# pyenv
.python-version
# pipenv
# According to pypa/pipenv#598, it is recommended to include Pipfile.lock in version control.
# However, in case of collaboration, if having platform-specific dependencies or dependencies
# having no cross-platform support, pipenv may install dependencies that don't work, or not
# install all needed dependencies.
#Pipfile.lock
# PEP 582; used by e.g. github.com/David-OConnor/pyflow
__pypackages__/
# Celery stuff
celerybeat-schedule
celerybeat.pid
# SageMath parsed files
*.sage.py
# Environments
# .env
.venv
env/
venv/
ENV/
env.bak/
venv.bak/
# Spyder project settings
.spyderproject
.spyproject
# Rope project settings
.ropeproject
# mkdocs documentation
/site
# mypy
.mypy_cache/
.dmypy.json
dmypy.json
# Pyre type checker
.pyre/
# Jetbrains' IDEs
.idea
# VS code
.vscode
# JUPITER
*.ipynb
# Postgres data
data/

30
Dockerfile
View File

@@ -1,12 +1,22 @@
FROM python:3.11
FROM python:3.13 as requirements-stage
WORKDIR /tmp
RUN pip install poetry
RUN poetry self add poetry-plugin-export
COPY ./pyproject.toml ./poetry.lock* /tmp/
RUN poetry export -f requirements.txt --output requirements.txt --with dev --without-hashes
FROM python:3.13
ENV PYTHONDONTWRITEBYTECODE=1
ENV PYTHONUNBUFFERED=1
RUN apt-get update \
&& apt-get -y install gcc postgresql \
&& apt-get clean # netcat
RUN pip install --upgrade pip
WORKDIR /code
COPY requirements.txt ./
RUN pip install --no-cache-dir -r requirements.txt
COPY ./src/alembic.ini ./
COPY ./src/app /code/app
COPY ./tests /code/tests
CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]
COPY --from=requirements-stage /tmp/requirements.txt ./requirements.txt
RUN pip install --no-cache-dir --upgrade -r ./requirements.txt
COPY . .
ENV PYTHONPATH=.

20
Makefile
View File

@@ -1,20 +0,0 @@
.PHONY: help build up start down destroy stop restart logs db-shell
help:
@echo "Available commands:"
@echo " make build - Build the Docker images"
@echo " make up - Start the containers"
@echo " make down - Stop and remove the containers"
@echo " make db-shell - Access the database shell"
build:
docker-compose -f docker-compose.yml build
up:
docker-compose -f docker-compose.yml up -d
down:
docker-compose -f docker-compose.yml down
db-shell:
docker-compose -f docker-compose.yml exec timescale psql -Upostgres

6
README.md
View File

@@ -1,4 +1,6 @@
# Book API
# LibraryAPI
## WARNING: The documentation is now out of date.
This project is a test web application built using FastAPI, a modern web framework for creating APIs in Python. It showcases the use of Pydantic for data validation, SQLModel for database interactions, Alembic for migration management, PostgreSQL as the database system, and Docker Compose for easy deployment.
@@ -18,7 +20,7 @@ For development:
1. Clone the repository:
```bash
git clone https://github.com/wowlikon/bookapi.git
git clone https://github.com/wowlikon/libraryapi.git
```
2. Navigate to the project directory:

24
alembic.ini Normal file → Executable file
View File

@@ -2,23 +2,22 @@
[alembic]
# path to migration scripts
script_location = app/migrations
script_location = migrations
# template used to generate migration file names; The default value is %%(rev)s_%%(slug)s
# Uncomment the line below if you want the files to be prepended with date and time
# see https://alembic.sqlalchemy.org/en/latest/tutorial.html#editing-the-ini-file
# for all available tokens
# file_template = %%(year)d_%%(month).2d_%%(day).2d_%%(hour).2d%%(minute).2d-%%(rev)s_%%(slug)s
# sys.path path, will be prepended to sys.path if present.
# defaults to the current working directory.
prepend_sys_path = .
path_separator = os
# timezone to use when rendering the date within the migration file
# as well as the filename.
# If specified, requires the python>=3.9 or backports.zoneinfo library.
# Any required deps can installed by adding `alembic[tz]` to the pip requirements
# string value is passed to ZoneInfo()
# If specified, requires the python-dateutil library that can be
# installed by adding `alembic[tz]` to the pip requirements
# string value is passed to dateutil.tz.gettz()
# leave blank for localtime
# timezone =
@@ -51,16 +50,11 @@ prepend_sys_path = .
# version_path_separator = space
version_path_separator = os # Use os.pathsep. Default configuration used for new projects.
# set to 'true' to search source files recursively
# in each "version_locations" directory
# new in Alembic version 1.10
# recursive_version_locations = false
# the output encoding used when revision files
# are written from script.py.mako
# output_encoding = utf-8
sqlalchemy.url = driver://user:pass@localhost/dbname
sqlalchemy.url = postgresql+asyncpg://user:pwd@db:5432/foo
[post_write_hooks]
@@ -74,12 +68,6 @@ sqlalchemy.url = driver://user:pass@localhost/dbname
# black.entrypoint = black
# black.options = -l 79 REVISION_SCRIPT_FILENAME
# lint with attempts to fix using "ruff" - use the exec runner, execute a binary
# hooks = ruff
# ruff.type = exec
# ruff.executable = %(here)s/.venv/bin/ruff
# ruff.options = --fix REVISION_SCRIPT_FILENAME
# Logging configuration
[loggers]
keys = root,sqlalchemy,alembic

14
app/database.py
View File

@@ -1,14 +0,0 @@
from sqlmodel import create_engine, SQLModel, Session
from decouple import config
# Get database configuration
DATABASE_URL = config('DATABASE_URL', cast=str, default='sqlite:///./bookapi.db')
# Create database engine
engine = create_engine(str(DATABASE_URL), echo=True)
# SQLModel.metadata.create_all(engine)
# Get database session
def get_session():
with Session(engine) as session:
yield session

220
app/main.py
View File

@@ -1,220 +0,0 @@
from datetime import datetime
from pathlib import Path
from typing import Dict, List
from alembic import command
from alembic.config import Config
from fastapi import FastAPI, Depends, Request, HTTPException
from fastapi.responses import HTMLResponse, JSONResponse
from fastapi.templating import Jinja2Templates
from sqlmodel import SQLModel, Session, select
from .database import engine, get_session
from .models import Author, AuthorBase, Book, BookBase, AuthorBookLink
alembic_cfg = Config("alembic.ini")
templates = Jinja2Templates(directory=Path(__file__).parent / "templates")
app = FastAPI(
title="LibraryAPI",
description="This is a sample API for managing authors and books.",
version="1.0.1",
openapi_tags=[
{
"name": "authors",
"description": "Operations with authors.",
},
{
"name": "books",
"description": "Operations with books.",
},
{
"name": "relations",
"description": "Operations with relations.",
},
{
"name": "misc",
"description": "Miscellaneous operations.",
}
]
)
def get_info() -> Dict:
return {
"status": "ok",
"app_info": {
"title": app.title,
"version": app.version,
"description": app.description,
},
"server_time": datetime.now().isoformat(),
}
# Initialize the database
@app.on_event("startup")
def on_startup():
# Apply database migrations
with engine.begin() as connection:
alembic_cfg.attributes['connection'] = connection
command.upgrade(alembic_cfg, "head")
# Root endpoint
@app.get("/", response_class=HTMLResponse)
async def root(request: Request):
return templates.TemplateResponse("index.html", {"request": request, "data": get_info()})
# API Information endpoint
@app.get("/api/info", tags=["misc"])
async def api_info():
return JSONResponse(content=get_info())
# Create an author
@app.post("/authors/", response_model=Author, tags=["authors"])
def create_author(author: AuthorBase, session: Session = Depends(get_session)):
db_author = Author(name=author.name)
session.add(db_author)
session.commit()
session.refresh(db_author)
return db_author
# Read authors
@app.get("/authors/", response_model=List[Author], tags=["authors"])
def read_authors(session: Session = Depends(get_session)):
authors = session.exec(select(Author)).all()
return authors
# Update an author
@app.put("/authors/{author_id}", response_model=Author, tags=["authors"])
def update_author(author_id: int, author: AuthorBase, session: Session = Depends(get_session)):
db_author = session.get(Author, author_id)
if not db_author:
raise HTTPException(status_code=404, detail="Author not found")
db_author.name = author.name
session.commit()
session.refresh(db_author)
return db_author
# Delete an author
@app.delete("/authors/{author_id}", response_model=AuthorBase, tags=["authors"])
def delete_author(author_id: int, session: Session = Depends(get_session)):
db_author = session.get(Author, author_id)
if not db_author:
raise HTTPException(status_code=404, detail="Author not found")
session.delete(db_author)
author = AuthorBase(name=db_author.name)
session.commit()
return author
# Create a book with authors
@app.post("/books/", response_model=Book, tags=["books"])
def create_book(book: BookBase, session: Session = Depends(get_session)):
db_book = Book(title=book.title, description=book.description)
session.add(db_book)
session.commit()
session.refresh(db_book)
return db_book
# Read books
@app.get("/books/", response_model=List[Book], tags=["books"])
def read_books(session: Session = Depends(get_session)):
books = session.exec(select(Book)).all()
return books
# Update a book with authors
@app.put("/books/{book_id}", response_model=Book, tags=["books"])
def update_book(book_id: int, book: BookBase, session: Session = Depends(get_session)):
db_book = session.get(Book, book_id)
if not db_book:
raise HTTPException(status_code=404, detail="Book not found")
db_book.title = book.title
db_book.description = book.description
session.commit()
session.refresh(db_book)
return db_book
# Delete a book
@app.delete("/books/{book_id}", response_model=BookBase, tags=["books"])
def delete_book(book_id: int, session: Session = Depends(get_session)):
db_book = session.get(Book, book_id)
if not db_book:
raise HTTPException(status_code=404, detail="Book not found")
book = Book(title=db_book.title, description=db_book.description)
session.delete(db_book)
session.commit()
return book
# Add author to book
@app.post("/relationships/", response_model=AuthorBookLink, tags=["relations"])
def add_author_to_book(author_id: int, book_id: int, session: Session = Depends(get_session)):
# Check if author and book exist
author = session.get(Author, author_id)
if not author:
raise HTTPException(status_code=404, detail="Author not found")
book = session.get(Book, book_id)
if not book:
raise HTTPException(status_code=404, detail="Book not found")
# Check if relationship already exists
existing_link = session.exec(
select(AuthorBookLink)
.where(AuthorBookLink.author_id == author_id)
.where(AuthorBookLink.book_id == book_id)
).first()
if existing_link:
raise HTTPException(status_code=400, detail="Relationship already exists")
# Create new relationship
link = AuthorBookLink(author_id=author_id, book_id=book_id)
session.add(link)
session.commit()
session.refresh(link)
return link
# Remove author from book
@app.delete("/relationships/", tags=["relations"])
def remove_author_from_book(author_id: int, book_id: int, session: Session = Depends(get_session)):
# Find the relationship
link = session.exec(
select(AuthorBookLink)
.where(AuthorBookLink.author_id == author_id)
.where(AuthorBookLink.book_id == book_id)
).first()
if not link:
raise HTTPException(status_code=404, detail="Relationship not found")
session.delete(link)
session.commit()
return {"message": "Relationship removed successfully"}
# Get all authors for a book
@app.get("/books/{book_id}/authors/", response_model=List[Author], tags=["books", "relations"])
def get_authors_for_book(book_id: int, session: Session = Depends(get_session)):
book = session.get(Book, book_id)
if not book:
raise HTTPException(status_code=404, detail="Book not found")
authors = session.exec(
select(Author)
.join(AuthorBookLink)
.where(AuthorBookLink.book_id == book_id)
).all()
return authors
# Get all books for an author
@app.get("/authors/{author_id}/books/", response_model=List[Book], tags=["authors", "relations"])
def get_books_for_author(author_id: int, session: Session = Depends(get_session)):
author = session.get(Author, author_id)
if not author:
raise HTTPException(status_code=404, detail="Author not found")
books = session.exec(
select(Book)
.join(AuthorBookLink)
.where(AuthorBookLink.author_id == author_id)
).all()
return books

1
app/migrations/README
View File

@@ -1 +0,0 @@
Generic single-database configuration.

BIN
app/migrations/versions/__pycache__/d266fdc61e99_init.cpython-311.pyc
View File

Binary file not shown.

41
app/models.py
View File

@@ -1,41 +0,0 @@
from typing import List
from sqlmodel import SQLModel, Field, Relationship
# Relationship model
class AuthorBookLink(SQLModel, table=True):
author_id: int | None = Field(default=None, foreign_key="author.id", primary_key=True)
book_id: int | None = Field(default=None, foreign_key="book.id", primary_key=True)
# Author DTO model
class AuthorBase(SQLModel):
name: str
class Config: # pyright: ignore
json_schema_extra = {
"example": {
"name": "author_name",
}
}
# Author DB model
class Author(AuthorBase, table=True):
id: int | None = Field(default=None, primary_key=True, index=True)
books: List["Book"] = Relationship(back_populates="authors", link_model=AuthorBookLink)
# Book DTO model
class BookBase(SQLModel):
title: str
description: str
class Config: # pyright: ignore
json_schema_extra = {
"example": {
"title": "book_title",
"description": "book_description",
}
}
# Book DB model
class Book(BookBase, table=True):
id: int | None = Field(default=None, primary_key=True, index=True)
authors: List[Author] = Relationship(back_populates="books", link_model=AuthorBookLink)

35
docker-compose.yml
View File

@@ -1,33 +1,30 @@
services:
db:
image: postgres:15-alpine
environment:
POSTGRES_USER: ${POSTGRES_USER}
POSTGRES_PASSWORD: ${POSTGRES_PASSWORD}
POSTGRES_DB: ${POSTGRES_DB}
ports:
- "5432:5432"
container_name: db_library
image: postgres
expose:
- 5432
volumes:
- postgres_data:/var/lib/postgresql/data
- ./data/db:/var/lib/postgresql/data
env_file:
- ./.env
api:
container_name: api_library
build: .
environment:
DATABASE_URL: postgresql://${POSTGRES_USER}:${POSTGRES_PASSWORD}@db:5432/${POSTGRES_DB}
command: bash -c "alembic upgrade head && uvicorn library_service.main:app --reload --host 0.0.0.0 --port 8000"
volumes:
- .:/code
ports:
- "8000:8000"
volumes:
- ./src/app/migrations/versions:/code/app/migrations/versions
depends_on:
- db
test:
tests:
container_name: tests
build: .
command: pytest
environment:
DATABASE_URL: postgresql://${POSTGRES_USER}:${POSTGRES_PASSWORD}@db:5432/${POSTGRES_DB}
command: bash -c "pytest tests/test_misc.py"
volumes:
- .:/code
depends_on:
- db
volumes:
postgres_data:

0
app/__init__.py → library_service/__init__.py
View File

6
library_service/api.py Normal file
View File

@@ -0,0 +1,6 @@
from fastapi import APIRouter
import asyncpg
router = APIRouter(
prefix='/devices'
)

28
library_service/main.py Normal file
View File

@@ -0,0 +1,28 @@
from alembic import command
from alembic.config import Config
from contextlib import asynccontextmanager
from fastapi import FastAPI
from toml import load
from .settings import engine, get_app
from .routers import api_router
from .routers.misc import get_info
app = get_app()
alembic_cfg = Config("alembic.ini")
@asynccontextmanager
async def lifespan(app: FastAPI):
print("[+] Initializing...")
# Initialize the database
with engine.begin() as connection:
alembic_cfg.attributes['connection'] = connection
command.upgrade(alembic_cfg, "head")
print("[+] Starting...")
yield # Here FastAPI will start handling requests;
print("[+] Application shutdown")
# Include routers
app.include_router(api_router)

2
library_service/models/__init__.py Normal file
View File

@@ -0,0 +1,2 @@
from .dto import *
from .db import *

7
library_service/models/db/__init__.py Normal file
View File

@@ -0,0 +1,7 @@
from .author import Author
from .book import Book
from .links import AuthorBookLink, AuthorWithBooks, BookWithAuthors
__all__ = [
'Author', 'Book', 'AuthorBookLink', 'AuthorWithBooks', 'BookWithAuthors'
]

14
library_service/models/db/author.py Normal file
View File

@@ -0,0 +1,14 @@
from typing import List, Optional, TYPE_CHECKING
from sqlmodel import SQLModel, Field, Relationship
from ..dto.author import AuthorBase
from .links import AuthorBookLink
if TYPE_CHECKING:
from .book import Book
class Author(AuthorBase, table=True):
id: Optional[int] = Field(default=None, primary_key=True, index=True)
books: List["Book"] = Relationship(
back_populates="authors",
link_model=AuthorBookLink
)

14
library_service/models/db/book.py Normal file
View File

@@ -0,0 +1,14 @@
from typing import List, Optional, TYPE_CHECKING
from sqlmodel import SQLModel, Field, Relationship
from ..dto.book import BookBase
from .links import AuthorBookLink
if TYPE_CHECKING:
from .author import Author
class Book(BookBase, table=True):
id: Optional[int] = Field(default=None, primary_key=True, index=True)
authors: List["Author"] = Relationship(
back_populates="books",
link_model=AuthorBookLink
)

15
library_service/models/db/links.py Normal file
View File

@@ -0,0 +1,15 @@
from sqlmodel import SQLModel, Field
from typing import List
from library_service.models.dto.author import AuthorRead
from library_service.models.dto.book import BookRead
class AuthorBookLink(SQLModel, table=True):
author_id: int | None = Field(default=None, foreign_key="author.id", primary_key=True)
book_id: int | None = Field(default=None, foreign_key="book.id", primary_key=True)
class AuthorWithBooks(AuthorRead):
books: List[BookRead] = Field(default_factory=list)
class BookWithAuthors(BookRead):
authors: List[AuthorRead] = Field(default_factory=list)

15
library_service/models/dto/__init__.py Normal file
View File

@@ -0,0 +1,15 @@
from .author import (
AuthorBase, AuthorCreate, AuthorUpdate,
AuthorRead, AuthorList
)
from .book import (
BookBase, BookCreate, BookUpdate,
BookRead, BookList
)
# from .common import PaginatedResponse
__all__ = [
'AuthorBase', 'AuthorCreate', 'AuthorUpdate', 'AuthorRead', 'AuthorList',
'BookBase', 'BookCreate', 'BookUpdate', 'BookRead', 'BookList',
# 'PaginatedResponse'
]

25
library_service/models/dto/author.py Normal file
View File

@@ -0,0 +1,25 @@
from sqlmodel import SQLModel
from pydantic import ConfigDict
from typing import Optional, List
class AuthorBase(SQLModel):
name: str
model_config = ConfigDict( #pyright: ignore
json_schema_extra={
"example": {"name": "author_name"}
}
)
class AuthorCreate(AuthorBase):
pass
class AuthorUpdate(SQLModel):
name: Optional[str] = None
class AuthorRead(AuthorBase):
id: int
class AuthorList(SQLModel):
authors: List[AuthorRead]
total: int

30
library_service/models/dto/book.py Normal file
View File

@@ -0,0 +1,30 @@
from sqlmodel import SQLModel
from pydantic import ConfigDict
from typing import Optional, List
class BookBase(SQLModel):
title: str
description: str
model_config = ConfigDict( #pyright: ignore
json_schema_extra={
"example": {
"title": "book_title",
"description": "book_description"
}
}
)
class BookCreate(BookBase):
pass
class BookUpdate(SQLModel):
title: Optional[str] = None
description: Optional[str] = None
class BookRead(BookBase):
id: int
class BookList(SQLModel):
books: List[BookRead]
total: int

1
src/app/routers/__init__.py → library_service/routers/__init__.py
View File

@@ -7,6 +7,7 @@ from .misc import router as misc_router
api_router = APIRouter()
# Including all routers
api_router.include_router(authors_router)
api_router.include_router(books_router)
api_router.include_router(relationships_router)

80
library_service/routers/authors.py Normal file
View File

@@ -0,0 +1,80 @@
from fastapi import APIRouter, Depends, HTTPException
from sqlmodel import Session, select
from library_service.settings import get_session
from library_service.models.db import Author, AuthorBookLink, Book, AuthorWithBooks
from library_service.models.dto import (
AuthorCreate, AuthorUpdate, AuthorRead,
AuthorList, BookRead
)
router = APIRouter(prefix="/authors", tags=["authors"])
# Create an author
@router.post("/", response_model=AuthorRead)
def create_author(author: AuthorCreate, session: Session = Depends(get_session)):
db_author = Author(**author.model_dump())
session.add(db_author)
session.commit()
session.refresh(db_author)
return AuthorRead(**db_author.model_dump())
# Read authors
@router.get("/", response_model=AuthorList)
def read_authors(session: Session = Depends(get_session)):
authors = session.exec(select(Author)).all()
return AuthorList(
authors=[AuthorRead(**author.model_dump()) for author in authors],
total=len(authors)
)
# Read an author with their books
@router.get("/{author_id}", response_model=AuthorWithBooks)
def get_author(author_id: int, session: Session = Depends(get_session)):
author = session.get(Author, author_id)
if not author:
raise HTTPException(status_code=404, detail="Author not found")
books = session.exec(
select(Book)
.join(AuthorBookLink)
.where(AuthorBookLink.author_id == author_id)
).all()
book_reads = [BookRead(**book.model_dump()) for book in books]
author_data = author.model_dump()
author_data['books'] = book_reads
return AuthorWithBooks(**author_data)
# Update an author
@router.put("/{author_id}", response_model=AuthorRead)
def update_author(
author_id: int,
author: AuthorUpdate,
session: Session = Depends(get_session)
):
db_author = session.get(Author, author_id)
if not db_author:
raise HTTPException(status_code=404, detail="Author not found")
update_data = author.model_dump(exclude_unset=True)
for field, value in update_data.items():
setattr(db_author, field, value)
session.commit()
session.refresh(db_author)
return AuthorRead(**db_author.model_dump())
# Delete an author
@router.delete("/{author_id}", response_model=AuthorRead)
def delete_author(author_id: int, session: Session = Depends(get_session)):
author = session.get(Author, author_id)
if not author:
raise HTTPException(status_code=404, detail="Author not found")
author_read = AuthorRead(**author.model_dump())
session.delete(author)
session.commit()
return author_read

77
library_service/routers/books.py Normal file
View File

@@ -0,0 +1,77 @@
from fastapi import APIRouter, Depends, HTTPException
from sqlmodel import Session, select
from typing import Deque, List
from library_service.settings import get_session
from library_service.models.db import Book, Author, AuthorBookLink, BookWithAuthors
from library_service.models.dto import (
BookCreate, BookUpdate, BookRead,
BookList, AuthorRead
)
router = APIRouter(prefix="/books", tags=["books"])
# Create a book
@router.post("/", response_model=Book)
def create_book(book: BookCreate, session: Session = Depends(get_session)):
db_book = Book(**book.model_dump())
session.add(db_book)
session.commit()
session.refresh(db_book)
return BookRead(**db_book.model_dump())
# Read books
@router.get("/", response_model=BookList)
def read_books(session: Session = Depends(get_session)):
books = session.exec(select(Book)).all()
return BookList(
books=[BookRead(**book.model_dump()) for book in books],
total=len(books)
)
# Read a book
@router.get("/{book_id}", response_model=BookWithAuthors)
def get_book(book_id: int, session: Session = Depends(get_session)):
book = session.get(Book, book_id)
if not book:
raise HTTPException(status_code=404, detail="Book not found")
return BookWithAuthors(**book.model_dump())
# Update a book
@router.put("/{book_id}", response_model=Book)
def update_book(book_id: int, book: BookUpdate, session: Session = Depends(get_session)):
db_book = session.get(Book, book_id)
if not db_book:
raise HTTPException(status_code=404, detail="Book not found")
db_book.title = book.title or db_book.title
db_book.description = book.description or db_book.description
session.commit()
session.refresh(db_book)
return db_book
# Delete a book
@router.delete("/{book_id}", response_model=BookRead)
def delete_book(book_id: int, session: Session = Depends(get_session)):
book = session.get(Book, book_id)
if not book:
raise HTTPException(status_code=404, detail="Book not found")
book_read = BookRead(id=(book.id or 0), title=book.title, description=book.description)
session.delete(book)
session.commit()
return book_read
# Get all authors for a book
@router.get("/{book_id}/authors/", response_model=List[AuthorRead])
def get_authors_for_book(book_id: int, session: Session = Depends(get_session)):
book = session.get(Book, book_id)
if not book:
raise HTTPException(status_code=404, detail="Book not found")
authors = session.exec(
select(Author)
.join(AuthorBookLink)
.where(AuthorBookLink.book_id == book_id)
).all()
return [AuthorRead(**author.model_dump()) for author in authors]

16
src/app/routers/misc.py → library_service/routers/misc.py
View File

@@ -1,15 +1,21 @@
from fastapi import APIRouter, Request
from fastapi import APIRouter, Request, FastAPI
from fastapi.params import Depends
from fastapi.responses import HTMLResponse, JSONResponse
from fastapi.templating import Jinja2Templates
from pathlib import Path
from datetime import datetime
from typing import Dict
# Инициализация шаблонов
from httpx import get
from library_service.settings import get_app
# Templates initialization
templates = Jinja2Templates(directory=Path(__file__).parent.parent / "templates")
router = APIRouter(tags=["misc"])
# Formatted information about the application
def get_info(app) -> Dict:
return {
"status": "ok",
@@ -23,10 +29,10 @@ def get_info(app) -> Dict:
# Root endpoint
@router.get("/", response_class=HTMLResponse)
async def root(request: Request, app=None):
return templates.TemplateResponse("index.html", {"request": request, "data": get_info(app)})
async def root(request: Request, app=Depends(get_app)):
return templates.TemplateResponse(request, "index.html", get_info(app))
# API Information endpoint
@router.get("/api/info")
async def api_info(app=None):
async def api_info(app=Depends(get_app)):
return JSONResponse(content=get_info(app))

8
src/app/routers/relationships.py → library_service/routers/relationships.py
View File

@@ -2,15 +2,14 @@ from fastapi import APIRouter, Depends, HTTPException
from sqlmodel import Session, select
from typing import List, Dict
from ..database import get_session
from ..models import Author, Book, AuthorBookLink
from library_service.settings import get_session
from library_service.models.db import Book, Author, AuthorBookLink
router = APIRouter(prefix="/relationships", tags=["relations"])
# Add author to book
@router.post("/", response_model=AuthorBookLink)
def add_author_to_book(author_id: int, book_id: int, session: Session = Depends(get_session)):
# Check if author and book exist
author = session.get(Author, author_id)
if not author:
raise HTTPException(status_code=404, detail="Author not found")
@@ -19,7 +18,6 @@ def add_author_to_book(author_id: int, book_id: int, session: Session = Depends(
if not book:
raise HTTPException(status_code=404, detail="Book not found")
# Check if relationship already exists
existing_link = session.exec(
select(AuthorBookLink)
.where(AuthorBookLink.author_id == author_id)
@@ -29,7 +27,6 @@ def add_author_to_book(author_id: int, book_id: int, session: Session = Depends(
if existing_link:
raise HTTPException(status_code=400, detail="Relationship already exists")
# Create new relationship
link = AuthorBookLink(author_id=author_id, book_id=book_id)
session.add(link)
session.commit()
@@ -39,7 +36,6 @@ def add_author_to_book(author_id: int, book_id: int, session: Session = Depends(
# Remove author from book
@router.delete("/", response_model=Dict[str, str])
def remove_author_from_book(author_id: int, book_id: int, session: Session = Depends(get_session)):
# Find the relationship
link = session.exec(
select(AuthorBookLink)
.where(AuthorBookLink.author_id == author_id)

52
library_service/settings.py Normal file
View File

@@ -0,0 +1,52 @@
import os
from dotenv import load_dotenv
from fastapi import FastAPI
from sqlmodel import create_engine, SQLModel, Session
from toml import load
load_dotenv()
with open("pyproject.toml") as f:
config = load(f)
# Dependency to get the FastAPI application instance
def get_app() -> FastAPI:
return FastAPI(
title=config["tool"]["poetry"]["name"],
description=config["tool"]["poetry"]["description"],
version=config["tool"]["poetry"]["version"],
openapi_tags=[
{
"name": "authors",
"description": "Operations with authors.",
},
{
"name": "books",
"description": "Operations with books.",
},
{
"name": "relations",
"description": "Operations with relations.",
},
{
"name": "misc",
"description": "Miscellaneous operations.",
}
]
)
USER = os.getenv("POSTGRES_USER")
PASSWORD = os.getenv("POSTGRES_PASSWORD")
DATABASE = os.getenv("POSTGRES_DB")
HOST = os.getenv("POSTGRES_SERVER")
if not USER or not PASSWORD or not DATABASE or not HOST:
raise ValueError("Missing environment variables")
POSTGRES_DATABASE_URL = f"postgresql://{USER}:{PASSWORD}@{HOST}:5432/{DATABASE}"
engine = create_engine(POSTGRES_DATABASE_URL, echo=True, future=True)
# Dependency to get a database session
def get_session():
with Session(engine) as session:
yield session

12
app/templates/index.html → library_service/templates/index.html
View File

@@ -3,7 +3,7 @@
<head>
<meta charset="UTF-8" />
<meta name="viewport" content="width=device-width, initial-scale=1.0" />
<title>{{ data.title }}</title>
<title>{{ app_info.title }}</title>
<style>
body {
font-family: Arial, sans-serif;
@@ -43,11 +43,11 @@
</style>
</head>
<body>
<h1>Welcome to {{ data.title }}!</h1>
<p>Description: {{ data.description }}</p>
<p>Version: {{ data.version }}</p>
<p>Current Time: {{ data.time }}</p>
<p>Status: {{ data.status }}</p>
<h1>Welcome to {{ app_info.title }}!</h1>
<p>Description: {{ app_info.description }}</p>
<p>Version: {{ app_info.version }}</p>
<p>Current Time: {{ server_time }}</p>
<p>Status: {{ status }}</p>
<ul>
<li><a href="/docs">Swagger UI</a></li>
<li><a href="/redoc">ReDoc</a></li>

1
migrations/README Executable file
View File

@@ -0,0 +1 @@
Generic single-database configuration with an async dbapi.

7
app/migrations/env.py → migrations/env.py Normal file → Executable file
View File

@@ -1,17 +1,16 @@
from logging.config import fileConfig
from decouple import config as conf
from alembic import context
from sqlalchemy import engine_from_config
from sqlalchemy import pool
from sqlmodel import SQLModel
DATABASE_URL = conf("DATABASE_URL")
from library_service.settings import POSTGRES_DATABASE_URL
# this is the Alembic Config object, which provides
# access to the values within the .ini file in use.
config = context.config
config.set_main_option("sqlalchemy.url", str(DATABASE_URL))
config.set_main_option("sqlalchemy.url", POSTGRES_DATABASE_URL)
# Interpret the config file for Python logging.
# This line sets up loggers basically.
@@ -20,7 +19,7 @@ if config.config_file_name is not None:
# add your model's MetaData object here
# for 'autogenerate' support
from app.models import Book, Author, AuthorBookLink
from library_service.models.db import *
target_metadata = SQLModel.metadata
# other values from the config, defined by the needs of env.py,

0
app/migrations/script.py.mako → migrations/script.py.mako Normal file → Executable file
View File

0
app/migrations/versions/d266fdc61e99_init.py → migrations/versions/d266fdc61e99_init.py
View File

1753
poetry.lock generated Normal file
View File

File diff suppressed because it is too large Load Diff

29
pyproject.toml Normal file
View File

@@ -0,0 +1,29 @@
[tool.poetry]
name = "library-service"
version = "0.1.1"
description = "This is a sample API for managing authors and books."
authors = ["wowlikon"]
readme = "README.md"
packages = [{ include = "library_service" }]
[tool.poetry.dependencies]
python = "^3.13"
fastapi = { extras = ["all"], version = "^0.115.12" }
psycopg2-binary = "^2.9.10"
alembic = "^1.16.1"
python-dotenv = "^0.21.0"
sqlmodel = "^0.0.24"
uvicorn = "^0.34.3"
jinja2 = "^3.1.6"
toml = "^0.10.2"
[tool.poetry.group.dev.dependencies]
black = "^25.1.0"
pytest = "^8.4.1"
[tool.poetry.requires-plugins]
poetry-plugin-export = ">=1.8"
[build-system]
requires = ["poetry-core"]
build-backend = "poetry.core.masonry.api"

9
requirements.txt
View File

@@ -1,9 +0,0 @@
fastapi
uvicorn[standard]
sqlmodel
psycopg2-binary
python-decouple
alembic
pytest
httpx
jinja2

116
src/alembic.ini
View File

@@ -1,116 +0,0 @@
# A generic, single database configuration.
[alembic]
# path to migration scripts
script_location = app/migrations
# template used to generate migration file names; The default value is %%(rev)s_%%(slug)s
# Uncomment the line below if you want the files to be prepended with date and time
# see https://alembic.sqlalchemy.org/en/latest/tutorial.html#editing-the-ini-file
# for all available tokens
# file_template = %%(year)d_%%(month).2d_%%(day).2d_%%(hour).2d%%(minute).2d-%%(rev)s_%%(slug)s
# sys.path path, will be prepended to sys.path if present.
# defaults to the current working directory.
prepend_sys_path = .
# timezone to use when rendering the date within the migration file
# as well as the filename.
# If specified, requires the python>=3.9 or backports.zoneinfo library.
# Any required deps can installed by adding `alembic[tz]` to the pip requirements
# string value is passed to ZoneInfo()
# leave blank for localtime
# timezone =
# max length of characters to apply to the
# "slug" field
# truncate_slug_length = 40
# set to 'true' to run the environment during
# the 'revision' command, regardless of autogenerate
# revision_environment = false
# set to 'true' to allow .pyc and .pyo files without
# a source .py file to be detected as revisions in the
# versions/ directory
# sourceless = false
# version location specification; This defaults
# to migrations/versions. When using multiple version
# directories, initial revisions must be specified with --version-path.
# The path separator used here should be the separator specified by "version_path_separator" below.
# version_locations = %(here)s/bar:%(here)s/bat:migrations/versions
# version path separator; As mentioned above, this is the character used to split
# version_locations. The default within new alembic.ini files is "os", which uses os.pathsep.
# If this key is omitted entirely, it falls back to the legacy behavior of splitting on spaces and/or commas.
# Valid values for version_path_separator are:
#
# version_path_separator = :
# version_path_separator = ;
# version_path_separator = space
version_path_separator = os # Use os.pathsep. Default configuration used for new projects.
# set to 'true' to search source files recursively
# in each "version_locations" directory
# new in Alembic version 1.10
# recursive_version_locations = false
# the output encoding used when revision files
# are written from script.py.mako
# output_encoding = utf-8
sqlalchemy.url = driver://user:pass@localhost/dbname
[post_write_hooks]
# post_write_hooks defines scripts or Python functions that are run
# on newly generated revision scripts. See the documentation for further
# detail and examples
# format using "black" - use the console_scripts runner, against the "black" entrypoint
# hooks = black
# black.type = console_scripts
# black.entrypoint = black
# black.options = -l 79 REVISION_SCRIPT_FILENAME
# lint with attempts to fix using "ruff" - use the exec runner, execute a binary
# hooks = ruff
# ruff.type = exec
# ruff.executable = %(here)s/.venv/bin/ruff
# ruff.options = --fix REVISION_SCRIPT_FILENAME
# Logging configuration
[loggers]
keys = root,sqlalchemy,alembic
[handlers]
keys = console
[formatters]
keys = generic
[logger_root]
level = WARN
handlers = console
qualname =
[logger_sqlalchemy]
level = WARN
handlers =
qualname = sqlalchemy.engine
[logger_alembic]
level = INFO
handlers =
qualname = alembic
[handler_console]
class = StreamHandler
args = (sys.stderr,)
level = NOTSET
formatter = generic
[formatter_generic]
format = %(levelname)-5.5s [%(name)s] %(message)s
datefmt = %H:%M:%S

0
src/app/__init__.py
View File

14
src/app/database.py
View File

@@ -1,14 +0,0 @@
from sqlmodel import create_engine, SQLModel, Session
from decouple import config
# Get database configuration
DATABASE_URL = config('DATABASE_URL', cast=str, default='sqlite:///./bookapi.db')
# Create database engine
engine = create_engine(str(DATABASE_URL), echo=True)
# SQLModel.metadata.create_all(engine)
# Get database session
def get_session():
with Session(engine) as session:
yield session

45
src/app/main.py
View File

@@ -1,45 +0,0 @@
from alembic import command
from alembic.config import Config
from fastapi import FastAPI
from sqlmodel import SQLModel
from .database import engine
from .routers import api_router
from .routers.misc import get_info
alembic_cfg = Config("alembic.ini")
app = FastAPI(
title="LibraryAPI",
description="This is a sample API for managing authors and books.",
version="1.0.1",
openapi_tags=[
{
"name": "authors",
"description": "Operations with authors.",
},
{
"name": "books",
"description": "Operations with books.",
},
{
"name": "relations",
"description": "Operations with relations.",
},
{
"name": "misc",
"description": "Miscellaneous operations.",
}
]
)
# Initialize the database
@app.on_event("startup")
def on_startup():
# Apply database migrations
with engine.begin() as connection:
alembic_cfg.attributes['connection'] = connection
command.upgrade(alembic_cfg, "head")
# Include routers
app.include_router(api_router)

1
src/app/migrations/README
View File

@@ -1 +0,0 @@
Generic single-database configuration.

81
src/app/migrations/env.py
View File

@@ -1,81 +0,0 @@
from logging.config import fileConfig
from decouple import config as conf
from alembic import context
from sqlalchemy import engine_from_config
from sqlalchemy import pool
from sqlmodel import SQLModel
DATABASE_URL = conf("DATABASE_URL")
# this is the Alembic Config object, which provides
# access to the values within the .ini file in use.
config = context.config
config.set_main_option("sqlalchemy.url", str(DATABASE_URL))
# Interpret the config file for Python logging.
# This line sets up loggers basically.
if config.config_file_name is not None:
fileConfig(config.config_file_name)
# add your model's MetaData object here
# for 'autogenerate' support
from app.models import Book, Author, AuthorBookLink
target_metadata = SQLModel.metadata
# other values from the config, defined by the needs of env.py,
# can be acquired:
# my_important_option = config.get_main_option("my_important_option")
# ... etc.
def run_migrations_offline() -> None:
"""Run migrations in 'offline' mode.
This configures the context with just a URL
and not an Engine, though an Engine is acceptable
here as well. By skipping the Engine creation
we don't even need a DBAPI to be available.
Calls to context.execute() here emit the given string to the
script output.
"""
url = config.get_main_option("sqlalchemy.url")
context.configure(
url=url,
target_metadata=target_metadata,
literal_binds=True,
dialect_opts={"paramstyle": "named"},
)
with context.begin_transaction():
context.run_migrations()
def run_migrations_online() -> None:
"""Run migrations in 'online' mode.
In this scenario we need to create an Engine
and associate a connection with the context.
"""
connectable = engine_from_config(
config.get_section(config.config_ini_section, {}),
prefix="sqlalchemy.",
poolclass=pool.NullPool,
)
with connectable.connect() as connection:
context.configure(
connection=connection, target_metadata=target_metadata
)
with context.begin_transaction():
context.run_migrations()
if context.is_offline_mode():
run_migrations_offline()
else:
run_migrations_online()

27
src/app/migrations/script.py.mako
View File

@@ -1,27 +0,0 @@
"""${message}
Revision ID: ${up_revision}
Revises: ${down_revision | comma,n}
Create Date: ${create_date}
"""
from typing import Sequence, Union
from alembic import op
import sqlalchemy as sa
import sqlmodel
${imports if imports else ""}
# revision identifiers, used by Alembic.
revision: str = ${repr(up_revision)}
down_revision: Union[str, None] = ${repr(down_revision)}
branch_labels: Union[str, Sequence[str], None] = ${repr(branch_labels)}
depends_on: Union[str, Sequence[str], None] = ${repr(depends_on)}
def upgrade() -> None:
${upgrades if upgrades else "pass"}
def downgrade() -> None:
${downgrades if downgrades else "pass"}

54
src/app/migrations/versions/d266fdc61e99_init.py
View File

@@ -1,54 +0,0 @@
"""init
Revision ID: d266fdc61e99
Revises:
Create Date: 2025-05-27 18:04:22.279035
"""
from typing import Sequence, Union
from alembic import op
import sqlalchemy as sa
import sqlmodel
# revision identifiers, used by Alembic.
revision: str = 'd266fdc61e99'
down_revision: Union[str, None] = None
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
def upgrade() -> None:
# ### commands auto generated by Alembic - please adjust! ###
op.create_table('author',
sa.Column('id', sa.Integer(), nullable=False),
sa.Column('name', sqlmodel.sql.sqltypes.AutoString(), nullable=False),
sa.PrimaryKeyConstraint('id')
)
op.create_index(op.f('ix_author_id'), 'author', ['id'], unique=False)
op.create_table('book',
sa.Column('id', sa.Integer(), nullable=False),
sa.Column('title', sqlmodel.sql.sqltypes.AutoString(), nullable=False),
sa.Column('description', sqlmodel.sql.sqltypes.AutoString(), nullable=False),
sa.PrimaryKeyConstraint('id')
)
op.create_index(op.f('ix_book_id'), 'book', ['id'], unique=False)
op.create_table('authorbooklink',
sa.Column('author_id', sa.Integer(), nullable=False),
sa.Column('book_id', sa.Integer(), nullable=False),
sa.ForeignKeyConstraint(['author_id'], ['author.id'], ),
sa.ForeignKeyConstraint(['book_id'], ['book.id'], ),
sa.PrimaryKeyConstraint('author_id', 'book_id')
)
# ### end Alembic commands ###
def downgrade() -> None:
# ### commands auto generated by Alembic - please adjust! ###
op.drop_table('authorbooklink')
op.drop_index(op.f('ix_book_id'), table_name='book')
op.drop_table('book')
op.drop_index(op.f('ix_author_id'), table_name='author')
op.drop_table('author')
# ### end Alembic commands ###

5
src/app/models/__init__.py
View File

@@ -1,5 +0,0 @@
from .author import Author, AuthorBase
from .book import Book, BookBase
from .links import AuthorBookLink
__all__ = ['Author', 'AuthorBase', 'Book', 'BookBase', 'AuthorBookLink']

20
src/app/models/author.py
View File

@@ -1,20 +0,0 @@
from typing import List, Optional
from sqlmodel import SQLModel, Field, Relationship
from .links import AuthorBookLink
from .book import Book
# Author DTO model
class AuthorBase(SQLModel):
name: str
class Config: # pyright: ignore
json_schema_extra = {
"example": {
"name": "author_name",
}
}
# Author DB model
class Author(AuthorBase, table=True):
id: Optional[int] = Field(default=None, primary_key=True, index=True)
books: List["Book"] = Relationship(back_populates="authors", link_model=AuthorBookLink)

22
src/app/models/book.py
View File

@@ -1,22 +0,0 @@
from typing import List, Optional
from sqlmodel import SQLModel, Field, Relationship
from .links import AuthorBookLink
from .author import Author
# Book DTO model
class BookBase(SQLModel):
title: str
description: str
class Config: # pyright: ignore
json_schema_extra = {
"example": {
"title": "book_title",
"description": "book_description",
}
}
# Book DB model
class Book(BookBase, table=True):
id: Optional[int] = Field(default=None, primary_key=True, index=True)
authors: List["Author"] = Relationship(back_populates="books", link_model=AuthorBookLink)

6
src/app/models/links.py
View File

@@ -1,6 +0,0 @@
from sqlmodel import SQLModel, Field
# Relationship model
class AuthorBookLink(SQLModel, table=True):
author_id: int | None = Field(default=None, foreign_key="author.id", primary_key=True)
book_id: int | None = Field(default=None, foreign_key="book.id", primary_key=True)

45
src/app/routers/authors.py
View File

@@ -1,45 +0,0 @@
from fastapi import APIRouter, Depends, HTTPException
from sqlmodel import Session, select
from typing import List
from ..database import get_session
from ..models import Author, AuthorBase
router = APIRouter(prefix="/authors", tags=["authors"])
# Create an author
@router.post("/", response_model=Author)
def create_author(author: AuthorBase, session: Session = Depends(get_session)):
db_author = Author(name=author.name)
session.add(db_author)
session.commit()
session.refresh(db_author)
return db_author
# Read authors
@router.get("/", response_model=List[Author])
def read_authors(session: Session = Depends(get_session)):
authors = session.exec(select(Author)).all()
return authors
# Update an author
@router.put("/{author_id}", response_model=Author)
def update_author(author_id: int, author: AuthorBase, session: Session = Depends(get_session)):
db_author = session.get(Author, author_id)
if not db_author:
raise HTTPException(status_code=404, detail="Author not found")
db_author.name = author.name
session.commit()
session.refresh(db_author)
return db_author
# Delete an author
@router.delete("/{author_id}", response_model=AuthorBase)
def delete_author(author_id: int, session: Session = Depends(get_session)):
db_author = session.get(Author, author_id)
if not db_author:
raise HTTPException(status_code=404, detail="Author not found")
session.delete(db_author)
author = AuthorBase(name=db_author.name)
session.commit()
return author

62
src/app/routers/books.py
View File

@@ -1,62 +0,0 @@
from fastapi import APIRouter, Depends, HTTPException
from sqlmodel import Session, select
from typing import List
from ..database import get_session
from ..models import Book, BookBase, Author, AuthorBookLink
router = APIRouter(prefix="/books", tags=["books"])
# Create a book
@router.post("/", response_model=Book)
def create_book(book: BookBase, session: Session = Depends(get_session)):
db_book = Book(title=book.title, description=book.description)
session.add(db_book)
session.commit()
session.refresh(db_book)
return db_book
# Read books
@router.get("/", response_model=List[Book])
def read_books(session: Session = Depends(get_session)):
books = session.exec(select(Book)).all()
return books
# Update a book
@router.put("/{book_id}", response_model=Book)
def update_book(book_id: int, book: BookBase, session: Session = Depends(get_session)):
db_book = session.get(Book, book_id)
if not db_book:
raise HTTPException(status_code=404, detail="Book not found")
db_book.title = book.title
db_book.description = book.description
session.commit()
session.refresh(db_book)
return db_book
# Delete a book
@router.delete("/{book_id}", response_model=BookBase)
def delete_book(book_id: int, session: Session = Depends(get_session)):
db_book = session.get(Book, book_id)
if not db_book:
raise HTTPException(status_code=404, detail="Book not found")
book = Book(title=db_book.title, description=db_book.description)
session.delete(db_book)
session.commit()
return book
# Get all authors for a book
@router.get("/{book_id}/authors/", response_model=List[Author], tags=["relations"])
def get_authors_for_book(book_id: int, session: Session = Depends(get_session)):
book = session.get(Book, book_id)
if not book:
raise HTTPException(status_code=404, detail="Book not found")
authors = session.exec(
select(Author)
.join(AuthorBookLink)
.where(AuthorBookLink.book_id == book_id)
).all()
return authors

56
src/app/templates/index.html
View File

@@ -1,56 +0,0 @@
<!doctype html>
<html lang="en">
<head>
<meta charset="UTF-8" />
<meta name="viewport" content="width=device-width, initial-scale=1.0" />
<title>{{ data.title }}</title>
<style>
body {
font-family: Arial, sans-serif;
max-width: 800px;
margin: 0 auto;
padding: 20px;
line-height: 1.6;
color: #333;
}
h1 {
color: #2c3e50;
border-bottom: 2px solid #3498db;
padding-bottom: 10px;
}
ul {
list-style: none;
padding: 0;
}
li {
margin: 15px 0;
}
a {
display: inline-block;
padding: 8px 15px;
background-color: #3498db;
color: white;
text-decoration: none;
border-radius: 4px;
transition: background-color 0.3s;
}
a:hover {
background-color: #2980b9;
}
p {
margin: 5px 0;
}
</style>
</head>
<body>
<h1>Welcome to {{ data.title }}!</h1>
<p>Description: {{ data.description }}</p>
<p>Version: {{ data.version }}</p>
<p>Current Time: {{ data.time }}</p>
<p>Status: {{ data.status }}</p>
<ul>
<li><a href="/docs">Swagger UI</a></li>
<li><a href="/redoc">ReDoc</a></li>
</ul>
</body>
</html>

12
tests/test_authors.py Normal file
View File

@@ -0,0 +1,12 @@
import pytest
from alembic import command
from alembic.config import Config
from fastapi.testclient import TestClient
from sqlmodel import select, delete, Session
from library_service.main import app
from tests.test_misc import setup_database
client = TestClient(app)
#TODO: add tests for author endpoints

58
tests/test_books.py Normal file
View File

@@ -0,0 +1,58 @@
import pytest
from alembic import command
from alembic.config import Config
from fastapi.testclient import TestClient
from sqlmodel import select, delete, Session
from library_service.main import app
from tests.test_misc import setup_database
client = TestClient(app)
#TODO: assert descriptions
#TODO: add comments
#TODO: update tests
def test_create_book(setup_database):
response = client.post("/books", json={"title": "Test Book", "description": "Test Description"})
print(response.json())
assert response.status_code == 200
assert response.json() == {"id": 1, "title": "Test Book", "description": "Test Description"}
def test_get_existing_book(setup_database):
response = client.get("/books/1")
print(response.json())
assert response.status_code == 200
assert response.json() == {"id": 1, "title": "Test Book", "description": "Test Description", 'authors': []}
def test_get_not_existing_book(setup_database):
response = client.get("/books/2")
print(response.json())
assert response.status_code == 404
assert response.json() == {"detail": "Book not found"}
def test_update_book(setup_database):
response = client.get("/books/1")
assert response.status_code == 200
response = client.put("/books/1", json={"title": "Updated Book", "description": "Updated Description"})
assert response.status_code == 200
assert response.json() == {"id": 1, "title": "Updated Book", "description": "Updated Description"}
def test_update_not_existing_book(setup_database):
response = client.put("/books/2", json={"title": "Updated Book", "description": "Updated Description"})
assert response.status_code == 404
assert response.json() == {"detail": "Book not found"}
def test_delete_book(setup_database):
response = client.get("/books/1")
assert response.status_code == 200
response = client.delete("/books/1")
assert response.status_code == 200
assert response.json() == {"id": 1, "title": "Updated Book", "description": "Updated Description"}
def test_not_existing_delete_book(setup_database):
response = client.delete("/books/2")
assert response.status_code == 404
assert response.json() == {"detail": "Book not found"}
#TODO: add tests for other books endpoints

70
tests/test_main.py
View File

@@ -1,70 +0,0 @@
import pytest # pyright: ignore
from fastapi.testclient import TestClient
from app.main import app
@pytest.fixture()
def client():
with TestClient(app) as test_client:
yield test_client
# Тесты для авторов
def test_create_author(client):
response = client.post("/authors/", json={"name": "Author Name"})
assert response.status_code == 200
assert response.json()["name"] == "Author Name"
def test_read_authors(client):
response = client.get("/authors/")
assert response.status_code == 200
assert isinstance(response.json(), list) # Проверяем, что ответ - это список
def test_update_author(client):
# Сначала создаем автора, чтобы его обновить
create_response = client.post("/authors/", json={"name": "Author Name"})
author_id = create_response.json()["id"]
response = client.put(f"/authors/{author_id}", json={"name": "Updated Author Name"})
assert response.status_code == 200
assert response.json()["name"] == "Updated Author Name"
def test_delete_author(client):
# Сначала создаем автора, чтобы его удалить
create_response = client.post("/authors/", json={"name": "Author Name"})
author_id = create_response.json()["id"]
author_name = create_response.json()["name"]
response = client.delete(f"/authors/{author_id}")
assert response.status_code == 200
assert response.json()["name"] == author_name
# Тесты для книг
def test_create_book(client):
response = client.post("/books/", json={"title": "Book Title", "description": "Book Description"})
assert response.status_code == 200
assert response.json()["title"] == "Book Title"
def test_read_books(client):
response = client.get("/books/")
assert response.status_code == 200
assert isinstance(response.json(), list) # Проверяем, что ответ - это список
def test_update_book(client):
# Сначала создаем книгу, чтобы ее обновить
create_response = client.post("/books/", json={"title": "Book Title", "description": "Book Description"})
book_id = create_response.json()["id"]
response = client.put(f"/books/{book_id}", json={"title": "Updated Book Title", "description": "Updated Description"})
assert response.status_code == 200
assert response.json()["title"] == "Updated Book Title"
def test_delete_book(client):
# Сначала создаем книгу, чтобы ее удалить
create_response = client.post("/books/", json={"title": "Book Title", "description": "Book Description"})
book_id = create_response.json()["id"]
book_title = create_response.json()["title"]
book_description = create_response.json()["description"]
response = client.delete(f"/books/{book_id}")
assert response.status_code == 200
assert response.json()["title"] == book_title
assert response.json()["description"] == book_description

75
tests/test_misc.py Normal file
View File

@@ -0,0 +1,75 @@
import pytest
from alembic import command
from alembic.config import Config
from fastapi.testclient import TestClient
from sqlmodel import select, delete, Session
from library_service.main import app, engine
from library_service.models.db import Author, Book, AuthorBookLink
client = TestClient(app)
@pytest.fixture(scope="module")
def setup_database():
# Save original data backup
with Session(engine) as session:
original_authors = session.exec(select(Author)).all()
original_books = session.exec(select(Book)).all()
original_links = session.exec(select(AuthorBookLink)).all()
# Reset database
alembic_cfg = Config("alembic.ini")
with engine.begin() as connection:
alembic_cfg.attributes['connection'] = connection
command.downgrade(alembic_cfg, 'base')
command.upgrade(alembic_cfg, 'head')
# Check database state after reset
with Session(engine) as session:
assert len(session.exec(select(Author)).all()) == 0
assert len(session.exec(select(Book)).all()) == 0
assert len(session.exec(select(AuthorBookLink)).all()) == 0
yield # Here pytest will start testing
# Restore original data from backup
with Session(engine) as session:
for author in original_authors:
session.add(author)
for book in original_books:
session.add(book)
for link in original_links:
session.add(link)
session.commit()
# Test the main page of the application
def test_main_page():
response = client.get("/") # Send GET request to the main page
try:
content = response.content.decode('utf-8') # Decode response content
# Find indices of key elements in the content
title_idx = content.index("Welcome to ")
description_idx = content.index("Description: ")
version_idx = content.index("Version: ")
time_idx = content.index("Current Time: ")
status_idx = content.index("Status: ")
assert response.status_code == 200, "Invalid response status"
assert content.startswith('<!doctype html>'), "Not HTML"
assert content.endswith('</html>'), "HTML tag not closed"
assert content[title_idx+1] != '<', "Title not provided"
assert content[description_idx+1] != '<', "Description not provided"
assert content[version_idx+1] != '<', "Version not provided"
assert content[time_idx+1] != '<', "Time not provided"
assert content[status_idx+1] != '<', "Status not provided"
except Exception as e:
print(f"Error: {e}") # Print error if an exception occurs
assert False, "Unexpected error" # Force test failure on unexpected error
# Test application info endpoint
def test_app_info_test():
response = client.get("/api/info") # Send GET request to the info endpoint
assert response.status_code == 200, "Invalid response status"
assert response.json()["status"] == "ok", "Status not ok"
assert response.json()["app_info"]["title"] != "", "Title not provided"
assert response.json()["app_info"]["description"] != "", "Description not provided"
assert response.json()["app_info"]["version"] != "", "Version not provided"
# Check time difference
assert 0 < (datetime.now() - datetime.fromisoformat(response.json()["server_time"])).total_seconds(), "Negative time difference"
assert (datetime.now() - datetime.fromisoformat(response.json()["server_time"])).total_seconds() < 1, "Time difference too large"

12
tests/test_relationships.py Normal file
View File

@@ -0,0 +1,12 @@
import pytest
from alembic import command
from alembic.config import Config
from fastapi.testclient import TestClient
from sqlmodel import select, delete, Session
from library_service.main import app
from tests.test_misc import setup_database
client = TestClient(app)
#TODO: add tests for relationships endpoints