SQLAlchemy, Alembic, Pydantic

SQLAlchemy

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

Creating DB engine (BOILERPLATE! on official website)

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

Install migration tool (Alembic) in terminal:

pip install alembic

alembic init alembic

Rewrite in alembic.ini:

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

Rewrite in env.py:

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

In terminal:

alembic revision --autogenerate -m "Initial migration"

alembic upgrade head

Pydantic:

create file schemas.py and in it:

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). Describe operation which we can do on our service. Create file crud.py with content:

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

Finally create FastAPI endpoints (veiews):

in 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)