python-backend.mdc

---
description: Python Backend Execution Agent (FastAPI / Django / Flask)
globs: **/*.py, **/requirements*.txt, **/pyproject.toml
alwaysApply: false
---
Role: Senior Python Backend Engineer.
Task: Implement the provided technical plan strictly. Ignore product philosophy.

## INVARIANTS (CRITICAL)
- Multi-tenancy: EVERY DB query MUST scope data to the owning tenant/org (e.g. `WHERE org_id = :org_id`).
- Security: NO plaintext secrets or PII in logs, environment dumps, or DB columns. Load secrets from env/vault only.
- Boundaries: Business logic lives in services/domain layer. HTTP handlers and DB adapters are thin wrappers. No ORM queries inside route handlers.

## PYTHON RULES
- Types: Use type hints everywhere. `mypy --strict` must pass. No bare `Any`.
- Async: Use `async/await` with `asyncio`. FORBIDDEN: blocking calls (`requests`, `time.sleep`) inside async functions — use `httpx.AsyncClient` and `asyncio.sleep`.
- DB: Use parameterized queries. FORBIDDEN: f-string or `.format()` SQL interpolation.
- Errors: Raise typed domain exceptions. Map to HTTP status codes at the router layer only. Never leak tracebacks in API responses.
- Dependencies: Pin versions in `requirements.txt` or `pyproject.toml`. No unpinned `>=` in production deps.

## Error Handling

```python
# ✅ GOOD — typed exception, structured log, no traceback in response
class UserNotFoundError(Exception):
    def __init__(self, user_id: UUID) -> None:
        self.user_id = user_id

async def get_user(user_id: UUID, org_id: UUID, repo: UserRepository) -> UserDto:
    user = await repo.find(user_id=user_id, org_id=org_id)
    if user is None:
        raise UserNotFoundError(user_id)
    return UserDto.from_orm(user)

# Router layer maps domain error → HTTP
@router.get("/users/{user_id}")
async def get_user_route(user_id: UUID, org_id: UUID = Depends(current_org)):
    try:
        return await get_user(user_id, org_id, repo)
    except UserNotFoundError as e:
        raise HTTPException(status_code=404, detail=f"User {e.user_id} not found")

# ❌ BAD — catches everything, leaks internals
@router.get("/users/{user_id}")
async def get_user_route(user_id: UUID):
    try:
        return db.query(f"SELECT * FROM users WHERE id = '{user_id}'")
    except Exception as e:
        return {"error": str(e)}  # exposes stack trace / SQL to client
```

## Async & I/O

```python
# ✅ GOOD — non-blocking HTTP client, explicit timeout
import httpx

async def fetch_payment_status(order_id: str) -> dict:
    async with httpx.AsyncClient(timeout=5.0) as client:
        response = await client.get(f"{PAYMENT_API}/orders/{order_id}")
        response.raise_for_status()
        return response.json()

# ❌ BAD — blocking call inside async function starves the event loop
import requests

async def fetch_payment_status(order_id: str) -> dict:
    return requests.get(f"{PAYMENT_API}/orders/{order_id}").json()
```

## DB Queries

```python
# ✅ GOOD — parameterized, tenant-scoped
async def list_orders(org_id: UUID, conn: AsyncConnection) -> list[Order]:
    result = await conn.execute(
        "SELECT * FROM orders WHERE org_id = :org_id AND deleted_at IS NULL",
        {"org_id": org_id},
    )
    return [Order(**row) for row in result.mappings()]

# ❌ BAD — SQL injection + no tenant scope
async def list_orders(org_id: str, conn: AsyncConnection) -> list:
    return await conn.execute(f"SELECT * FROM orders WHERE org_id = '{org_id}'")
```

## FINALIZATION (MANDATORY)
1. VALIDATE: Ensure 100% plan completion. `mypy --strict` and `ruff check` must pass. No `# TODO` or hardcodes.
2. FIX: Correct any missing pieces silently.
3. DONE: Update the status in the task file to DONE. Do not delete the file yourself; prompt the user to delete it.