Backend Architecture Enforcer

Enforce FastAPI Clean Architecture with BLOCKING validation.

Architecture Overview

+-------------------------------------------------------------------+
|                        ROUTERS LAYER                               |
|  HTTP concerns only: request parsing, response formatting          |
|  Files: router_*.py, routes_*.py, api_*.py                        |
+-------------------------------------------------------------------+
|                        SERVICES LAYER                              |
|  Business logic: orchestration, validation, transformations        |
|  Files: *_service.py                                              |
+-------------------------------------------------------------------+
|                      REPOSITORIES LAYER                            |
|  Data access: database queries, external API calls                 |
|  Files: *_repository.py, *_repo.py                                |
+-------------------------------------------------------------------+
|                        MODELS LAYER                                |
|  Data structures: SQLAlchemy models, Pydantic schemas             |
|  Files: *_model.py (ORM), *_schema.py (Pydantic)                 |
+-------------------------------------------------------------------+

Validation Rules (BLOCKING)

File Naming Conventions

Quick Reference

Layer Separation Summary

Routers (HTTP Only)

• Request parsing and response formatting
• HTTP status codes and auth checks
• Delegate to services via Depends()

Services (Business Logic)

• Validation and orchestration
• Data transformations
• Raise domain exceptions (NOT HTTPException)

Repositories (Data Access)

• Database queries and persistence
• External API calls
• Return domain objects or None

Dependency Injection Quick Reference

# deps.py - Dependency providers
def get_user_repository(
    db: AsyncSession = Depends(get_db),
) -> UserRepository:
    return UserRepository(db)

def get_user_service(
    repo: UserRepository = Depends(get_user_repository),
) -> UserService:
    return UserService(repo)

# router_users.py - Usage
@router.get("/{user_id}")
async def get_user(
    user_id: int,
    service: UserService = Depends(get_user_service),
):
    return await service.get_user(user_id)

Blocked DI Patterns

# BLOCKED - Direct instantiation
service = UserService()

# BLOCKED - Global instance
user_service = UserService()

# BLOCKED - Missing Depends()
async def get_users(db: AsyncSession):  # Missing Depends()

Common Violations

Exception Pattern

# Domain exceptions (services/repositories)
class UserNotFoundError(DomainException):
    def __init__(self, user_id: int):
        super().__init__(f"User {user_id} not found")

# Router converts to HTTP
@router.get("/{user_id}")
async def get_user(user_id: int, service: UserService = Depends(get_user_service)):
    try:
        return await service.get_user(user_id)
    except UserNotFoundError:
        raise HTTPException(404, "User not found")

Async Rules

# GOOD - Async all the way
result = await db.execute(select(User))

# BLOCKED - Sync in async function
result = db.execute(select(User))  # Missing await

# For sync code, use executor
await loop.run_in_executor(None, sync_function)

References

For detailed patterns and examples, see:

Related Skills

• clean-architecture - DDD patterns
• fastapi-advanced - Advanced FastAPI patterns
• dependency-injection - DI patterns
• project-structure-enforcer - Folder structure

Capability Details

layer-separation

Keywords: router, service, repository, layer, clean architecture, separation Solves:

• Prevent database operations in routers
• Block business logic in route handlers
• Ensure proper layer boundaries

dependency-injection

Keywords: depends, dependency injection, DI, fastapi depends, inject Solves:

• Enforce use of FastAPI Depends() pattern
• Block direct instantiation in routers
• Ensure testable code structure

file-naming

Keywords: naming convention, file name, router_, _service, _repository Solves:

• Enforce consistent file naming patterns
• Validate router/service/repository naming
• Maintain codebase consistency

async-patterns

Keywords: async, await, sync, blocking call, asyncio Solves:

• Detect sync calls in async functions
• Prevent blocking operations in async code
• Ensure async consistency

Dependency Injection Patterns

FastAPI dependency injection patterns using Depends() for Clean Architecture.

---

Core Principles

1. Never instantiate services/repositories directly in route handlers 2. Always use `Depends()` for injecting dependencies 3. Chain dependencies for proper layering (router -> service -> repository -> db) 4. Keep dependency providers in a dedicated deps.py file

---

Dependency Provider Pattern

Basic Setup

# app/routers/deps.py
from fastapi import Depends
from sqlalchemy.ext.asyncio import AsyncSession
from app.core.database import get_db
from app.repositories.user_repository import UserRepository
from app.services.user_service import UserService

def get_user_repository(
    db: AsyncSession = Depends(get_db),
) -> UserRepository:
    """Repository depends on database session."""
    return UserRepository(db)

def get_user_service(
    repo: UserRepository = Depends(get_user_repository),
) -> UserService:
    """Service depends on repository."""
    return UserService(repo)

Usage in Router

# app/routers/router_users.py
from fastapi import APIRouter, Depends
from app.services.user_service import UserService
from app.routers.deps import get_user_service

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

@router.get("/{user_id}")
async def get_user(
    user_id: int,
    service: UserService = Depends(get_user_service),
):
    return await service.get_user(user_id)

@router.post("/")
async def create_user(
    user_data: UserCreate,
    service: UserService = Depends(get_user_service),
    current_user: User = Depends(get_current_user),  # Auth dependency
):
    return await service.create_user(user_data)

---

Dependency Chaining

Request
    │
    ▼
┌─────────────────────────────────────────────────┐
│ get_current_user (auth)                          │
│   └── Depends(get_db) for token validation       │
└─────────────────────────────────────────────────┘
    │
    ▼
┌─────────────────────────────────────────────────┐
│ get_user_service                                 │
│   └── Depends(get_user_repository)               │
│         └── Depends(get_db)                      │
└─────────────────────────────────────────────────┘
    │
    ▼
Route Handler

---

Common DI Patterns

1. Database Session Dependency

# app/core/database.py
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine, async_sessionmaker

engine = create_async_engine(DATABASE_URL)
async_session_maker = async_sessionmaker(engine, expire_on_commit=False)

async def get_db() -> AsyncGenerator[AsyncSession, None]:
    async with async_session_maker() as session:
        try:
            yield session
            await session.commit()
        except Exception:
            await session.rollback()
            raise

2. Authentication Dependency

# app/routers/deps.py
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from jose import jwt, JWTError

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="auth/token")

async def get_current_user(
    token: str = Depends(oauth2_scheme),
    user_service: UserService = Depends(get_user_service),
) -> User:
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        user_id: int = payload.get("sub")
        if user_id is None:
            raise HTTPException(status_code=401, detail="Invalid token")
    except JWTError:
        raise HTTPException(status_code=401, detail="Invalid token")

    user = await user_service.get_user(user_id)
    if user is None:
        raise HTTPException(status_code=401, detail="User not found")
    return user

async def get_current_active_user(
    current_user: User = Depends(get_current_user),
) -> User:
    if not current_user.is_active:
        raise HTTPException(status_code=400, detail="Inactive user")
    return current_user

3. Permission Dependency

# app/routers/deps.py
from typing import Callable

def require_permissions(*permissions: str) -> Callable:
    """Factory for permission-checking dependencies."""
    async def permission_checker(
        current_user: User = Depends(get_current_active_user),
    ) -> User:
        user_permissions = set(current_user.permissions)
        required = set(permissions)
        if not required.issubset(user_permissions):
            raise HTTPException(
                status_code=403,
                detail="Insufficient permissions"
            )
        return current_user
    return permission_checker

# Usage
@router.delete("/{user_id}")
async def delete_user(
    user_id: int,
    current_user: User = Depends(require_permissions("admin", "user:delete")),
    service: UserService = Depends(get_user_service),
):
    return await service.delete_user(user_id)

4. Pagination Dependency

# app/routers/deps.py
from pydantic import BaseModel

class PaginationParams(BaseModel):
    skip: int = 0
    limit: int = 100

def get_pagination(
    skip: int = 0,
    limit: int = 100,
) -> PaginationParams:
    return PaginationParams(skip=skip, limit=min(limit, 100))

# Usage
@router.get("/")
async def list_users(
    pagination: PaginationParams = Depends(get_pagination),
    service: UserService = Depends(get_user_service),
):
    return await service.list_users(
        skip=pagination.skip,
        limit=pagination.limit
    )

---

Blocked Patterns

1. Direct Instantiation

# BLOCKED
@router.get("/{user_id}")
async def get_user(user_id: int):
    service = UserService()  # Direct instantiation
    return await service.get_user(user_id)

2. Global Instance

# BLOCKED
user_service = UserService()  # Global instance

@router.get("/{user_id}")
async def get_user(user_id: int):
    return await user_service.get_user(user_id)

3. Missing Depends()

# BLOCKED
@router.get("/users")
async def get_users(db: AsyncSession):  # Missing Depends()
    return await db.execute(select(User)).scalars().all()

4. Instantiation Inside Handler

# BLOCKED
@router.get("/{user_id}")
async def get_user(
    user_id: int,
    db: AsyncSession = Depends(get_db),
):
    repo = UserRepository(db)      # Instantiation in handler
    service = UserService(repo)    # Should use Depends()
    return await service.get_user(user_id)

---

Testing with DI

Override Dependencies in Tests

# tests/conftest.py
import pytest
from fastapi.testclient import TestClient
from app.main import app
from app.routers.deps import get_db, get_user_service

# Mock database session
@pytest.fixture
def mock_db():
    return AsyncMock(spec=AsyncSession)

# Mock service
@pytest.fixture
def mock_user_service():
    service = Mock(spec=UserService)
    service.get_user = AsyncMock(return_value=User(id=1, email="test@test.com"))
    return service

@pytest.fixture
def client(mock_db, mock_user_service):
    app.dependency_overrides[get_db] = lambda: mock_db
    app.dependency_overrides[get_user_service] = lambda: mock_user_service
    yield TestClient(app)
    app.dependency_overrides.clear()

Test with Dependency Overrides

# tests/test_routers/test_users.py
def test_get_user(client, mock_user_service):
    response = client.get("/users/1")
    assert response.status_code == 200
    mock_user_service.get_user.assert_called_once_with(1)

---

Best Practices

Layer Separation Rules

Detailed rules for Router-Service-Repository layer separation in FastAPI Clean Architecture.

---

Routers Layer (HTTP Only)

Routers should ONLY handle:

• Request parsing and validation
• Response formatting
• HTTP status codes
• Authentication/authorization checks
• Calling services

# GOOD - Router delegates to service
@router.post("/users", response_model=UserResponse)
async def create_user(
    user_data: UserCreate,
    service: UserService = Depends(get_user_service),
):
    user = await service.create_user(user_data)
    return user

# BLOCKED - Business logic in router
@router.post("/users")
async def create_user(
    user_data: UserCreate,
    db: AsyncSession = Depends(get_db),
):
    # Database operation in router
    existing = await db.execute(
        select(User).where(User.email == user_data.email)
    )
    if existing.scalar():
        raise HTTPException(400, "Email exists")

    # Business logic in router
    user = User(**user_data.dict())
    user.created_at = datetime.now(timezone.utc)
    db.add(user)
    await db.commit()
    return user

---

Services Layer (Business Logic)

Services should:

• Contain business logic and validation
• Orchestrate repositories
• Transform data between layers
• Raise domain exceptions (NOT HTTPException)

# GOOD - Service with business logic
class UserService:
    def __init__(self, repo: UserRepository):
        self.repo = repo

    async def create_user(self, data: UserCreate) -> User:
        if await self.repo.exists_by_email(data.email):
            raise UserAlreadyExistsError(data.email)

        user = User(
            email=data.email,
            password_hash=hash_password(data.password),
            created_at=datetime.now(timezone.utc),
        )
        return await self.repo.create(user)

# BLOCKED - HTTP concerns in service
class UserService:
    async def create_user(self, data: UserCreate) -> User:
        if await self.repo.exists_by_email(data.email):
            # HTTPException in service - BLOCKED
            raise HTTPException(400, "Email already exists")

---

Repositories Layer (Data Access)

Repositories should:

• Execute database queries
• Call external APIs
• Handle data persistence
• Return domain objects or None

# GOOD - Repository handles data access only
class UserRepository:
    def __init__(self, db: AsyncSession):
        self.db = db

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

    async def create(self, user: User) -> User:
        self.db.add(user)
        await self.db.commit()
        await self.db.refresh(user)
        return user

# BLOCKED - HTTP concerns in repository
class UserRepository:
    async def get_by_id(self, user_id: int) -> User:
        user = await self.db.get(User, user_id)
        if not user:
            # HTTPException in repository - BLOCKED
            raise HTTPException(404, "User not found")
        return user

---

Exception Handling Pattern

Domain Exceptions

# app/core/exceptions.py
class DomainException(Exception):
    """Base domain exception."""
    pass

class UserNotFoundError(DomainException):
    def __init__(self, user_id: int):
        self.user_id = user_id
        super().__init__(f"User {user_id} not found")

class UserAlreadyExistsError(DomainException):
    def __init__(self, email: str):
        self.email = email
        super().__init__(f"User with email {email} already exists")

Router Exception Handler

# app/routers/deps.py
def handle_domain_exception(exc: DomainException) -> HTTPException:
    """Convert domain exceptions to HTTP responses."""
    if isinstance(exc, UserNotFoundError):
        return HTTPException(404, str(exc))
    if isinstance(exc, UserAlreadyExistsError):
        return HTTPException(409, str(exc))
    return HTTPException(500, "Internal error")

# Usage in router
@router.get("/users/{user_id}")
async def get_user(
    user_id: int,
    service: UserService = Depends(get_user_service),
):
    try:
        return await service.get_user(user_id)
    except DomainException as e:
        raise handle_domain_exception(e)

---

Async Consistency Rules

No Sync Calls in Async Functions

# GOOD - Async all the way
async def get_user(user_id: int) -> User:
    result = await db.execute(select(User).where(User.id == user_id))
    return result.scalar_one_or_none()

# BLOCKED - Sync call in async function
async def get_user(user_id: int) -> User:
    # Missing await - blocks event loop
    result = db.execute(select(User).where(User.id == user_id))
    return result.scalar_one_or_none()

# For unavoidable sync code, use run_in_executor
async def process_file(file_path: str) -> bytes:
    loop = asyncio.get_event_loop()
    return await loop.run_in_executor(
        None,
        lambda: open(file_path, 'rb').read()
    )

---

Layer Boundaries Summary

Backend Architecture Violations

Reference guide for common Clean Architecture violations in FastAPI applications.

---

1. Database Operations in Routers

Proper Pattern

# app/routers/router_users.py
from fastapi import APIRouter, Depends
from app.services.user_service import UserService
from app.routers.deps import get_user_service

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

@router.post("/", response_model=UserResponse, status_code=201)
async def create_user(
    user_data: UserCreate,
    service: UserService = Depends(get_user_service),
):
    """Router delegates ALL data operations to service layer."""
    return await service.create_user(user_data)

Anti-Pattern (VIOLATION)

# app/routers/router_users.py
from fastapi import APIRouter, Depends
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select
from app.models.user_model import User
from app.core.database import get_db

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

@router.post("/", response_model=UserResponse, status_code=201)
async def create_user(
    user_data: UserCreate,
    db: AsyncSession = Depends(get_db),  # VIOLATION: DB session in router
):
    # VIOLATION: Database query in router
    existing = await db.execute(
        select(User).where(User.email == user_data.email)
    )
    if existing.scalar():
        raise HTTPException(400, "Email already exists")

    # VIOLATION: Direct ORM operations in router
    user = User(**user_data.model_dump())
    db.add(user)
    await db.commit()
    await db.refresh(user)
    return user

Why It Matters

• Testability: Routers with database access require full database setup for testing
• Separation of Concerns: Routers should only handle HTTP request/response mapping
• Reusability: Business logic locked in routers cannot be reused by other services
• Maintainability: Changes to data access require modifying HTTP layer

Auto-Fix Suggestion

1. Extract database operations to user_repository.py 2. Create user_service.py to orchestrate repository calls 3. Inject service via Depends(get_user_service) 4. Router should only call service.create_user(user_data)

---

2. HTTPException in Service Layer

Proper Pattern

# app/services/user_service.py
from app.core.exceptions import UserNotFoundError, UserAlreadyExistsError

class UserService:
    def __init__(self, repo: UserRepository):
        self.repo = repo

    async def create_user(self, data: UserCreate) -> User:
        # Raise DOMAIN exceptions, not HTTP exceptions
        if await self.repo.exists_by_email(data.email):
            raise UserAlreadyExistsError(data.email)

        user = User(
            email=data.email,
            password_hash=hash_password(data.password),
        )
        return await self.repo.create(user)

    async def get_user(self, user_id: int) -> User:
        user = await self.repo.get_by_id(user_id)
        if not user:
            raise UserNotFoundError(user_id)
        return user

Anti-Pattern (VIOLATION)

# app/services/user_service.py
from fastapi import HTTPException  # VIOLATION: HTTP import in service

class UserService:
    def __init__(self, repo: UserRepository):
        self.repo = repo

    async def create_user(self, data: UserCreate) -> User:
        if await self.repo.exists_by_email(data.email):
            # VIOLATION: HTTPException in service layer
            raise HTTPException(
                status_code=400,
                detail="Email already exists"
            )

        return await self.repo.create(User(**data.model_dump()))

    async def get_user(self, user_id: int) -> User:
        user = await self.repo.get_by_id(user_id)
        if not user:
            # VIOLATION: HTTP status codes in business logic
            raise HTTPException(status_code=404, detail="User not found")
        return user

Why It Matters

• Framework Coupling: Services become tied to FastAPI, cannot be reused in CLI/workers
• Layer Bleeding: HTTP concerns (status codes) leak into business layer
• Testing Complexity: Tests must handle HTTP exceptions instead of domain exceptions
• Protocol Independence: Business logic should work with any transport (HTTP, gRPC, CLI)

Auto-Fix Suggestion

1. Create domain exceptions in app/core/exceptions.py:

   class UserNotFoundError(DomainException): ...
   class UserAlreadyExistsError(DomainException): ...

2. Replace HTTPException with domain exceptions in services 3. Add exception handler in router that converts domain exceptions to HTTP responses

---

3. Direct Instantiation (Missing Depends)

Proper Pattern

# app/routers/deps.py
from fastapi import Depends
from sqlalchemy.ext.asyncio import AsyncSession
from app.core.database import get_db
from app.repositories.user_repository import UserRepository
from app.services.user_service import UserService

def get_user_repository(
    db: AsyncSession = Depends(get_db),
) -> UserRepository:
    return UserRepository(db)

def get_user_service(
    repo: UserRepository = Depends(get_user_repository),
) -> UserService:
    return UserService(repo)

# app/routers/router_users.py
@router.get("/{user_id}")
async def get_user(
    user_id: int,
    service: UserService = Depends(get_user_service),  # Proper DI
):
    return await service.get_user(user_id)

Anti-Pattern (VIOLATION)

# app/routers/router_users.py

# VIOLATION: Global instance (no DI)
user_service = UserService(UserRepository())

@router.get("/{user_id}")
async def get_user(user_id: int):
    # VIOLATION: Using global instance
    return await user_service.get_user(user_id)

# --- OR ---

@router.get("/{user_id}")
async def get_user(user_id: int):
    # VIOLATION: Direct instantiation in route handler
    repo = UserRepository(get_db())
    service = UserService(repo)
    return await service.get_user(user_id)

# --- OR ---

@router.get("/{user_id}")
async def get_user(
    user_id: int,
    db: AsyncSession,  # VIOLATION: Missing Depends()
):
    repo = UserRepository(db)
    service = UserService(repo)
    return await service.get_user(user_id)

Why It Matters

• Testing: Cannot easily mock dependencies without DI
• Lifecycle Management: FastAPI cannot manage object lifecycles (sessions, connections)
• Request Scope: Global instances share state across requests (thread-safety issues)
• Configuration: Cannot configure different instances per environment

Auto-Fix Suggestion

1. Create dependency providers in app/routers/deps.py 2. Use Depends() for ALL service/repository injections 3. Chain dependencies: get_user_service depends on get_user_repository 4. Never instantiate services directly in route handlers

---

4. Wrong File Naming Convention

Proper Pattern

app/
├── routers/
│   ├── router_users.py      # router_ prefix
│   ├── router_auth.py
│   ├── routes_orders.py     # routes_ prefix also valid
│   ├── api_v1.py            # api_ prefix for versioned
│   └── deps.py              # deps/dependencies allowed
├── services/
│   ├── user_service.py      # _service suffix
│   ├── auth_service.py
│   └── email_service.py
├── repositories/
│   ├── user_repository.py   # _repository suffix
│   ├── user_repo.py         # _repo suffix also valid
│   └── base_repository.py
├── schemas/
│   ├── user_schema.py       # _schema suffix
│   ├── user_dto.py          # _dto suffix also valid
│   ├── user_request.py      # _request suffix
│   └── user_response.py     # _response suffix
└── models/
    ├── user_model.py        # _model suffix
    ├── user_entity.py       # _entity suffix also valid
    └── base.py              # base.py allowed

Anti-Pattern (VIOLATION)

app/
├── routers/
│   ├── users.py             # VIOLATION: Missing router_ prefix
│   ├── UserRouter.py        # VIOLATION: PascalCase
│   └── user_routes.py       # VIOLATION: Wrong format (should be routes_user.py)
├── services/
│   ├── users.py             # VIOLATION: Missing _service suffix
│   ├── UserService.py       # VIOLATION: PascalCase
│   └── service_user.py      # VIOLATION: Wrong order
├── repositories/
│   ├── users.py             # VIOLATION: Missing _repository suffix
│   └── repository_user.py   # VIOLATION: Wrong order
├── schemas/
│   ├── users.py             # VIOLATION: Missing _schema suffix
│   └── UserSchema.py        # VIOLATION: PascalCase
└── models/
    ├── users.py             # VIOLATION: Missing _model suffix
    └── UserModel.py         # VIOLATION: PascalCase

Why It Matters

• Discoverability: Consistent naming helps developers find files quickly
• Automation: Scripts and tools can identify file types from naming patterns
• Onboarding: New team members understand file purposes immediately
• Import Clarity: Import statements clearly indicate what is being imported

Auto-Fix Suggestion

---

5. Sync Calls in Async Functions

Proper Pattern

# app/repositories/user_repository.py
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select

class UserRepository:
    def __init__(self, db: AsyncSession):
        self.db = db

    async def get_by_id(self, user_id: int) -> User | None:
        # CORRECT: Using await with async session
        result = await self.db.execute(
            select(User).where(User.id == user_id)
        )
        return result.scalar_one_or_none()

    async def get_all(self) -> list[User]:
        # CORRECT: Async all the way
        result = await self.db.execute(select(User))
        return list(result.scalars().all())

# For unavoidable sync operations
import asyncio

async def process_file(file_path: str) -> bytes:
    # CORRECT: Run sync code in executor
    loop = asyncio.get_event_loop()
    return await loop.run_in_executor(
        None,
        lambda: open(file_path, 'rb').read()
    )

Anti-Pattern (VIOLATION)

# app/repositories/user_repository.py
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select

class UserRepository:
    def __init__(self, db: AsyncSession):
        self.db = db

    async def get_by_id(self, user_id: int) -> User | None:
        # VIOLATION: Missing await - sync call in async function
        result = self.db.execute(
            select(User).where(User.id == user_id)
        )
        return result.scalar_one_or_none()

    async def get_all(self) -> list[User]:
        # VIOLATION: Using sync session methods
        return self.db.query(User).all()

# VIOLATION: Blocking I/O in async function
async def process_file(file_path: str) -> bytes:
    # This blocks the event loop!
    with open(file_path, 'rb') as f:
        return f.read()

# VIOLATION: Sync HTTP call in async function
import requests

async def fetch_external_data(url: str) -> dict:
    # Blocks event loop - use httpx or aiohttp instead
    response = requests.get(url)
    return response.json()

Why It Matters

• Event Loop Blocking: Sync calls block the entire event loop, killing concurrency
• Performance: Defeats the purpose of using async FastAPI
• Scalability: Blocked event loop cannot handle other requests
• Timeout Issues: Long sync operations can cause request timeouts

Auto-Fix Suggestion

1. Replace db.execute() with await db.execute() 2. Use await db.scalars() instead of db.query() 3. Replace requests with httpx.AsyncClient or aiohttp 4. Wrap unavoidable sync operations in run_in_executor() 5. Use aiofiles for async file operations

---

Quick Reference: Common Violations