Solution

crud.py

@@ -0,0 +1,65 @@
+from fastapi import HTTPException
+from sqlalchemy import select
+from sqlalchemy.orm import Session
+
+from db.models import Author, Book
+from db.schemas import AuthorCreate, BookCreate
+
+
+def get_all_authors_by_skip_and_limit(db: Session, skip: int, limit: int):
+    stmt = select(Author)
+
+    if skip is not None:
+        stmt = stmt.offset(skip)
+
+    if limit is not None:
+        stmt = stmt.limit(limit)
+
+    return db.execute(stmt).scalars().all()
+
+def create_new_author(db: Session, author: AuthorCreate):
+    db_author = Author(name=author.name, bio=author.bio)
+
+    db.add(db_author)
+    db.commit()
+    db.refresh(db_author)
+
+    return db_author
+
+def get_author_by_id(db: Session, item_id: int):
+    return db.scalar(select(Author).where(Author.id == item_id))
+
+def create_book_for_author(db: Session, book: BookCreate):
+    author = db.scalar(select(Author).where(Author.id == book.author_id))
+    if author is None:
+        raise HTTPException(status_code=404, detail="Author with such id does not exist")
+
+    db_book = Book(
+        title=book.title,
+        summary=book.summary,
+        publication_date=book.publication_date,
+        author_id=book.author_id
+    )
+
+    db.add(db_book)
+    db.commit()
+
+    db.refresh(db_book)
+
+    return db_book
+
+
+
+def get_books_by_filtering(db: Session, skip: int | None, limit: int | None, author_id: int | None):
+    stmt = select(Book)
+
+    if author_id is not None:
+        stmt = stmt.where(Book.author_id == author_id)
+
+    if skip is not None:
+        stmt = stmt.offset(skip)
+
+    if limit is not None:
+        stmt = stmt.limit(limit)
+
+    return db.execute(stmt).scalars().all()

db/database.py

@@ -0,0 +1,14 @@
+from sqlalchemy import create_engine
+from sqlalchemy.orm import sessionmaker
+from sqlalchemy.ext.declarative import declarative_base
+
+
+sqlite_file_name = "database.db"
+sqlite_url = f"sqlite:///{sqlite_file_name}"
+
+connect_args = {"check_same_thread": False}
+engine = create_engine(sqlite_url, connect_args=connect_args)
+
+SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
+
+Base = declarative_base()

db/models.py

@@ -0,0 +1,26 @@
+from sqlalchemy import DateTime
+from sqlalchemy import ForeignKey, String, Text
+from sqlalchemy.orm import Mapped, mapped_column, relationship
+from datetime import datetime
+
+from db.database import Base
+
+
+class Author(Base):
+    __tablename__ = "authors"
+
+    id: Mapped[int] = mapped_column(primary_key=True, index=True)
+    name: Mapped[str] = mapped_column(String(255), unique=True, nullable=False, index=True)
+    bio: Mapped[str] = mapped_column(Text, nullable=False)
+    books: Mapped[list["Book"]] = relationship(back_populates="author", cascade="all, delete-orphan")
+
+
+class Book(Base):
+    __tablename__ = "books"
+
+    id: Mapped[int] = mapped_column(primary_key=True, index=True)
+    title: Mapped[str] = mapped_column(String(255), nullable=False, index=True)
+    summary: Mapped[str] = mapped_column(Text, nullable=False)
+    publication_date: Mapped[datetime] = mapped_column(DateTime, nullable=False)
+    author_id: Mapped[int] = mapped_column(ForeignKey("authors.id"), nullable=False, index=True)
+    author: Mapped["Author"] = relationship(back_populates="books")

db/schemas.py

@@ -0,0 +1,35 @@
+from pydantic import BaseModel, ConfigDict, Field
+from datetime import datetime
+
+
+class AuthorBase(BaseModel):
+    name: str
+    bio: str
+
+
+class BookBase(BaseModel):
+    title: str
+    summary: str
+    publication_date: datetime
+
+
+class AuthorCreate(AuthorBase):
+    pass
+
+
+class BookCreate(BookBase):
+    author_id: int
+
+
+class Book(BookBase):
+    id: int
+    author_id: int
+
+    model_config = ConfigDict(from_attributes=True)
+
+
+class Author(AuthorBase):
+    id: int
+    books: list[Book] = Field(default_factory=list)
+
+    model_config = ConfigDict(from_attributes=True)

main.py

@@ -0,0 +1,55 @@
+from typing import Any, Generator
+
+from sqlalchemy.orm import Session
+from fastapi import FastAPI, Depends
+
+from crud import create_new_author, get_all_authors_by_skip_and_limit, get_author_by_id, create_book_for_author, \
+    get_books_by_filtering
+from db import schemas
+from db.database import SessionLocal, Base, engine
+
+app = FastAPI()
+
+
+@app.on_event("startup")
+def startup():
+    Base.metadata.create_all(bind=engine)
+
+
+def get_db() -> Generator[Any, Any, None]:
+    db = SessionLocal()
+
+    try:
+        yield db
+    finally:
+        db.close()
+
+
+@app.get("/")
+def read_root():
+    return {"Hello": "World"}
+
+
+@app.post("/authors", response_model=schemas.Author)
+def create_author(author: schemas.AuthorCreate, db: Session = Depends(get_db)):
+    return create_new_author(db, author)
+
+
+@app.get("/authors", response_model=list[schemas.Author])
+def read_authors(skip: int | None, limit: int | None, db: Session = Depends(get_db)):
+    return get_all_authors_by_skip_and_limit(db, skip, limit)
+
+
+@app.get("/authors/{item_id}", response_model=schemas.Author)
+def retrieve_author(item_id: int, db: Session = Depends(get_db)):
+    return get_author_by_id(db, item_id)
+
+
+@app.post("/books", response_model=schemas.Book)
+def create_book(book: schemas.BookCreate, db: Session = Depends(get_db)):
+    return create_book_for_author(db, book)
+
+
+@app.get("/books", response_model=list[schemas.Book])
+def get_books(skip: int | None = None, limit: int | None = None, author_id: int = 0, db: Session = Depends(get_db)):
+    return get_books_by_filtering(db, skip, limit, author_id)