python-backend-expert

name	python-backend-expert
description	Python backend expert for Django, FastAPI, Flask, and async Python APIs. Use when building web APIs, working with ORMs (SQLAlchemy, Django ORM), implementing async patterns, or using Pydantic for validation. Triggers: fastapi, django, flask, sqlalchemy, pydantic, async def, APIRouter, Base = declarative_base

Python Expert Skill

Expert-level Python backend patterns for Django, FastAPI, Flask, and async programming.

---

Auto-Detection

This skill activates when:

• Working with Python backend projects
• Detected django, fastapi, or flask in requirements.txt/pyproject.toml
• Working with *.py files in API/backend context
• Using SQLAlchemy, Pydantic, or Django ORM

---

1. FastAPI Patterns

Pydantic Models

# ✅ GOOD - Type-safe validation
from pydantic import BaseModel, EmailStr, Field, field_validator

class UserCreate(BaseModel):
    email: EmailStr
    name: str = Field(..., min_length=2, max_length=100)
    password: str = Field(..., min_length=8)

    @field_validator('name')
    @classmethod
    def name_must_not_be_empty(cls, v: str) -> str:
        if not v.strip():
            raise ValueError('Name cannot be empty')
        return v.strip()

    model_config = {'str_strip_whitespace': True}

Dependency Injection

# ✅ GOOD - Dependencies for shared logic
from typing import Annotated
from fastapi import Depends, HTTPException

async def get_current_user(
    token: str = Depends(oauth2_scheme),
    db: AsyncSession = Depends(get_db)
) -> User:
    user = await verify_token(token, db)
    if user is None:
        raise HTTPException(status_code=401, detail="Invalid token")
    return user

CurrentUser = Annotated[User, Depends(get_current_user)]
DB = Annotated[AsyncSession, Depends(get_db)]

@app.get("/me")
async def get_me(user: CurrentUser, db: DB):
    return user

Background Tasks

# ✅ GOOD - Non-blocking operations
from fastapi import BackgroundTasks

@app.post("/users/")
async def create_user(
    user: UserCreate,
    background_tasks: BackgroundTasks,
    db: AsyncSession = Depends(get_db)
):
    db_user = await crud.create_user(db, user)
    background_tasks.add_task(send_welcome_email, user.email)
    return db_user

Exception Handling

# ✅ GOOD - Custom exceptions
from fastapi import HTTPException
from fastapi.responses import JSONResponse

class AppException(Exception):
    def __init__(self, code: str, message: str, status_code: int = 400):
        self.code = code
        self.message = message
        self.status_code = status_code

@app.exception_handler(AppException)
async def app_exception_handler(request, exc: AppException):
    return JSONResponse(
        status_code=exc.status_code,
        content={"code": exc.code, "message": exc.message}
    )

Lifespan Events

# ✅ GOOD - Startup/shutdown handling
from contextlib import asynccontextmanager

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Startup
    await database.connect()
    yield
    # Shutdown
    await database.disconnect()

app = FastAPI(lifespan=lifespan)

---

2. Django Patterns

Prevent N+1 Queries

# ❌ BAD - N+1 queries
users = User.objects.all()
for user in users:
    print(user.profile.bio)  # 1 query per user!

# ✅ GOOD - select_related for FK
users = User.objects.select_related('profile').all()

# ✅ GOOD - prefetch_related for M2M/reverse FK
posts = Post.objects.prefetch_related('tags', 'comments').all()

Atomic Updates

from django.db.models import F

# ❌ BAD - Race condition
article = Article.objects.get(pk=1)
article.views += 1
article.save()

# ✅ GOOD - Atomic update
Article.objects.filter(pk=1).update(views=F('views') + 1)

Complex Queries

from django.db.models import Q

# ✅ GOOD - OR query
users = User.objects.filter(
    Q(is_staff=True) | Q(is_superuser=True)
)

# ✅ GOOD - Combined conditions
users = User.objects.filter(
    Q(email__endswith='@company.com') &
    (Q(is_active=True) | Q(is_staff=True))
)

Efficient Checks

# ❌ BAD - Loads all objects
if len(User.objects.filter(email=email)) > 0:
    ...

# ✅ GOOD - Efficient existence check
if User.objects.filter(email=email).exists():
    ...

Bulk Operations

# ❌ BAD - N queries
for user_data in users_data:
    User.objects.create(**user_data)

# ✅ GOOD - 1 query
User.objects.bulk_create([
    User(**data) for data in users_data
])

Transactions

from django.db import transaction

@transaction.atomic
def transfer_funds(from_account, to_account, amount):
    from_account.balance = F('balance') - amount
    from_account.save()
    to_account.balance = F('balance') + amount
    to_account.save()

---

3. SQLAlchemy 2.0 Patterns

Modern Declarative Syntax

from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship
from sqlalchemy import ForeignKey

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(primary_key=True)
    email: Mapped[str] = mapped_column(unique=True, index=True)
    name: Mapped[str]
    is_active: Mapped[bool] = mapped_column(default=True)

    posts: Mapped[list["Post"]] = relationship(back_populates="author")

Async Queries

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

async def get_user(db: AsyncSession, user_id: int) -> User | None:
    stmt = select(User).where(User.id == user_id)
    result = await db.execute(stmt)
    return result.scalar_one_or_none()

Eager Loading

from sqlalchemy.orm import selectinload, joinedload

# ✅ GOOD - Prevent N+1
stmt = select(User).options(
    selectinload(User.posts),  # Separate query
    joinedload(User.profile)   # JOIN
)

Pagination

from sqlalchemy import func

async def get_users_paginated(
    db: AsyncSession,
    page: int = 1,
    per_page: int = 20
) -> tuple[list[User], int]:
    # Count total
    count_stmt = select(func.count()).select_from(User)
    total = await db.scalar(count_stmt)

    # Get page
    stmt = select(User).offset((page - 1) * per_page).limit(per_page)
    result = await db.execute(stmt)
    users = result.scalars().all()

    return users, total

---

4. Async Best Practices

Parallel Operations

import asyncio

# ✅ GOOD - Concurrent execution
async def get_dashboard_data(user_id: int) -> dict:
    user, posts, notifications = await asyncio.gather(
        get_user(user_id),
        get_user_posts(user_id),
        get_notifications(user_id)
    )
    return {"user": user, "posts": posts, "notifications": notifications}

Timeout Handling

# ✅ GOOD - Proper timeout
async def fetch_with_timeout(url: str, timeout: float = 5.0):
    async with asyncio.timeout(timeout):
        async with httpx.AsyncClient() as client:
            return await client.get(url)

Rate Limiting

# ✅ GOOD - Semaphore for concurrency control
semaphore = asyncio.Semaphore(10)  # Max 10 concurrent

async def fetch_limited(url: str):
    async with semaphore:
        return await fetch(url)

Task Groups (Python 3.11+)

# ✅ GOOD - Structured concurrency
async def process_all(items: list[Item]):
    async with asyncio.TaskGroup() as tg:
        for item in items:
            tg.create_task(process_item(item))
    # All tasks completed or exception raised

---

5. Type Hints Best Practices

# ✅ GOOD - Modern type hints (Python 3.10+)
def get_users(active: bool | None = None) -> list[User]:
    ...

# ✅ GOOD - TypeVar for generics
from typing import TypeVar

T = TypeVar('T', bound=BaseModel)

async def get_or_404(db: AsyncSession, model: type[T], id: int) -> T:
    obj = await db.get(model, id)
    if obj is None:
        raise HTTPException(status_code=404)
    return obj

# ✅ GOOD - Protocol for structural typing
from typing import Protocol

class Repository(Protocol):
    async def get(self, id: int) -> Model | None: ...
    async def create(self, data: dict) -> Model: ...
    async def delete(self, id: int) -> bool: ...

---

6. Testing Patterns

Pytest Fixtures

import pytest
from httpx import AsyncClient

@pytest.fixture
async def client(app):
    async with AsyncClient(app=app, base_url="http://test") as client:
        yield client

@pytest.fixture
async def db_session():
    async with async_session() as session:
        yield session
        await session.rollback()

Parametrized Tests

@pytest.mark.parametrize("email,valid", [
    ("[e-mail]", True),
    ("invalid", False),
    ("", False),
    ("a@b.c", True),
])
def test_validate_email(email: str, valid: bool):
    result = validate_email(email)
    assert result == valid

Factory Pattern

from factory import Factory, Faker

class UserFactory(Factory):
    class Meta:
        model = User

    email = Faker('email')
    name = Faker('name')
    is_active = True

# Usage
user = UserFactory()
inactive_user = UserFactory(is_active=False)

Mocking

from unittest.mock import AsyncMock, patch

@pytest.mark.asyncio
async def test_create_user_sends_email():
    with patch('app.services.send_email', new_callable=AsyncMock) as mock:
        user = await create_user(UserCreate(email="[e-mail]"))
        mock.assert_called_once()

---

7. Error Handling

Custom Exception Hierarchy

class AppError(Exception):
    """Base application error"""
    def __init__(self, message: str, code: str = "UNKNOWN"):
        self.message = message
        self.code = code
        super().__init__(message)

class NotFoundError(AppError):
    def __init__(self, resource: str, id: int):
        super().__init__(f"{resource} with id {id} not found", "NOT_FOUND")

class ValidationError(AppError):
    def __init__(self, field: str, message: str):
        super().__init__(message, f"VALIDATION_{field.upper()}")

Result Pattern

from dataclasses import dataclass
from typing import Generic, TypeVar

T = TypeVar('T')
E = TypeVar('E')

@dataclass
class Ok(Generic[T]):
    value: T

@dataclass
class Err(Generic[E]):
    error: E

Result = Ok[T] | Err[E]

async def get_user(user_id: int) -> Result[User, str]:
    user = await db.get(User, user_id)
    if user is None:
        return Err("User not found")
    return Ok(user)

# Usage
match await get_user(123):
    case Ok(user):
        print(f"Found: {user.name}")
    case Err(error):
        print(f"Error: {error}")

---

Quick Reference

checklist[12]{pattern,best_practice}:
  FastAPI,Pydantic models + Depends injection
  Django ORM,select_related/prefetch_related
  SQLAlchemy,Mapped types + selectinload
  Async,asyncio.gather for parallel
  Typing,Modern 3.10+ syntax X | None
  Testing,pytest fixtures + parametrize
  Validation,Pydantic or Django Forms
  Errors,Custom exception hierarchy
  Transactions,atomic or db.transaction
  Queries,F() objects for atomic updates
  Bulk,bulk_create over loops
  Cache,Redis or Django cache

---

Version: 1.3.0