Backend Patterns

Quick Reference

Project Structure

src/backend/
├── api/
│   ├── v1/                    # API routers
│   │   └── router_{feature}.py
│   ├── schemas/               # Pydantic DTOs
│   │   ├── _base.py          # CamelModel base class
│   │   └── {feature}.py
│   ├── services/              # Business logic
│   │   └── {feature}_service.py
│   ├── repositories/          # Data access
│   │   └── {feature}_repository.py
│   └── deps.py               # Dependency injection
├── db/
│   ├── models.py             # SQLAlchemy models
│   └── maria_database.py     # Database connection
├── core/
│   ├── exceptions.py         # Domain exceptions
│   └── pagination.py         # Pagination utilities
└── app.py                    # FastAPI application

File Naming Conventions

Core Patterns

1. Schema Pattern (CRITICAL)

# ALWAYS inherit from CamelModel, NOT BaseModel
from api.schemas._base import CamelModel

class ProductCreate(CamelModel):
    name_en: str           # Becomes "nameEn" in JSON
    is_active: bool        # Becomes "isActive" in JSON
    category_id: int       # Becomes "categoryId" in JSON

    model_config = ConfigDict(from_attributes=True)

DO NOT:

# WRONG - Don't use BaseModel directly
from pydantic import BaseModel
class ProductCreate(BaseModel):  # ❌ Wrong!
    pass

# WRONG - Don't use manual aliases
class ProductCreate(CamelModel):
    name: str = Field(alias="name")  # ❌ Unnecessary

2. Router Pattern

from fastapi import APIRouter, Depends, status
from sqlalchemy.ext.asyncio import AsyncSession
from api.deps import get_session, require_admin

router = APIRouter(prefix="/products", tags=["products"])

@router.post("", response_model=ProductResponse, status_code=status.HTTP_201_CREATED)
async def create_product(
    data: ProductCreate,
    session: AsyncSession = Depends(get_session),
    _: dict = Depends(require_admin),
):
    """Create a new product."""
    service = ProductService()
    return await service.create(session, data)

3. Service Pattern

class ProductService:
    def __init__(self):
        self._repo = ProductRepository()
        # ❌ DON'T: self._session = session

    async def create(self, session: AsyncSession, data: ProductCreate) -> Product:
        # Validate
        if not data.name_en:
            raise ValidationError(errors=[{"field": "name_en", "message": "Required"}])

        # Check conflicts
        existing = await self._repo.get_by_name(session, data.name_en)
        if existing:
            raise ConflictError(entity="Product", field="name_en", value=data.name_en)

        # Create
        entity = Product(name_en=data.name_en, ...)
        return await self._repo.create(session, entity)

4. Repository Pattern

class ProductRepository:
    async def create(self, session: AsyncSession, entity: Product) -> Product:
        session.add(entity)
        await session.flush()  # NOT commit()!
        await session.refresh(entity)
        return entity

    async def get_by_id(self, session: AsyncSession, id: str) -> Optional[Product]:
        result = await session.execute(
            select(Product).where(Product.id == id)
        )
        return result.scalar_one_or_none()

5. Model Pattern

from sqlalchemy import String, Boolean, DateTime, func
from sqlalchemy.orm import Mapped, mapped_column, relationship
from db.maria_database import Base

class Product(Base):
    __tablename__ = "product"

    id: Mapped[str] = mapped_column(CHAR(36), primary_key=True, default=lambda: str(uuid.uuid4()))
    name_en: Mapped[str] = mapped_column(String(128), nullable=False)
    name_ar: Mapped[Optional[str]] = mapped_column(String(128), nullable=True)
    is_active: Mapped[bool] = mapped_column(Boolean, default=True)
    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now())

Exception Hierarchy

# In service
raise NotFoundError(entity="Product", identifier=product_id)
raise ConflictError(entity="Product", field="name_en", value=name)
raise ValidationError(errors=[{"field": "price", "message": "Must be positive"}])

Dependency Injection

# Session dependency (use in every endpoint)