SQLAlchemy, Alembic, Pydantic

SQLAlchemy

pip install sqlalchemy
pip install sqlalchemy-stubs
#stub - підказка синтаксису

Створення DB engine (ШАБЛОННИЙ КОД! з офіційного сайту)

engine.py :

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.declarative import declarative_base

SQLALCHEMY_DATABASE_URL = 'sqlite:///./cheese-catalog.db'

engine = create_engine(
    SQLALCHEMY_DATABASE_URL,
    connect_args={"check_same_thread": False}
)

SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

Base = declarative_base()

models.py :

from enum import StrEnum, auto
from sqlalchemy import Column, Integer, String, Enum, ForeignKey
from sqlalchemy.orm import relationship

from db.engine import Base


class PackagingType(StrEnum):
    IN_PACKAGE = auto()
    WEIGHT = auto()


class DBCheeseType(Base):
    __tablename__ = 'cheese_type'
    id = Column(Integer, primary_key=True, index=True)
    name = Column(String(255), unique=True, nullable=False)
    description = Column(String(511), unique=True, nullable=False)


class DBCheese(Base):
    __tablename__ = 'cheese'
    id = Column(Integer, primary_key=True, index=True)
    title = Column(String(255), unique=True, nullable=False)
    price = Column(Integer)
    package_type = Column(Enum(PackagingType), nullable=False)
    cheese_type_id = Column(Integer, ForeignKey('cheese_type.id'))
    cheese_type = relationship(DBCheeseType)

Alembic

Встановлення інструменту міграцій (Alembic) у терміналі:

pip install alembic

alembic init alembic

Перезаписати в alembic.ini:

...
sqlalchemy.url = sqlite:///./cheese-catalog.db
...

Перезаписати в env.py:

...
from db.models import Base
...
...
target_metadata = Base.metadata
...

В терміналі:

alembic revision --autogenerate -m "Initial migration"

alembic upgrade head

Pydantic:

створити файл schemas.py і в ньому:

from pydantic import BaseModel


class CheeseTypeBase(BaseModel):
    name: str
    description: str

class CheeseTypeCreate(CheeseTypeBase):
    pass

class CheeseType(CheeseTypeBase):
    id: int

    class Config:
        orm_mode = True

FastAPI CRUD

CRUD (Create, Read, Update, Delete). Описує операції, які ми можемо виконувати в нашому сервісі. Створити файл crud.py з вмістом:

from sqlalchemy.orm import Session

from models import DBAuthor
from schemas import AuthorCreate

#Створюється для перевірки чи вже існує автор в базі
def get_author_by_name(db: Session, name: str):
    return db.query(DBAuthor).filter(DBAuthor.name == name).first()


def get_all_authors(db: Session):
    return db.query(DBAuthor).all()


def create_author(db: Session, author: AuthorCreate):
    db_author = DBAuthor(
        name=author.name,
        bio=author.bio
    )
    db.add(db_author)
    db.commit()
    db.refresh(db_author)
    return db_author

Нарешті створюємо FastAPI ендпоінти (views):

в main.py

from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy.orm import Session

import schemas
from crud import get_all_authors, create_author, get_author_by_name
from database import SessionLocal


#Сінглтон сесії бази даних
app = FastAPI()
def get_db() -> Session:
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

@app.get("/")
async def root() -> dict:
    return {"message": "Hello World"}


#Перший параметр - ендпоінт, другий - те що ми очікуємо отримати після виконання запиту
@app.get("/authors", response_model=list[schemas.Author])
async def get_authors(db: Session = Depends(get_db)):
    return get_all_authors(db)

@app.post("/authors", response_model=schemas.Author)
async def create_new_author(author: schemas.AuthorCreate, db: Session = Depends(get_db)):
    db_author = get_author_by_name(db, author.name)
    if db_author:
        raise HTTPException(status_code=400, detail="Author already registered")
    return create_author(db, author)