Async Database - szybki dostęp do danych

Witaj, @name! Darwin z asynchronicznymi bazami danych w FastAPI! 🗄️⚡

Do tej pory używaliśmy in-memory dict. Teraz podłączymy prawdziwą bazę danych PostgreSQL z async SQLAlchemy!

Analogia Safari: Async database to jak system real-time tracking zwierząt - tysiące obserwacji zapisywanych jednocześnie bez blokowania! 📡🦁

Instalacja

1pip install sqlalchemy[asyncio] asyncpg

• SQLAlchemy - ORM
• asyncpg - async PostgreSQL driver

Konfiguracja Async SQLAlchemy

1# database.py
2from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker
3from sqlalchemy.orm import DeclarativeBase
4
5DATABASE_URL = "postgresql+asyncpg://user:password@localhost/safari_db"
6
7# Async engine
8engine = create_async_engine(DATABASE_URL, echo=True)
9
10# Session factory
11AsyncSessionLocal = async_sessionmaker(
12    engine,
13    class_=AsyncSession,
14    expire_on_commit=False
15)
16
17# Base class
18class Base(DeclarativeBase):
19    pass
20
21# Dependency
22async def get_db():
23    async with AsyncSessionLocal() as session:
24        yield session

Model SQLAlchemy

1# models.py
2from sqlalchemy import Integer, String, Boolean
3from sqlalchemy.orm import Mapped, mapped_column
4from database import Base
5
6class SpeciesModel(Base):
7    __tablename__ = "species"
8
9    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
10    name: Mapped[str] = mapped_column(String(100), nullable=False)
11    scientific_name: Mapped[str] = mapped_column(String(150), nullable=False)
12    population: Mapped[int] = mapped_column(Integer, default=0)
13    endangered: Mapped[bool] = mapped_column(Boolean, default=False)

CRUD Operations - Async

1# crud.py
2from sqlalchemy.ext.asyncio import AsyncSession
3from sqlalchemy import select
4from models import SpeciesModel
5from schemas import SpeciesCreate
6
7async def get_species(db: AsyncSession, species_id: int):
8    result = await db.execute(select(SpeciesModel).where(SpeciesModel.id == species_id))
9    return result.scalar_one_or_none()
10
11async def get_all_species(db: AsyncSession, skip: int = 0, limit: int = 10):
12    result = await db.execute(select(SpeciesModel).offset(skip).limit(limit))
13    return result.scalars().all()
14
15async def create_species(db: AsyncSession, species: SpeciesCreate):
16    db_species = SpeciesModel(**species.dict())
17    db.add(db_species)
18    await db.commit()
19    await db.refresh(db_species)
20    return db_species
21
22async def delete_species(db: AsyncSession, species_id: int):
23    species = await get_species(db, species_id)
24    if species:
25        await db.delete(species)
26        await db.commit()
27    return species

FastAPI Endpoints z Async DB

1# main.py
2from fastapi import FastAPI, Depends, HTTPException
3from sqlalchemy.ext.asyncio import AsyncSession
4from database import get_db
5import crud
6import schemas
7
8app = FastAPI()
9
10@app.get("/species", response_model=list[schemas.Species])
11async def list_species(skip: int = 0, limit: int = 10, db: AsyncSession = Depends(get_db)):
12    species = await crud.get_all_species(db, skip=skip, limit=limit)
13    return species
14
15@app.get("/species/{species_id}", response_model=schemas.Species)
16async def get_species(species_id: int, db: AsyncSession = Depends(get_db)):
17    species = await crud.get_species(db, species_id)
18    if not species:
19        raise HTTPException(status_code=404, detail="Species not found")
20    return species
21
22@app.post("/species", response_model=schemas.Species)
23async def create_species(species: schemas.SpeciesCreate, db: AsyncSession = Depends(get_db)):
24    return await crud.create_species(db, species)

Zalet y async database:

• ⚡ Tysiące requestów jednocześnie
• 🚀 Nie blokuje event loop
• 📈 Wysoka wydajność

Następna lekcja: JWT Authentication! 🔐