orion/docs/backend/overview.md

Backend Development

Guide for developing backend features, services, and API endpoints.

Overview

The Orion backend is built with FastAPI and follows a service-oriented architecture pattern. This guide covers backend development practices, patterns, and technical references.

Backend Structure

app/
├── api/                    # API routes (REST endpoints)
│   ├── v1/                # Version 1 API
│   │   ├── admin/        # Admin API endpoints
│   │   ├── store/       # Store API endpoints
│   │   └── shop/         # Shop API endpoints
│   └── main.py           # API router configuration
│
├── routes/               # Page routes (HTML)
│   ├── admin_pages.py   # Admin page routes
│   ├── store_pages.py  # Store page routes
│   └── shop_pages.py    # Shop page routes
│
├── services/            # Business logic layer
│   ├── admin_service.py
│   ├── product_service.py
│   ├── order_service.py
│   └── ...
│
├── core/                # Core functionality
│   ├── database.py     # Database connection
│   ├── config.py       # Configuration
│   └── lifespan.py     # Application lifecycle
│
└── exceptions/          # Custom exceptions
    ├── base.py
    └── handler.py

Development Workflow

1. Creating a New API Endpoint

Steps:

1. Define Pydantic schema in models/schema/
2. Create service method in app/services/
3. Add API route in app/api/v1/
4. Write tests in tests/

Example - Adding a new product endpoint:

# Step 1: Schema (models/schema/product.py)
from pydantic import BaseModel

class ProductCreate(BaseModel):
    name: str
    description: str
    price: float
    store_id: int

class ProductResponse(BaseModel):
    id: int
    name: str
    description: str
    price: float
    store_id: int

    class Config:
        from_attributes = True

# Step 2: Service (app/services/product_service.py)
from sqlalchemy.orm import Session
from models.database.product import Product

class ProductService:
    def create_product(self, db: Session, data: ProductCreate) -> Product:
        product = Product(**data.dict())
        db.add(product)
        db.commit()
        db.refresh(product)
        return product

product_service = ProductService()

# Step 3: API Route (app/api/v1/shop/products.py)
from fastapi import APIRouter, Depends
from sqlalchemy.orm import Session
from app.core.database import get_db
from app.services.product_service import product_service
from app.modules.catalog.schemas import ProductCreate, ProductResponse

router = APIRouter(prefix="/products")

@router.post("/", response_model=ProductResponse)
async def create_product(
    data: ProductCreate,
    db: Session = Depends(get_db)
):
    return product_service.create_product(db, data)

2. Creating a Service

Services contain business logic and should be reusable across routes.

Pattern:

# app/services/example_service.py
from sqlalchemy.orm import Session
from typing import List, Optional
import logging

logger = logging.getLogger(__name__)

class ExampleService:
    """Service for handling example operations."""

    def get_items(
        self,
        db: Session,
        store_id: int,
        skip: int = 0,
        limit: int = 100
    ) -> List[Item]:
        """Get items for a store with pagination."""
        try:
            items = db.query(Item).filter(
                Item.store_id == store_id
            ).offset(skip).limit(limit).all()

            logger.info(f"Retrieved {len(items)} items for store {store_id}")
            return items

        except Exception as e:
            logger.error(f"Error retrieving items: {e}")
            raise

    def create_item(
        self,
        db: Session,
        store_id: int,
        data: ItemCreate
    ) -> Item:
        """Create a new item."""
        try:
            item = Item(
                store_id=store_id,
                **data.dict()
            )
            db.add(item)
            db.commit()
            db.refresh(item)

            logger.info(f"Created item {item.id} for store {store_id}")
            return item

        except Exception as e:
            logger.error(f"Error creating item: {e}")
            db.rollback()
            raise

# Singleton instance
example_service = ExampleService()

3. Database Operations

Best Practices:

# ✅ Good - Use service layer
@router.get("/products")
async def get_products(
    store_id: int,
    db: Session = Depends(get_db)
):
    products = product_service.get_products(db, store_id)
    return {"products": products}

# ❌ Bad - Database queries in route handler
@router.get("/products")
async def get_products(
    store_id: int,
    db: Session = Depends(get_db)
):
    products = db.query(Product).filter(
        Product.store_id == store_id
    ).all()
    return {"products": products}

Always Scope to Store:

# ✅ Good - Scoped to store
products = db.query(Product).filter(
    Product.store_id == request.state.store_id
).all()

# ❌ Bad - Not scoped, security risk!
products = db.query(Product).all()

Dependency Injection

FastAPI's dependency system provides clean code organization:

from fastapi import Depends
from sqlalchemy.orm import Session
from app.core.database import get_db
from middleware.auth import auth_manager
from models.database.user import User

@router.get("/protected")
async def protected_endpoint(
    current_user: User = Depends(auth_manager.get_current_user),
    db: Session = Depends(get_db)
):
    # current_user is automatically injected
    # db session is automatically injected
    return {"user": current_user.username}

Error Handling

Custom Exceptions:

from app.exceptions import (
    ResourceNotFoundException,
    ValidationException,
    InsufficientPermissionsException
)

@router.get("/products/{product_id}")
async def get_product(
    product_id: int,
    db: Session = Depends(get_db)
):
    product = db.query(Product).filter(Product.id == product_id).first()

    if not product:
        raise ResourceNotFoundException(f"Product {product_id} not found")

    return product

Exception Handlers are configured globally in app/exceptions/handler.py

Testing

Unit Tests

# tests/unit/services/test_product_service.py
import pytest
from app.services.product_service import ProductService

def test_create_product(db_session):
    service = ProductService()
    data = ProductCreate(
        name="Test Product",
        price=29.99,
        store_id=1
    )

    product = service.create_product(db_session, data)

    assert product.id is not None
    assert product.name == "Test Product"
    assert product.store_id == 1

Integration Tests

# tests/integration/test_product_api.py
def test_create_product_endpoint(client, auth_headers):
    response = client.post(
        "/api/v1/products",
        json={
            "name": "Test Product",
            "price": 29.99,
            "store_id": 1
        },
        headers=auth_headers
    )

    assert response.status_code == 200
    data = response.json()
    assert data["name"] == "Test Product"

Code Quality

Linting

# Run linters
make lint

# Auto-fix formatting
make format

Type Checking

# Run mypy
make lint

Use Type Hints:

from typing import List, Optional
from sqlalchemy.orm import Session

def get_items(
    db: Session,
    store_id: int,
    limit: Optional[int] = None
) -> List[Item]:
    query = db.query(Item).filter(Item.store_id == store_id)

    if limit:
        query = query.limit(limit)

    return query.all()

Database Migrations

Creating a Migration:

make migrate-create message="add_product_table"

Applying Migrations:

make migrate-up

See: Database Migrations

API Documentation

FastAPI automatically generates API documentation:

• Swagger UI: http://localhost:8000/docs
• ReDoc: http://localhost:8000/redoc
• OpenAPI JSON: http://localhost:8000/openapi.json

Document your endpoints:

@router.post(
    "/products",
    response_model=ProductResponse,
    summary="Create a new product",
    description="Creates a new product for the authenticated store",
    responses={
        201: {"description": "Product created successfully"},
        400: {"description": "Invalid product data"},
        401: {"description": "Not authenticated"},
        403: {"description": "Not authorized to create products"}
    }
)
async def create_product(
    data: ProductCreate,
    current_user: User = Depends(auth_manager.require_store),
    db: Session = Depends(get_db)
):
    """
    Create a new product.

    - **name**: Product name (required)
    - **description**: Product description
    - **price**: Product price (required)
    - **store_id**: Store ID (required)
    """
    return product_service.create_product(db, current_user.id, data)

Configuration

Environment Variables (.env):

# Database
DATABASE_URL=postgresql://user:password@localhost:5432/dbname

# JWT
JWT_SECRET_KEY=your-secret-key
JWT_EXPIRE_MINUTES=30

# Application
ENVIRONMENT=development
DEBUG=true
LOG_LEVEL=INFO

Accessing Config:

from app.core.config import settings

database_url = settings.database_url
secret_key = settings.jwt_secret_key

Logging

Use Structured Logging:

import logging

logger = logging.getLogger(__name__)

def process_order(order_id: int):
    logger.info(f"Processing order", extra={"order_id": order_id})

    try:
        # Process order
        logger.info(f"Order processed successfully", extra={"order_id": order_id})

    except Exception as e:
        logger.error(
            f"Order processing failed",
            extra={"order_id": order_id, "error": str(e)},
            exc_info=True
        )
        raise

Performance

Database Query Optimization

# ✅ Good - Use joins to avoid N+1 queries
from sqlalchemy.orm import joinedload

products = db.query(Product).options(
    joinedload(Product.category),
    joinedload(Product.store)
).filter(Product.store_id == store_id).all()

# ❌ Bad - N+1 query problem
products = db.query(Product).filter(
    Product.store_id == store_id
).all()

for product in products:
    category = product.category  # Additional query!

Caching

from functools import lru_cache

@lru_cache(maxsize=100)
def get_store_settings(store_id: int) -> dict:
    # Expensive operation cached in memory
    return db.query(StoreSettings).filter(
        StoreSettings.store_id == store_id
    ).first()

Security Best Practices

1. Always validate input with Pydantic schemas
2. Always scope queries to store/user
3. Use parameterized queries (SQLAlchemy ORM does this)
4. Never log sensitive data (passwords, tokens, credit cards)
5. Use HTTPS in production
6. Implement rate limiting on sensitive endpoints
7. Validate file uploads (type, size, content)
8. Sanitize user input before rendering in templates

Next Steps

• Middleware Reference - Technical API documentation
• Architecture Overview - System architecture
• Database Migrations - Migration guide
• Testing Guide - Testing practices