ai-development-scaffold/.agents/skills/languages/python/SKILL.md

name, description

name	description
python-fastapi	Python development with FastAPI, pytest, Ruff, Pydantic v2, and SQLAlchemy async patterns. Use when writing Python code, tests, or APIs.

Python Development Skill

Project Structure

src/
├── backend/
│   ├── __init__.py
│   ├── main.py              # FastAPI app entry
│   ├── config.py            # Pydantic Settings
│   ├── domains/
│   │   └── users/
│   │       ├── __init__.py
│   │       ├── router.py    # API routes
│   │       ├── service.py   # Business logic
│   │       ├── models.py    # SQLAlchemy models
│   │       └── schemas.py   # Pydantic schemas
│   └── shared/
│       ├── database.py      # DB session management
│       └── models/          # Base models
tests/
├── backend/
│   ├── conftest.py          # Shared fixtures
│   └── domains/
│       └── users/
│           └── test_service.py

Testing with pytest

Configuration (pyproject.toml)

[tool.pytest.ini_options]
testpaths = ["tests"]
asyncio_mode = "auto"
asyncio_default_fixture_loop_scope = "function"
addopts = [
    "-v",
    "--strict-markers",
    "--cov=src",
    "--cov-report=term-missing",
    "--cov-fail-under=80"
]

[tool.coverage.run]
branch = true
source = ["src"]
omit = ["*/__init__.py", "*/migrations/*"]

Async Test Fixtures

# tests/conftest.py
import pytest
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.orm import sessionmaker

from src.backend.shared.models.base import Base


@pytest.fixture
async def db_session():
    """Create in-memory SQLite for fast tests."""
    engine = create_async_engine(
        "sqlite+aiosqlite:///:memory:",
        echo=False,
    )

    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)

    async_session = sessionmaker(
        engine, class_=AsyncSession, expire_on_commit=False
    )

    async with async_session() as session:
        yield session
        await session.rollback()

    await engine.dispose()

Factory Functions for Test Data

# tests/factories.py
from typing import Any

from src.backend.domains.users.schemas import UserCreate


def get_mock_user_create(overrides: dict[str, Any] | None = None) -> UserCreate:
    """Factory for UserCreate with sensible defaults."""
    defaults = {
        "email": "[email protected]",
        "name": "Test User",
        "password": "securepassword123",
    }
    return UserCreate(**(defaults | (overrides or {})))


def get_mock_user_dict(overrides: dict[str, Any] | None = None) -> dict[str, Any]:
    """Factory for raw user dict."""
    defaults = {
        "id": "user_123",
        "email": "[email protected]",
        "name": "Test User",
        "is_active": True,
    }
    return defaults | (overrides or {})

Test Patterns

# tests/backend/domains/users/test_service.py
import pytest
from unittest.mock import AsyncMock

from src.backend.domains.users.service import UserService
from tests.factories import get_mock_user_create


class TestUserService:
    """Test user service behavior through public API."""

    async def test_create_user_returns_user_with_id(self, db_session):
        """Creating a user should return user with generated ID."""
        service = UserService(db_session)
        user_data = get_mock_user_create()

        result = await service.create_user(user_data)

        assert result.id is not None
        assert result.email == user_data.email

    async def test_create_user_hashes_password(self, db_session):
        """Password should be hashed, not stored in plain text."""
        service = UserService(db_session)
        user_data = get_mock_user_create({"password": "mypassword"})

        result = await service.create_user(user_data)

        assert result.hashed_password != "mypassword"
        assert len(result.hashed_password) > 50  # Hashed

    async def test_create_duplicate_email_raises_error(self, db_session):
        """Duplicate email should raise ValueError."""
        service = UserService(db_session)
        user_data = get_mock_user_create()
        await service.create_user(user_data)

        with pytest.raises(ValueError, match="Email already exists"):
            await service.create_user(user_data)

    async def test_get_user_not_found_returns_none(self, db_session):
        """Non-existent user should return None."""
        service = UserService(db_session)

        result = await service.get_user("nonexistent_id")

        assert result is None

Pydantic v2 Patterns

Schemas with Validation

# src/backend/domains/users/schemas.py
from datetime import datetime
from pydantic import BaseModel, EmailStr, Field, field_validator


class UserBase(BaseModel):
    """Base user schema with shared fields."""

    email: EmailStr
    name: str = Field(..., min_length=1, max_length=100)


class UserCreate(UserBase):
    """Schema for creating a user."""

    password: str = Field(..., min_length=8)

    @field_validator("password")
    @classmethod
    def password_must_be_strong(cls, v: str) -> str:
        if not any(c.isupper() for c in v):
            raise ValueError("Password must contain uppercase letter")
        if not any(c.isdigit() for c in v):
            raise ValueError("Password must contain a digit")
        return v


class UserResponse(UserBase):
    """Schema for user responses (no password)."""

    id: str
    is_active: bool
    created_at: datetime

    model_config = {"from_attributes": True}

Configuration with Pydantic Settings

# src/backend/config.py
from functools import lru_cache
from pydantic_settings import BaseSettings, SettingsConfigDict


class Settings(BaseSettings):
    """Application settings from environment."""

    model_config = SettingsConfigDict(
        env_file=".env",
        env_file_encoding="utf-8",
        case_sensitive=False,
    )

    # Database
    database_url: str
    database_pool_size: int = 5

    # Redis
    redis_url: str = "redis://localhost:6379"

    # Security
    secret_key: str
    access_token_expire_minutes: int = 30

    # AWS
    aws_region: str = "eu-west-2"


@lru_cache
def get_settings() -> Settings:
    return Settings()

FastAPI Patterns

Router with Dependency Injection

# src/backend/domains/users/router.py
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.ext.asyncio import AsyncSession

from src.backend.shared.database import get_db_session
from .schemas import UserCreate, UserResponse
from .service import UserService

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


def get_user_service(
    session: AsyncSession = Depends(get_db_session),
) -> UserService:
    return UserService(session)


@router.post("/", response_model=UserResponse, status_code=status.HTTP_201_CREATED)
async def create_user(
    user_data: UserCreate,
    service: UserService = Depends(get_user_service),
) -> UserResponse:
    """Create a new user."""
    try:
        return await service.create_user(user_data)
    except ValueError as e:
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(e))


@router.get("/{user_id}", response_model=UserResponse)
async def get_user(
    user_id: str,
    service: UserService = Depends(get_user_service),
) -> UserResponse:
    """Get user by ID."""
    user = await service.get_user(user_id)
    if not user:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="User not found")
    return user

SQLAlchemy 2.0 Async Patterns

Model Definition

# src/backend/domains/users/models.py
from datetime import datetime
from sqlalchemy import String, Boolean, DateTime
from sqlalchemy.orm import Mapped, mapped_column

from src.backend.shared.models.base import Base


class User(Base):
    __tablename__ = "users"

    id: Mapped[str] = mapped_column(String(36), primary_key=True)
    email: Mapped[str] = mapped_column(String(255), unique=True, index=True)
    name: Mapped[str] = mapped_column(String(100))
    hashed_password: Mapped[str] = mapped_column(String(255))
    is_active: Mapped[bool] = mapped_column(Boolean, default=True)
    created_at: Mapped[datetime] = mapped_column(
        DateTime, default=datetime.utcnow
    )

Async Repository Pattern

# src/backend/domains/users/repository.py
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from .models import User


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

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

    async def get_by_email(self, email: str) -> User | None:
        result = await self._session.execute(
            select(User).where(User.email == email)
        )
        return result.scalar_one_or_none()

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

Ruff Configuration

# pyproject.toml
[tool.ruff]
line-length = 120
target-version = "py311"

[tool.ruff.lint]
select = [
    "E",      # pycodestyle errors
    "F",      # pyflakes
    "I",      # isort
    "UP",     # pyupgrade
    "B",      # flake8-bugbear
    "SIM",    # flake8-simplify
    "ASYNC",  # flake8-async
]
ignore = [
    "B008",   # function-call-in-default-argument (FastAPI Depends)
    "E501",   # line-too-long (handled by formatter)
]

[tool.ruff.lint.per-file-ignores]
"__init__.py" = ["F401"]  # unused imports (re-exports)
"tests/*" = ["S101"]      # assert allowed in tests

Commands

# Testing
pytest                                    # Run all tests
pytest tests/backend/domains/users/       # Run specific directory
pytest -k "test_create"                   # Run tests matching pattern
pytest --cov=src --cov-report=html        # Coverage with HTML report
pytest -x                                 # Stop on first failure
pytest --lf                               # Run last failed tests

# Linting & Formatting
ruff check .                              # Lint check
ruff check . --fix                        # Auto-fix issues
ruff format .                             # Format code
ruff format . --check                     # Check formatting

# Type Checking
mypy src                                  # Full type check
pyright src                               # Alternative type checker

# Development
uvicorn src.backend.main:app --reload     # Dev server
alembic revision --autogenerate -m "msg"  # Create migration
alembic upgrade head                      # Apply migrations

Anti-Patterns to Avoid

# BAD: Mutable default argument
def process_items(items: list = []):  # Creates shared state!
    items.append("new")
    return items

# GOOD: Use None default
def process_items(items: list | None = None) -> list:
    if items is None:
        items = []
    return [*items, "new"]


# BAD: Bare except
try:
    risky_operation()
except:  # Catches everything including SystemExit!
    pass

# GOOD: Specific exceptions
try:
    risky_operation()
except ValueError as e:
    logger.error(f"Validation failed: {e}")
    raise


# BAD: String formatting in SQL (injection risk!)
query = f"SELECT * FROM users WHERE email = '{email}'"

# GOOD: Parameterized query
stmt = select(User).where(User.email == email)


# BAD: Sync in async context
async def get_data():
    return requests.get(url)  # Blocks event loop!

# GOOD: Use async client
async def get_data():
    async with httpx.AsyncClient() as client:
        return await client.get(url)