fastapi_with_db.py

from sqlalchemy import select
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column

from fastapi import FastAPI, Depends

from pydantic import BaseModel
from typing import Annotated

app = FastAPI()

engine = create_async_engine('sqlite+aiosqlite:///books.db')

new_session = async_sessionmaker(bind=engine, expire_on_commit=False)


async def get_session():
    async with new_session() as session:
        yield session


SessionDep = Annotated[AsyncSession, Depends(get_session)]


class Base(DeclarativeBase):
    pass


class BooksModel(Base):
    __tablename__ = 'books'

    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str]
    author: Mapped[str]


@app.post("/setup_database")
async def setup_database():
    async with engine.begin() as connection:
        await connection.run_sync(BaseModel.metadata.drop_all)
        await connection.run_sync(BaseModel.metadata.create_all)

    return {"ok": True}


class BookAddSchema(BaseModel):
    title: str
    author: str


class BookSchema(BookAddSchema):
    id: int


@app.post("/add_books")
async def add_books(data: BookAddSchema, session: SessionDep):
    new_book = BooksModel(title=data.title, author=data.author)
    session.add(new_book)
    await session.commit()
    return {"ok": True}


@app.get("/get_books")
async def get_books(session: SessionDep):
    query = select(BooksModel)
    result = await session.execute(query)
    return result.scalars().all()