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](#project-setup)
2. [Project Structure](#project-structure)
3. [The Four Layers](#the-four-layers)
4. [Code Standards](#code-standards)
5. [Module Size Guidelines](#module-size-guidelines)
6. [Database Patterns](#database-patterns)
7. [Error Handling](#error-handling)
8. [Logging](#logging)
9. [Testing](#testing)
10. [Security Essentials](#security-essentials)
11. [Daily Checklist](#daily-checklist)

---

## Project Setup

### Initial Setup
```bash
# 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`
```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`
```bash
# 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.

```python
# 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.

```python
# 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.

```python
# 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.

```python
# 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

```python
# 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)
```python
# 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
```python
# 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
```python
# 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

```python
# 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

```python
# 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
```python
# 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
```python
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
```python
# 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
```python
# 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)
```python
# 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)
```python
# 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
```python
# 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
```python
# 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
```python
# 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)
```python
# 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
```python
# 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
```python
# 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
```python
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
```python
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
```python
# 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

```bash
# 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
```python
@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
```python
@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
```python
def get_user(user_id):  # What type? What returns?
    return db.query(User).get(user_id)
```

### ✅ DO: Always use type hints
```python
def get_user(user_id: int) -> Optional[User]:
    return db.query(User).filter(User.id == user_id).first()
```

---

### ❌ DON'T: Ignore errors
```python
try:
    result = risky_operation()
except:
    pass  # Silent failure!
```

### ✅ DO: Handle and log errors
```python
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!** 🚀