Fast API #

Structure Clean Architecture #

finance-api/
├── domain/
│   ├── base.py                
│   ├── category.py            # Model : e.g., Transaction, Category, etc.
│   └── __init__.py
├── application/
│   ├── use_cases/
│   │   ├── create_transaction.py
│   │   ├── forecast_balances.py
│   │   └── __init__.py
│   ├── ports/
│   │   ├── transaction_repository.py
│   │   └── __init__.py
│   └── __init__.py
├── infrastructure/
│   ├── db/
│   │   ├── repositories.py
│   │   └── __init__.py
│   ├── config/
│   │   └── settings.py
│   └── __init__.py
├── presentation/
│   ├── routers/
│   │   ├── transactions.py
│   │   └── __init__.py
│   └── __init__.py
├── main.py
└── tests/
    └── ...

Steps #

Routes #

Create static routes first in presentation/routers

from fastapi import APIRouter

router = APIRouter()

@router.get("/categories")
def list_categories():
    return {"message": "Categories list"}

include routes in main.py

from fastapi import FastAPI
from app.presentation.routers.categories import router as categories_router

app = FastAPI()

# Include routers
app.include_router(categories_router)

To read : https://fastapi.tiangolo.com/async/#in-a-hurry

DB interaction #

Create entities

#domain/base.py
from sqlalchemy.orm import declarative_base

Base = declarative_base()

#domain/category.py
from sqlalchemy import Boolean, Column, DateTime, Integer, String, func
from .base import Base

class Category(Base):
    __tablename__ = "categories"

    id = Column(Integer, primary_key=True)
    name = Column(String, nullable=False)
    archived = Column(Boolean, default=False, nullable=False)
    created_at = Column(DateTime(timezone=True), server_default=func.now(), nullable=False)
    updated_at = Column(DateTime(timezone=True), server_default=func.now(), onupdate=func.now(), nullable=False)

Create db connection

#infrastructure/db/session.py
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, Session

DATABASE_URL = "postgresql://username:password@db:5432/dbname"

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

def get_db() -> Session:
    try:
        db = SessionLocal()
        yield db
    finally:
        db.close()