oicd/docs/guides/python-quick-start-guide.md

Python Developer Quick Start Guide

FastAPI & Flask - Essential Patterns for Daily Development

This is your ONE reference document for Python API development. Everything you need to write clean, production-ready code.

---

Table of Contents

1. Project Setup
2. Project Structure
3. The Four Layers
4. Code Standards
5. Module Size Guidelines
6. Database Patterns
7. Error Handling
8. Logging
9. Testing
10. Security Essentials
11. Daily Checklist

---

Project Setup

Initial Setup

# Create project
mkdir my-api && cd my-api

# Create virtual environment
python -m venv venv
source venv/bin/activate  # Linux/Mac
# venv\Scripts\activate   # Windows

# Install dependencies
pip install fastapi uvicorn sqlalchemy pydantic-settings
pip install pytest pytest-cov black ruff mypy --dev

# Create requirements files
pip freeze > requirements.txt

Essential Configuration Files

pyproject.toml

[tool.black]
line-length = 100
target-version = ['py311']

[tool.ruff]
line-length = 100
select = ["E", "W", "F", "I", "B", "C4", "UP"]

[tool.mypy]
python_version = "3.11"
warn_return_any = true
disallow_untyped_defs = true

[tool.pytest.ini_options]
testpaths = ["tests"]
addopts = "--cov=app --cov-report=html"

.env.example

# Application
ENVIRONMENT=development
LOG_LEVEL=INFO
SECRET_KEY=your-secret-key-change-in-production

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

# External Services
REDIS_URL=redis://localhost:6379/0
SMTP_HOST=smtp.gmail.com
SMTP_PORT=587

---

Project Structure

FastAPI Structure (Recommended)

my-api/
├── app/
│   ├── __init__.py
│   ├── main.py                    # FastAPI app initialization
│   ├── config.py                  # Settings (Pydantic)
│   ├── dependencies.py            # Dependency injection
│   │
│   ├── api/
│   │   └── v1/
│   │       ├── endpoints/
│   │       │   ├── users.py       # Thin endpoints
│   │       │   └── auth.py
│   │       └── router.py
│   │
│   ├── services/                  # Business logic HERE
│   │   ├── user_service.py
│   │   └── auth_service.py
│   │
│   ├── repositories/              # Database operations HERE
│   │   ├── base_repository.py
│   │   └── user_repository.py
│   │
│   ├── models/                    # SQLAlchemy models
│   │   └── user.py
│   │
│   ├── schemas/                   # Pydantic schemas
│   │   └── user.py
│   │
│   ├── core/
│   │   ├── database.py
│   │   ├── security.py
│   │   └── logging_config.py
│   │
│   ├── exceptions.py
│   └── utils/
│
├── tests/
│   ├── conftest.py
│   ├── test_api/
│   └── test_services/
│
├── .env
├── .env.example
├── .gitignore
├── pyproject.toml
├── requirements.txt
└── README.md

---

The Four Layers

CRITICAL: Always follow this pattern to avoid spaghetti code.

┌─────────────────────────────────────┐
│   1. API LAYER (endpoints)          │  ← Thin controllers (5-10 lines)
├─────────────────────────────────────┤
│   2. SERVICE LAYER                  │  ← Business logic lives HERE
├─────────────────────────────────────┤
│   3. REPOSITORY LAYER               │  ← Database operations only
├─────────────────────────────────────┤
│   4. MODEL LAYER                    │  ← Data structures
└─────────────────────────────────────┘

1. API Layer (Thin Endpoints)

Rule: Endpoints should be < 10 lines. Just call service layer.

# app/api/v1/endpoints/users.py
from fastapi import APIRouter, Depends, status
from typing import Annotated

from app.schemas.user import UserCreate, UserResponse
from app.services.user_service import UserService
from app.dependencies import get_user_service

router = APIRouter()


@router.post("/users", response_model=UserResponse, status_code=status.HTTP_201_CREATED)
async def create_user(
    user_in: UserCreate,
    service: Annotated[UserService, Depends(get_user_service)],
) -> UserResponse:
    """Create new user - endpoint is THIN."""
    return await service.register_user(user_in)


@router.get("/users/{user_id}", response_model=UserResponse)
async def get_user(
    user_id: int,
    service: Annotated[UserService, Depends(get_user_service)],
) -> UserResponse:
    """Get user by ID."""
    return await service.get_user(user_id)

Key Points:

• ✅ No business logic in endpoints
• ✅ No database calls in endpoints
• ✅ Just parse input → call service → return response

---

2. Service Layer (Business Logic)

Rule: ALL business logic goes here. Services orchestrate workflows.

# app/services/user_service.py
from sqlalchemy.orm import Session
from loguru import logger

from app.repositories.user_repository import UserRepository
from app.schemas.user import UserCreate
from app.models.user import User
from app.exceptions import ConflictError, NotFoundError
from app.core.security import get_password_hash


class UserService:
    """User service - contains ALL business logic for users."""

    def __init__(
        self,
        user_repo: UserRepository,
        email_service: EmailService,
        db: Session,
    ):
        self.user_repo = user_repo
        self.email_service = email_service
        self.db = db

    async def register_user(self, user_data: UserCreate) -> User:
        """
        Register a new user - complete workflow.

        Business Rules:
        1. Email must be unique
        2. Password must be hashed
        3. Send welcome email
        4. User starts inactive until email verified
        """
        logger.info("Registering user", extra={"email": user_data.email})

        # Rule 1: Check email uniqueness
        existing = await self.user_repo.get_by_email(user_data.email)
        if existing:
            raise ConflictError("Email already registered")

        # Rule 2: Hash password
        hashed_password = get_password_hash(user_data.password)

        # Create user
        try:
            user = await self.user_repo.create(
                email=user_data.email,
                username=user_data.username,
                hashed_password=hashed_password,
                is_active=False,  # Rule 4
            )

            # Rule 3: Send email
            await self.email_service.send_welcome_email(user.email)

            self.db.commit()
            logger.info("User registered", extra={"user_id": user.id})
            return user

        except Exception as e:
            self.db.rollback()
            logger.error("Registration failed", exc_info=True)
            raise

    async def get_user(self, user_id: int) -> User:
        """Get user by ID with business logic."""
        user = await self.user_repo.get_by_id(user_id)

        if not user:
            raise NotFoundError("User", user_id)

        # Business rule: Don't return deleted users
        if user.is_deleted:
            raise NotFoundError("User", user_id)

        return user

Key Points:

• ✅ Business rules are explicit
• ✅ Orchestrates multiple operations
• ✅ Handles transactions
• ✅ Logged properly

---

3. Repository Layer (Data Access)

Rule: Only database operations. No business logic.

# app/repositories/user_repository.py
from typing import Optional, List
from sqlalchemy.orm import Session

from app.models.user import User


class UserRepository:
    """User repository - handles ALL database operations for users."""

    def __init__(self, db: Session):
        self.db = db

    async def get_by_id(self, user_id: int) -> Optional[User]:
        """Get user by ID."""
        return self.db.query(User).filter(User.id == user_id).first()

    async def get_by_email(self, email: str) -> Optional[User]:
        """Get user by email."""
        return self.db.query(User).filter(User.email == email).first()

    async def get_all(self, skip: int = 0, limit: int = 100) -> List[User]:
        """Get all users with pagination."""
        return self.db.query(User).offset(skip).limit(limit).all()

    async def create(
        self,
        email: str,
        username: str,
        hashed_password: str,
        is_active: bool = True,
    ) -> User:
        """Create new user."""
        user = User(
            email=email,
            username=username,
            hashed_password=hashed_password,
            is_active=is_active,
        )
        self.db.add(user)
        self.db.flush()  # Get ID without committing
        return user

    async def update(self, user: User, **kwargs) -> User:
        """Update user."""
        for key, value in kwargs.items():
            setattr(user, key, value)
        self.db.flush()
        return user

    async def delete(self, user: User) -> None:
        """Delete user."""
        self.db.delete(user)
        self.db.flush()

Key Points:

• ✅ Pure data access
• ✅ No business logic
• ✅ Easy to test with mocks

---

4. Model Layer (Domain Models)

Rule: Data structure + simple helper methods only.

# app/models/user.py
from sqlalchemy import Column, Integer, String, Boolean, DateTime
from datetime import datetime

from app.core.database import Base


class User(Base):
    """User domain model."""
    __tablename__ = "users"

    id = Column(Integer, primary_key=True, index=True)
    email = Column(String(255), unique=True, index=True, nullable=False)
    username = Column(String(100), unique=True, nullable=False)
    hashed_password = Column(String(255), nullable=False)
    is_active = Column(Boolean, default=True, nullable=False)
    is_deleted = Column(Boolean, default=False, nullable=False)
    created_at = Column(DateTime, default=datetime.utcnow)

    def __repr__(self) -> str:
        return f"<User {self.username}>"

    # Simple domain methods are OK
    def can_login(self) -> bool:
        """Check if user can login."""
        return self.is_active and not self.is_deleted

---

Dependency Injection

# app/dependencies.py
from typing import Annotated, Generator
from fastapi import Depends
from sqlalchemy.orm import Session

from app.core.database import SessionLocal
from app.repositories.user_repository import UserRepository
from app.services.user_service import UserService
from app.services.email_service import EmailService


def get_db() -> Generator[Session, None, None]:
    """Get database session."""
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()


def get_user_repository(
    db: Annotated[Session, Depends(get_db)]
) -> UserRepository:
    """Get user repository."""
    return UserRepository(db)


def get_email_service() -> EmailService:
    """Get email service."""
    return EmailService()


def get_user_service(
    user_repo: Annotated[UserRepository, Depends(get_user_repository)],
    email_service: Annotated[EmailService, Depends(get_email_service)],
    db: Annotated[Session, Depends(get_db)],
) -> UserService:
    """Get user service with all dependencies."""
    return UserService(
        user_repo=user_repo,
        email_service=email_service,
        db=db,
    )

---

Code Standards

Type Hints (MANDATORY)

# BAD: No type hints
def get_user(user_id):
    return db.query(User).get(user_id)

# GOOD: Clear type hints
def get_user(user_id: int) -> Optional[User]:
    return db.query(User).filter(User.id == user_id).first()

# Modern Python 3.10+ syntax
def get_users(limit: int = 10) -> list[User]:
    return db.query(User).limit(limit).all()

Pydantic Schemas

# app/schemas/user.py
from pydantic import BaseModel, EmailStr, Field, ConfigDict
from datetime import datetime


class UserBase(BaseModel):
    email: EmailStr
    username: str = Field(..., min_length=3, max_length=50)


class UserCreate(UserBase):
    password: str = Field(..., min_length=8)


class UserUpdate(BaseModel):
    email: EmailStr | None = None
    username: str | None = None


class UserResponse(UserBase):
    id: int
    is_active: bool
    created_at: datetime

    model_config = ConfigDict(from_attributes=True)

Naming Conventions

# Constants
MAX_CONNECTIONS = 100
API_BASE_URL = "https://api.example.com"

# Classes
class UserService:
    pass

# Functions and variables
def get_user_by_email(email: str) -> User:
    pass

# Private methods
def _internal_helper():
    pass

# Boolean variables
is_active = True
has_permission = False
should_retry = True

---

Module Size Guidelines

Maximum Lines Per Module

Keep modules focused and manageable:

Module Type	Ideal Size	Maximum	Action If Exceeds
Endpoints	200-250 lines	300 lines	Split by resource/feature
Services	300-400 lines	500 lines	Split by subdomain
Repositories	200-300 lines	400 lines	Use base repository pattern
Models	150-200 lines	300 lines	One model per file
Utils	100-150 lines	200 lines	Split by function category

General Rule: Never exceed 1000 lines in a single module.

When to Split a Module

Warning Signs:

• Hard to find specific functions
• Constant scrolling up and down
• Multiple unrelated responsibilities
• Test file is becoming huge
• Frequent merge conflicts

Example: Splitting Large Service

# BAD: One giant service (1200 lines)
# app/services/user_service.py
class UserService:
    def register_user(): pass        # 80 lines
    def login(): pass                 # 60 lines
    def logout(): pass                # 40 lines
    def reset_password(): pass        # 70 lines
    def update_profile(): pass        # 50 lines
    def upload_avatar(): pass         # 80 lines
    def send_notification(): pass     # 60 lines
    def export_data(): pass           # 90 lines
    # ... 15 more methods (1200 lines total) ❌

# GOOD: Split into focused services
# app/services/user_service.py (400 lines)
class UserService:
    def register_user(): pass
    def get_user(): pass
    def update_user(): pass
    def delete_user(): pass
    # Core CRUD operations ✅

# app/services/user_auth_service.py (300 lines)
class UserAuthService:
    def login(): pass
    def logout(): pass
    def reset_password(): pass
    def verify_email(): pass
    # Authentication logic ✅

# app/services/user_profile_service.py (250 lines)
class UserProfileService:
    def update_profile(): pass
    def upload_avatar(): pass
    def get_statistics(): pass
    # Profile management ✅

# app/services/user_notification_service.py (200 lines)
class UserNotificationService:
    def send_welcome_email(): pass
    def send_notification(): pass
    # Notifications ✅

Function Size Guidelines

# Functions should be short and focused
# Ideal: 10-30 lines
# Maximum: 50 lines

# BAD: Function too long (100+ lines)
def process_order(order_data):
    # 100 lines of logic
    pass  # ❌

# GOOD: Break into smaller functions
def process_order(order_data):
    """Process order - orchestrates workflow."""
    validate_order(order_data)        # 15 lines
    calculate_total(order_data)       # 20 lines
    process_payment(order_data)       # 25 lines
    send_confirmation(order_data)     # 15 lines
    # Each function is focused and testable ✅

---

Database Patterns

Database Setup

# app/core/database.py
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, declarative_base

from app.core.config import settings

engine = create_engine(
    settings.DATABASE_URL,
    pool_pre_ping=True,
    pool_size=10,
    max_overflow=20,
)

SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()

Models with Relationships

from sqlalchemy import Column, Integer, String, ForeignKey
from sqlalchemy.orm import relationship


class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True)
    email = Column(String, unique=True, nullable=False)

    # Relationships
    posts = relationship("Post", back_populates="author", cascade="all, delete-orphan")


class Post(Base):
    __tablename__ = "posts"

    id = Column(Integer, primary_key=True)
    title = Column(String, nullable=False)
    author_id = Column(Integer, ForeignKey("users.id"), nullable=False)

    # Relationships
    author = relationship("User", back_populates="posts")

Query Optimization

# BAD: N+1 query problem
users = db.query(User).all()
for user in users:
    print(user.posts)  # Separate query for each user!

# GOOD: Eager loading
from sqlalchemy.orm import joinedload

users = db.query(User).options(joinedload(User.posts)).all()
for user in users:
    print(user.posts)  # No additional queries

---

Error Handling

Custom Exception Hierarchy

# app/exceptions.py
from typing import Optional, Any


class AppException(Exception):
    """Base application exception."""

    def __init__(
        self,
        message: str,
        error_code: str,
        status_code: int = 500,
        details: dict[str, Any] | None = None,
    ):
        self.message = message
        self.error_code = error_code
        self.status_code = status_code
        self.details = details or {}
        super().__init__(self.message)


class ValidationError(AppException):
    def __init__(self, message: str, field: str | None = None):
        super().__init__(
            message=message,
            error_code="VALIDATION_ERROR",
            status_code=400,
            details={"field": field} if field else {},
        )


class NotFoundError(AppException):
    def __init__(self, resource: str, identifier: Any):
        super().__init__(
            message=f"{resource} not found",
            error_code="NOT_FOUND",
            status_code=404,
            details={"resource": resource, "identifier": str(identifier)},
        )


class ConflictError(AppException):
    def __init__(self, message: str):
        super().__init__(
            message=message,
            error_code="CONFLICT",
            status_code=409,
        )


class AuthenticationError(AppException):
    def __init__(self, message: str = "Authentication failed"):
        super().__init__(
            message=message,
            error_code="AUTHENTICATION_ERROR",
            status_code=401,
        )

Exception Handlers (FastAPI)

# app/main.py
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
from loguru import logger

from app.exceptions import AppException

app = FastAPI()


@app.exception_handler(AppException)
async def app_exception_handler(request: Request, exc: AppException):
    """Handle custom application exceptions."""

    # Log based on severity
    if exc.status_code >= 500:
        logger.error(f"Server error: {exc.message}", extra={"error_code": exc.error_code}, exc_info=True)
    else:
        logger.warning(f"Client error: {exc.message}", extra={"error_code": exc.error_code})

    return JSONResponse(
        status_code=exc.status_code,
        content={
            "error": {
                "code": exc.error_code,
                "message": exc.message,
                "details": exc.details,
            }
        },
    )


@app.exception_handler(Exception)
async def global_exception_handler(request: Request, exc: Exception):
    """Catch-all exception handler."""
    logger.critical("Unhandled exception", exc_info=True)

    return JSONResponse(
        status_code=500,
        content={
            "error": {
                "code": "INTERNAL_SERVER_ERROR",
                "message": "An unexpected error occurred",
                "details": {},
            }
        },
    )

---

Logging

Setup (Loguru - Recommended)

# app/core/logging_config.py
import sys
from loguru import logger
from app.core.config import settings


def setup_logging() -> None:
    """Configure application logging."""
    logger.remove()

    # Console logging
    if settings.ENVIRONMENT == "development":
        logger.add(
            sys.stdout,
            format="<green>{time:HH:mm:ss}</green> | <level>{level: <8}</level> | <cyan>{name}:{function}:{line}</cyan> | <level>{message}</level>",
            level=settings.LOG_LEVEL,
            colorize=True,
        )
    else:
        # JSON logging for production
        logger.add(
            sys.stdout,
            level=settings.LOG_LEVEL,
            serialize=True,
        )

    # File logging
    logger.add(
        "logs/app_{time:YYYY-MM-DD}.log",
        rotation="00:00",
        retention="30 days",
        level=settings.LOG_LEVEL,
        serialize=True,
    )

Request Logging Middleware

# app/middleware/logging.py
import time
import uuid
from fastapi import Request
from starlette.middleware.base import BaseHTTPMiddleware
from loguru import logger


class RequestLoggingMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        request_id = str(uuid.uuid4())
        request.state.request_id = request_id

        start_time = time.time()

        logger.info(
            "Incoming request",
            extra={
                "request_id": request_id,
                "method": request.method,
                "path": request.url.path,
            }
        )

        response = await call_next(request)
        duration = time.time() - start_time

        logger.info(
            "Request completed",
            extra={
                "request_id": request_id,
                "status_code": response.status_code,
                "duration_ms": round(duration * 1000, 2),
            }
        )

        response.headers["X-Request-ID"] = request_id
        return response


# Add to app
app.add_middleware(RequestLoggingMiddleware)

Log Levels Usage

# DEBUG - Detailed diagnostic info
logger.debug("Database query executed", extra={"query": sql, "duration_ms": 45.2})

# INFO - Important business events
logger.info("User registered", extra={"user_id": user.id})

# WARNING - Unexpected but handled
logger.warning("API rate limit approaching", extra={"requests": 950, "limit": 1000})

# ERROR - Operation failed but app continues
logger.error("Failed to send email", extra={"user_id": user.id}, exc_info=True)

# CRITICAL - System failure
logger.critical("Database connection pool exhausted", extra={"pool_size": 10})

---

Testing

Test Setup

# tests/conftest.py
import pytest
from typing import Generator
from fastapi.testclient import TestClient
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

from app.main import app
from app.core.database import Base
from app.dependencies import get_db

# Test database
SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db"
engine = create_engine(SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False})
TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)


@pytest.fixture(scope="function")
def db() -> Generator:
    """Create test database."""
    Base.metadata.create_all(bind=engine)
    db = TestingSessionLocal()
    try:
        yield db
    finally:
        db.close()
        Base.metadata.drop_all(bind=engine)


@pytest.fixture(scope="function")
def client(db) -> Generator:
    """Create test client."""
    def override_get_db():
        try:
            yield db
        finally:
            pass

    app.dependency_overrides[get_db] = override_get_db
    with TestClient(app) as test_client:
        yield test_client
    app.dependency_overrides.clear()

Testing Services (Unit Tests)

# tests/test_services/test_user_service.py
import pytest
from unittest.mock import Mock, AsyncMock

from app.services.user_service import UserService
from app.schemas.user import UserCreate
from app.exceptions import ConflictError


@pytest.fixture
def mock_user_repo():
    return Mock()


@pytest.fixture
def mock_email_service():
    return Mock()


@pytest.fixture
def user_service(mock_user_repo, mock_email_service):
    return UserService(
        user_repo=mock_user_repo,
        email_service=mock_email_service,
        db=Mock(),
    )


async def test_register_user_success(user_service, mock_user_repo):
    """Test successful user registration."""
    # Arrange
    mock_user_repo.get_by_email = AsyncMock(return_value=None)
    mock_user_repo.create = AsyncMock(return_value=Mock(id=1, email="test@example.com"))

    user_data = UserCreate(email="test@example.com", username="test", password="pass123")

    # Act
    result = await user_service.register_user(user_data)

    # Assert
    assert result.id == 1
    mock_user_repo.create.assert_called_once()


async def test_register_user_email_exists(user_service, mock_user_repo):
    """Test registration with existing email."""
    # Arrange
    mock_user_repo.get_by_email = AsyncMock(return_value=Mock(id=1))

    user_data = UserCreate(email="test@example.com", username="test", password="pass123")

    # Act & Assert
    with pytest.raises(ConflictError):
        await user_service.register_user(user_data)

Testing API Endpoints

# tests/test_api/test_users.py
def test_create_user(client):
    """Test POST /api/v1/users."""
    response = client.post(
        "/api/v1/users",
        json={
            "email": "test@example.com",
            "username": "testuser",
            "password": "password123",
        },
    )

    assert response.status_code == 201
    data = response.json()
    assert data["email"] == "test@example.com"
    assert "password" not in data


def test_get_user(client):
    """Test GET /api/v1/users/{id}."""
    # Create user first
    create_response = client.post("/api/v1/users", json={...})
    user_id = create_response.json()["id"]

    # Get user
    response = client.get(f"/api/v1/users/{user_id}")

    assert response.status_code == 200
    assert response.json()["id"] == user_id

---

Security Essentials

Password Hashing

# app/core/security.py
from passlib.context import CryptContext

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")


def get_password_hash(password: str) -> str:
    """Hash a password."""
    return pwd_context.hash(password)


def verify_password(plain_password: str, hashed_password: str) -> bool:
    """Verify a password."""
    return pwd_context.verify(plain_password, hashed_password)

JWT Authentication

from datetime import datetime, timedelta
import jwt

SECRET_KEY = "your-secret-key"
ALGORITHM = "HS256"


def create_access_token(user_id: int) -> str:
    """Create JWT access token."""
    expire = datetime.utcnow() + timedelta(minutes=30)
    to_encode = {"sub": str(user_id), "exp": expire}
    return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)


def decode_access_token(token: str) -> dict | None:
    """Decode JWT token."""
    try:
        return jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
    except jwt.PyJWTError:
        return None

Input Validation

from pydantic import validator


class UserCreate(BaseModel):
    email: EmailStr
    password: str

    @validator("password")
    def password_strength(cls, v: str) -> str:
        if len(v) < 8:
            raise ValueError("Password must be at least 8 characters")
        if not any(char.isdigit() for char in v):
            raise ValueError("Password must contain a digit")
        return v

SQL Injection Prevention

# BAD: Never do this
query = f"SELECT * FROM users WHERE email = '{email}'"

# GOOD: Use ORM
user = db.query(User).filter(User.email == email).first()

# GOOD: Parameterized queries
from sqlalchemy import text
query = text("SELECT * FROM users WHERE email = :email")
db.execute(query, {"email": email})

---

Daily Checklist

Before Writing Code

• Understand which layer you're working in
• Business logic goes in SERVICE layer
• Database operations go in REPOSITORY layer
• Endpoints stay THIN (< 10 lines)

While Writing Code

• Add type hints to all functions
• Log important operations
• Handle errors with custom exceptions
• Validate all inputs with Pydantic
• Use dependency injection
• Keep functions under 50 lines
• Keep modules under 500 lines

Before Committing

• Run tests: pytest
• Format code: black .
• Lint code: ruff check .
• Type check: mypy app/
• No hardcoded secrets
• No print() or console.log()

Code Review Checklist

• Endpoints are thin (just call service)
• Business logic is in services
• Database queries are in repositories
• All functions have type hints
• Functions are under 50 lines
• Modules are under 500 lines (split if larger)
• Errors are logged properly
• Tests are included
• No security vulnerabilities

---

Quick Reference Commands

# Development
uvicorn app.main:app --reload

# Testing
pytest
pytest --cov=app --cov-report=html

# Code Quality
black .
ruff check . --fix
mypy app/

# Database
alembic revision --autogenerate -m "message"
alembic upgrade head

---

Common Mistakes to Avoid

❌ DON'T: Put business logic in endpoints

@router.post("/users")
async def create_user(user_in: UserCreate, db: Session = Depends(get_db)):
    # Checking email exists - BUSINESS LOGIC!
    if db.query(User).filter(User.email == user_in.email).first():
        raise HTTPException(400, "Email exists")
    # Creating user - DATABASE OPERATION!
    user = User(**user_in.dict())
    db.add(user)
    db.commit()
    return user  # WRONG!

✅ DO: Thin endpoint, call service

@router.post("/users")
async def create_user(
    user_in: UserCreate,
    service: Annotated[UserService, Depends(get_user_service)],
):
    return await service.register_user(user_in)

---

❌ DON'T: Skip type hints

def get_user(user_id):  # What type? What returns?
    return db.query(User).get(user_id)

✅ DO: Always use type hints

def get_user(user_id: int) -> Optional[User]:
    return db.query(User).filter(User.id == user_id).first()

---

❌ DON'T: Ignore errors

try:
    result = risky_operation()
except:
    pass  # Silent failure!

✅ DO: Handle and log errors

try:
    result = risky_operation()
except SpecificError as e:
    logger.error("Operation failed", exc_info=True)
    raise AppException("Failed to process", error_code="OPERATION_FAILED")

---

Summary

The Golden Rules:

1. Four Layers Always:

   • API → Service → Repository → Model

2. Thin Endpoints:

   • Just call service layer (< 10 lines)

3. Business Logic in Services:

   • All orchestration, rules, workflows

4. Type Hints Everywhere:

   • Every function signature

5. Log Everything Important:

   • With context and correlation IDs

6. Test Each Layer:

   • Unit tests for services
   • Integration tests for APIs

7. Security First:

   • Hash passwords, validate inputs, prevent SQL injection

---

This guide covers 90% of daily Python API development. Keep it open while coding! 🚀