FastAPI: API на Python с валидацией и документацией

mkdir leads-api
cd leads-api
python -m venv .venv
source .venv/bin/activate          # Windows PowerShell: .venv\Scripts\Activate.ps1
python -m pip install "fastapi[standard]" pytest httpx2
mkdir app tests                    # и пустой файл app/__init__.py

[tool.pytest]
pythonpath = ["."]
testpaths = ["tests"]

import os
import secrets
from typing import Annotated, Literal

from fastapi import Depends, FastAPI, Header, HTTPException, Query, status
from pydantic import BaseModel, Field

app = FastAPI(title="Leads API", version="1.0.0")

Service = Literal["website", "mobile-app", "crm"]
LeadStatus = Literal["new", "in_progress", "done"]


class LeadIn(BaseModel):
    name: Annotated[str, Field(min_length=2, max_length=80)]
    phone: Annotated[str, Field(pattern=r"^\+?\d{10,15}$", examples=["+77011234567"])]
    service: Service
    comment: Annotated[str | None, Field(max_length=500)] = None


class Lead(LeadIn):
    id: int
    status: LeadStatus = "new"


class LeadStatusUpdate(BaseModel):
    status: LeadStatus

class LeadStore:
    """In-memory storage for the lesson and tests. A real project uses a database."""

    def __init__(self) -> None:
        self._items: dict[int, Lead] = {}
        self._next_id = 1

    def add(self, data: LeadIn) -> Lead:
        lead = Lead(id=self._next_id, **data.model_dump())
        self._items[lead.id] = lead
        self._next_id += 1
        return lead

    def list(self, status: LeadStatus | None, limit: int) -> list[Lead]:
        leads = [lead for lead in self._items.values() if status is None or lead.status == status]
        return leads[:limit]

    def get(self, lead_id: int) -> Lead | None:
        return self._items.get(lead_id)

    def save(self, lead: Lead) -> None:
        self._items[lead.id] = lead

    def delete(self, lead_id: int) -> bool:
        return self._items.pop(lead_id, None) is not None


store = LeadStore()


def get_store() -> LeadStore:
    return store


def require_api_key(x_api_key: Annotated[str | None, Header()] = None) -> None:
    expected = os.environ.get("LEADS_API_KEY")
    if not expected or x_api_key is None or not secrets.compare_digest(x_api_key, expected):
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid or missing API key")


StoreDep = Annotated[LeadStore, Depends(get_store)]
admin_only = [Depends(require_api_key)]


def find_lead(lead_id: int, leads: LeadStore) -> Lead:
    lead = leads.get(lead_id)
    if lead is None:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Lead not found")
    return lead

@app.get("/health")
def health() -> dict[str, str]:
    return {"status": "ok"}


@app.post("/leads", status_code=status.HTTP_201_CREATED)
def create_lead(data: LeadIn, leads: StoreDep) -> Lead:
    return leads.add(data)


@app.get("/leads", dependencies=admin_only)
def list_leads(
    leads: StoreDep,
    status_filter: Annotated[LeadStatus | None, Query(alias="status")] = None,
    limit: Annotated[int, Query(ge=1, le=100)] = 50,
) -> list[Lead]:
    return leads.list(status_filter, limit)


@app.get("/leads/{lead_id}", dependencies=admin_only)
def read_lead(lead_id: int, leads: StoreDep) -> Lead:
    return find_lead(lead_id, leads)


@app.patch("/leads/{lead_id}", dependencies=admin_only)
def update_lead_status(lead_id: int, update: LeadStatusUpdate, leads: StoreDep) -> Lead:
    lead = find_lead(lead_id, leads)
    updated = lead.model_copy(update=update.model_dump())
    leads.save(updated)
    return updated


@app.delete("/leads/{lead_id}", status_code=status.HTTP_204_NO_CONTENT, dependencies=admin_only)
def delete_lead(lead_id: int, leads: StoreDep) -> None:
    if not leads.delete(lead_id):
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Lead not found")

$ export LEADS_API_KEY=dev-secret-change-me   # PowerShell: $env:LEADS_API_KEY = "dev-secret-change-me"
$ fastapi dev app/main.py

 ⚡️ Starting FastAPI in development mode

 🐍 Using import string: app.main:app

 🌐 Server started at http://127.0.0.1:8000
    Documentation at http://127.0.0.1:8000/docs

  Logs:

INFO:     Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)
INFO:     Started reloader process [74464] using WatchFiles
INFO:     Started server process [42936]
INFO:     Waiting for application startup.
INFO:     Application startup complete.

$ curl -i -X POST http://127.0.0.1:8000/leads -H "Content-Type: application/json" -d '{"name": "Aigerim", "phone": "+77011234567", "service": "website"}'
HTTP/1.1 201 Created
date: Sun, 11 Oct 2026 23:13:28 GMT
server: uvicorn
content-length: 98
content-type: application/json

{"name":"Aigerim","phone":"+77011234567","service":"website","comment":null,"id":1,"status":"new"}

$ curl -s -X POST http://127.0.0.1:8000/leads -H "Content-Type: application/json" -d '{"name": "A", "phone": "call me", "service": "seo"}' | python -m json.tool
{
    "detail": [
        {
            "type": "string_too_short",
            "loc": [
                "body",
                "name"
            ],
            "msg": "String should have at least 2 characters",
            "input": "A",
            "ctx": {
                "min_length": 2
            }
        },
        {
            "type": "string_pattern_mismatch",
            "loc": [
                "body",
                "phone"
            ],
            "msg": "String should match pattern '^\\+?\\d{10,15}$'",
            "input": "call me",
            "ctx": {
                "pattern": "^\\+?\\d{10,15}$"
            }
        },
        {
            "type": "literal_error",
            "loc": [
                "body",
                "service"
            ],
            "msg": "Input should be 'website', 'mobile-app' or 'crm'",
            "input": "seo",
            "ctx": {
                "expected": "'website', 'mobile-app' or 'crm'"
            }
        }
    ]
}
$ curl -s http://127.0.0.1:8000/leads
{"detail":"Invalid or missing API key"}
$ curl -s http://127.0.0.1:8000/leads -H "X-API-Key: dev-secret-change-me"
[{"name":"Aigerim","phone":"+77011234567","service":"website","comment":null,"id":1,"status":"new"}]
$ curl -s http://127.0.0.1:8000/leads/42 -H "X-API-Key: dev-secret-change-me"
{"detail":"Lead not found"}

import pytest
from fastapi.testclient import TestClient

from app.main import LeadStore, app, get_store

API_KEY = "test-key"
ADMIN = {"X-API-Key": API_KEY}
VALID_LEAD = {"name": "Aigerim", "phone": "+77011234567", "service": "website"}


@pytest.fixture
def client(monkeypatch):
    monkeypatch.setenv("LEADS_API_KEY", API_KEY)
    fresh_store = LeadStore()
    app.dependency_overrides[get_store] = lambda: fresh_store
    yield TestClient(app)
    app.dependency_overrides = {}


def test_create_lead_returns_201_with_id(client):
    response = client.post("/leads", json=VALID_LEAD)
    assert response.status_code == 201
    assert response.json() == {**VALID_LEAD, "comment": None, "id": 1, "status": "new"}


def test_invalid_phone_returns_422(client):
    response = client.post("/leads", json={**VALID_LEAD, "phone": "call me"})
    assert response.status_code == 422
    error = response.json()["detail"][0]
    assert error["loc"] == ["body", "phone"]
    assert error["type"] == "string_pattern_mismatch"


def test_admin_routes_require_api_key(client):
    assert client.get("/leads").status_code == 401
    assert client.get("/leads", headers={"X-API-Key": "wrong"}).status_code == 401
    assert client.get("/leads", headers=ADMIN).status_code == 200


def test_missing_lead_returns_404(client):
    response = client.get("/leads/999", headers=ADMIN)
    assert response.status_code == 404
    assert response.json() == {"detail": "Lead not found"}


def test_update_status_and_filter(client):
    client.post("/leads", json=VALID_LEAD)
    client.post("/leads", json={**VALID_LEAD, "service": "crm"})

    response = client.patch("/leads/2", json={"status": "in_progress"}, headers=ADMIN)
    assert response.status_code == 200
    assert response.json()["status"] == "in_progress"

    in_progress = client.get("/leads", params={"status": "in_progress"}, headers=ADMIN).json()
    assert [lead["id"] for lead in in_progress] == [2]


def test_limit_is_validated(client):
    response = client.get("/leads", params={"limit": 0}, headers=ADMIN)
    assert response.status_code == 422


def test_delete_returns_204_and_removes_lead(client):
    client.post("/leads", json=VALID_LEAD)
    response = client.delete("/leads/1", headers=ADMIN)
    assert response.status_code == 204
    assert response.content == b""
    assert client.get("/leads/1", headers=ADMIN).status_code == 404

$ pytest -v
============================= test session starts =============================
platform win32 -- Python 3.12.10, pytest-9.1.1, pluggy-1.6.0 -- …\.venv\Scripts\python.exe
cachedir: .pytest_cache
rootdir: …\leads-api
configfile: pyproject.toml
testpaths: tests
plugins: anyio-4.15.1
collecting ... collected 7 items

tests/test_leads.py::test_create_lead_returns_201_with_id PASSED         [ 14%]
tests/test_leads.py::test_invalid_phone_returns_422 PASSED               [ 28%]
tests/test_leads.py::test_admin_routes_require_api_key PASSED            [ 42%]
tests/test_leads.py::test_missing_lead_returns_404 PASSED                [ 57%]
tests/test_leads.py::test_update_status_and_filter PASSED                [ 71%]
tests/test_leads.py::test_limit_is_validated PASSED                      [ 85%]
tests/test_leads.py::test_delete_returns_204_and_removes_lead PASSED     [100%]

============================== 7 passed in 0.43s ==============================