Generated Architecture¶

Understanding the project structure and patterns generated by FCube.

Project Overview¶

my_project/
├── app/                     # Application code
│   ├── apis/
│   │   └── v1.py            # API versioning
│   └── core/                # Core infrastructure
│       ├── __init__.py
│       ├── database.py      # Database connection
│       ├── models.py        # Base models
│       ├── settings.py      # Configuration
│       ├── crud.py          # Base CRUD class
│       ├── exceptions.py    # Custom exceptions
│       ├── logging.py       # Logging configuration
│       ├── main.py          # FastAPI entry point
│       ├── dependencies.py  # Dependency injection
│       ├── alembic_models_import.py  # For migrations
│       └── celery_app.py    # Celery configuration
├── migrations/              # Database migrations (Alembic)
├── docker/                  # Docker configuration
│   ├── Dockerfile
│   └── docker-entrypoint.sh
├── docker-compose.yml       # Multi-container setup
├── alembic.ini              # Alembic configuration
├── pyproject.toml           # Python dependencies
├── .env.example             # Environment variables template
├── .gitignore               # Git ignore rules
└── README.md                # Project documentation

Layered Architecture¶

FCube follows the Layered Architecture pattern:

┌─────────────────────────────────────┐
│          Routes (HTTP Layer)        │
│  - Request validation               │
│  - Authentication/Authorization     │
│  - Response serialization           │
└───────────────┬─────────────────────┘
                 │
┌───────────────▼─────────────────────┐
│        Services (Business Logic)    │
│  - Business rules                   │
│  - Transaction boundaries           │
│  - Orchestration                    │
└───────────────┬─────────────────────┘
                 │
┌───────────────▼─────────────────────┐
│         CRUD (Data Access)          │
│  - Pure database operations         │
│  - NO session.commit()              │
│  - flush() and refresh() only       │
└───────────────┬─────────────────────┘
                 │
┌───────────────▼─────────────────────┐
│        Models (Database Schema)     │
│  - SQLAlchemy ORM models            │
│  - Relationships                    │
└─────────────────────────────────────┘

Request Flow¶

HTTP Request
    │
    ▼
┌─────────────────┐
│     Route       │  Validates input, calls service
└────────┬────────┘
         │
         ▼
┌─────────────────┐
│    Service      │  Orchestrates business logic, manages transactions
└────────┬────────┘
         │
         ▼
┌─────────────────┐
│      CRUD       │  Executes database operations (no commit)
└────────┬────────┘
         │
         ▼
┌─────────────────┐
│     Model       │  Maps to database tables
└─────────────────┘

Core Components¶

main.py¶

The application entry point:

from fastapi import FastAPI
from app.apis import v1
from app.core.config import settings
from app.core.database import engine

app = FastAPI(
    title=settings.PROJECT_NAME,
    version="1.0.0",
    docs_url="/docs",
    redoc_url="/redoc",
)

# Include API router
app.include_router(v1.router, prefix="/api/v1")


@app.on_event("startup")
async def startup():
    # Initialize database connection
    pass


@app.on_event("shutdown")
async def shutdown():
    # Cleanup
    pass

settings.py¶

Centralized settings using Pydantic:

from pydantic_settings import BaseSettings


class Settings(BaseSettings):
    # Database
    DATABASE_URL: str

    # Security
    SECRET_KEY: str
    JWT_SECRET_KEY: str
    JWT_ALGORITHM: str = "HS256"
    ACCESS_TOKEN_EXPIRE_MINUTES: int = 30

    # Application
    PROJECT_NAME: str = "MyApp"
    DEBUG: bool = False

    class Config:
        env_file = ".env"
        case_sensitive = True


settings = Settings()

database.py¶

Async database connection:

from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker, declarative_base

from app.core.config import settings

# Create async engine
engine = create_async_engine(
    settings.DATABASE_URL,
    echo=settings.DEBUG,
    pool_pre_ping=True,
)

# Session factory
AsyncSessionLocal = sessionmaker(
    engine,
    class_=AsyncSession,
    expire_on_commit=False,
)

# Base class for models
Base = declarative_base()


async def get_db():
    """Dependency for database sessions."""
    async with AsyncSessionLocal() as session:
        try:
            yield session
            await session.commit()
        except Exception:
            await session.rollback()
            raise
        finally:
            await session.close()

Module Structure¶

Each module follows the same pattern:

Model (`models/product.py`)¶

from datetime import datetime
from uuid import UUID, uuid4
from sqlalchemy import String, Numeric, Boolean, DateTime, func
from sqlalchemy.orm import Mapped, mapped_column

from app.core.database import Base


class Product(Base):
    __tablename__ = "products"

    id: Mapped[UUID] = mapped_column(primary_key=True, default=uuid4)
    name: Mapped[str] = mapped_column(String(255))
    description: Mapped[str | None] = mapped_column(String(1000))
    price: Mapped[Decimal] = mapped_column(Numeric(10, 2))
    is_active: Mapped[bool] = mapped_column(default=True)
    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True),
        server_default=func.now()
    )
    updated_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True),
        server_default=func.now(),
        onupdate=func.now()
    )

Schemas (`schemas/product_schemas.py`)¶

from datetime import datetime
from decimal import Decimal
from uuid import UUID
from pydantic import BaseModel, Field


class ProductBase(BaseModel):
    name: str = Field(..., min_length=1, max_length=255)
    description: str | None = Field(None, max_length=1000)
    price: Decimal = Field(..., gt=0)


class ProductCreate(ProductBase):
    pass


class ProductUpdate(BaseModel):
    name: str | None = Field(None, min_length=1, max_length=255)
    description: str | None = None
    price: Decimal | None = Field(None, gt=0)
    is_active: bool | None = None


class ProductResponse(ProductBase):
    id: UUID
    is_active: bool
    created_at: datetime
    updated_at: datetime

    model_config = {"from_attributes": True}

CRUD (`crud/product_crud.py`)¶

from uuid import UUID
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from app.models.product import Product
from app.schemas.product_schemas import ProductCreate, ProductUpdate


class ProductCRUD:
    def __init__(self, db: AsyncSession):
        self.db = db

    async def create(self, data: ProductCreate) -> Product:
        product = Product(**data.model_dump())
        self.db.add(product)
        await self.db.flush()
        await self.db.refresh(product)
        return product

    async def get(self, id: UUID) -> Product | None:
        stmt = select(Product).where(Product.id == id)
        result = await self.db.execute(stmt)
        return result.scalar_one_or_none()

    async def get_all(
        self,
        skip: int = 0,
        limit: int = 100,
        active_only: bool = True,
    ) -> list[Product]:
        stmt = select(Product)
        if active_only:
            stmt = stmt.where(Product.is_active == True)
        stmt = stmt.offset(skip).limit(limit)
        result = await self.db.execute(stmt)
        return result.scalars().all()

    async def update(self, id: UUID, data: ProductUpdate) -> Product | None:
        product = await self.get(id)
        if not product:
            return None

        update_data = data.model_dump(exclude_unset=True)
        for key, value in update_data.items():
            setattr(product, key, value)

        await self.db.flush()
        await self.db.refresh(product)
        return product

    async def delete(self, id: UUID) -> bool:
        product = await self.get(id)
        if not product:
            return False
        await self.db.delete(product)
        return True

Service (`services/product_service.py`)¶

from uuid import UUID
from fastapi import HTTPException, status

from app.crud.product_crud import ProductCRUD
from app.schemas.product_schemas import ProductCreate, ProductUpdate, ProductResponse


class ProductService:
    def __init__(self, crud: ProductCRUD):
        self.crud = crud

    async def create_product(self, data: ProductCreate) -> ProductResponse:
        product = await self.crud.create(data)
        return ProductResponse.model_validate(product)

    async def get_product(self, id: UUID) -> ProductResponse:
        product = await self.crud.get(id)
        if not product:
            raise HTTPException(
                status_code=status.HTTP_404_NOT_FOUND,
                detail="Product not found"
            )
        return ProductResponse.model_validate(product)

    async def list_products(
        self,
        skip: int = 0,
        limit: int = 100,
    ) -> list[ProductResponse]:
        products = await self.crud.get_all(skip=skip, limit=limit)
        return [ProductResponse.model_validate(p) for p in products]

    async def update_product(
        self,
        id: UUID,
        data: ProductUpdate,
    ) -> ProductResponse:
        product = await self.crud.update(id, data)
        if not product:
            raise HTTPException(
                status_code=status.HTTP_404_NOT_FOUND,
                detail="Product not found"
            )
        return ProductResponse.model_validate(product)

    async def delete_product(self, id: UUID) -> None:
        deleted = await self.crud.delete(id)
        if not deleted:
            raise HTTPException(
                status_code=status.HTTP_404_NOT_FOUND,
                detail="Product not found"
            )

Routes (`routes/public/product.py`)¶

from uuid import UUID
from fastapi import APIRouter, Depends, status

from app.schemas.product_schemas import ProductCreate, ProductUpdate, ProductResponse
from app.services.product_service import ProductService
from app.core.dependencies import get_product_service

router = APIRouter()


@router.post("/", response_model=ProductResponse, status_code=status.HTTP_201_CREATED)
async def create_product(
    data: ProductCreate,
    service: ProductService = Depends(get_product_service),
):
    """Create a new product."""
    return await service.create_product(data)


@router.get("/", response_model=list[ProductResponse])
async def list_products(
    skip: int = 0,
    limit: int = 100,
    service: ProductService = Depends(get_product_service),
):
    """List all products."""
    return await service.list_products(skip=skip, limit=limit)


@router.get("/{id}", response_model=ProductResponse)
async def get_product(
    id: UUID,
    service: ProductService = Depends(get_product_service),
):
    """Get a product by ID."""
    return await service.get_product(id)


@router.patch("/{id}", response_model=ProductResponse)
async def update_product(
    id: UUID,
    data: ProductUpdate,
    service: ProductService = Depends(get_product_service),
):
    """Update a product."""
    return await service.update_product(id, data)


@router.delete("/{id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_product(
    id: UUID,
    service: ProductService = Depends(get_product_service),
):
    """Delete a product."""
    await service.delete_product(id)

Dependency Injection¶

dependencies.py¶

from functools import lru_cache
from fastapi import Depends
from sqlalchemy.ext.asyncio import AsyncSession

from app.core.database import get_db
from app.crud.product_crud import ProductCRUD
from app.services.product_service import ProductService


@lru_cache()
def get_product_crud(
    db: AsyncSession = Depends(get_db),
) -> ProductCRUD:
    return ProductCRUD(db)


@lru_cache()
def get_product_service(
    crud: ProductCRUD = Depends(get_product_crud),
) -> ProductService:
    return ProductService(crud)

Benefits¶

Testability — Mock dependencies easily
Clarity — All dependencies are explicit
Singleton Pattern — @lru_cache() ensures single instance
Flexibility — Swap implementations without changing code

Key Patterns¶

Transaction Management¶

# CRUD: No commit
async def create(self, session, obj_in):
    db_obj = self.model(**obj_in.model_dump())
    session.add(db_obj)
    await session.flush()
    await session.refresh(db_obj)
    # NO commit here
    return db_obj

# Service: Owns commit
async def create_product(self, session, data):
    product = await product_crud.create(session, obj_in=data)
    await session.commit()
    await session.refresh(product)
    return product

Permission System¶

# permissions.py
PRODUCTS_READ = "products:read"
PRODUCTS_WRITE = "products:write"

def require_product_write_permission():
    return require_permission(PRODUCTS_WRITE)

# In routes
@router.post("/", dependencies=[Depends(require_product_write_permission)])
async def create_product(...):
    ...

Role-Based Routes¶

app/product/routes/
├── public/              # Public endpoints
│   ├── __init__.py
│   └── product.py
└── admin/               # Admin endpoints
    ├── __init__.py
    └── product_management.py

Key Architecture Decisions¶

Decision	Reason
Async SQLAlchemy	Better scalability for I/O bound operations
No Commit in CRUD	Services manage transaction boundaries
@lru_cache() for DI	Simple singleton pattern with minimal overhead
Separate Route Files	Clear distinction between public and admin endpoints
Alembic Migrations	Industry standard for database version control
Docker Compose	Easy development and production setup