Quando um projeto é iniciado, os recursos são priorizados sobre a estrutura. À medida que a complexidade cresce, a localização do código do terminal fica mais difícil e a duplicação e as importações circulares entram. Embora a estrutura exata varia de acordo com as necessidades de negócios, a maioria dos servidores da Web compartilham preocupações semelhantes e podem consistir nas seguintes pastas, separadas com base na funcionalidade:

main.py – Criar aplicativo; fiação de vida útil; Registre middleware e manipuladores de erros.

Validadores/ – Verificações de entrada reutilizáveis ​​(IDs, enumes, regras de negócios) usadas pelo Pydantic.
middleware/ -Camadas cruzadas (por exemplo, guarda de tamanho útil).

error_handlers/ – Mapeamento de resposta de exceção-http;
utilitários/ – Pequenos ajudantes (invólucros de resposta, paginação, formatação).
modelos/ – Modelos Pydantic para validar solicitações de entrada e respostas de saída;

serviços/ – lógica de negócios; orquestra repositórios/clientes.

clientes/ – Clientes: DB Engine/SessionMaker, HTTP Client, adaptadores de log/rastreamento.

fábricas/ -Construtores favoráveis ​​à injeção de dependência para clientes/serviços.

Pontos de extremidade/ – Apirouters agrupados por domínio de negócios (contas/, compras/, pedidos/);
testes/ – Testes de unidade/integração; Use substituições de dependência e configurações de teste.

A parte complicada não são as pastas – é o relacionamento entre pais/filhos do roteador.

Vamos considerar esse design da API como objetivo.

/accounts
/accounts/{account_id}
/accounts/{account_id}/orders
/accounts/{account_id}/orders/{order_id}

Para conseguir isso, criaremos dois roteadores:

localização: App/endpoints/Accounts/Orders/Root.py

serve: /{Account_id}/Ordens

orders = APIRouter(
    prefix="/{account_id}/orders", 
    tags=("orders")
)

@orders.get("
async def list_orders(
    account_id: Annotated(UUID, Path()), 
    last_id: str | None = None, 
    limit: int = 50
):
    return {
      "account": str(account_id), 
      "items": (), 
      "next": None
    }

@orders.post("
async def create_order(
    account_id: Annotated(UUID, Path())
): return { "account": str(account_id), "ok": True}

localização: App/endpoints/Accounts/root.py

serve: /Contas

from .orders.root import orders as orders_router

accounts = APIRouter(
    prefix="/accounts", 
    tags=("accounts")
)

async def load_account(
    account_id: Annotated(UUID, Path())
): return {"id": account_id}

@accounts.get("
async def list_accounts(): return ({"id": "12345"})

@accounts.get("/{account_id}")
async def get_account(account_id: UUID):
    return {"id": account_id}

accounts.include_router(
    orders_router,
    # DI imposed on the order routes
    dependencies=(Depends(load_account)),
)

Essa estrutura garante que “ordens” não possam ser acessadas sem especificar um account_ide permite que os desenvolvedores identifiquem rapidamente a localização de um terminal e criem várias camadas de roteadores infantis. Com esse design, é possível reutilizar o “roteador de pedidos” em outros locais com dependências ajustadas.

Uma fábrica encapsula a construção de clientes e serviços usando o estado do aplicativo e várias variáveis ​​de ambiente e configurações.

Uma fábrica pode ser usada para gerenciar:

1. Clientes em nuvem – SQS, S3, SNS, etc.
2. Clientes de banco de dados – Redis, Postgres, Mongo
3. Fiação da configuração-Nível de aplicativo e nível de solicitação (relevante se cada cliente tiver uma configuração específica)
4. Classes de serviço, encapsulando a lógica de negócios – Service de conta, OrderService.
5. Classes auxiliares – madeireiros, sessões de usuário, etc.

class Factory:
    @classmethod
    async def get_http_client(
        cls, 
        settings
    ): return AsyncClient(base_url=settings.api)

    @classmethod
    async def get_account_service(
      cls, 
      request: Request
    ) -> AccountService:
      settings: Settings = request.app.state.settings 
      return AccountService(
          client=await cls.get_http_client(settings),
      )

Com base neste exemplo, vemos que uma fábrica bem projetada pode ser conectada à injeção de dependência da FASTAPI com um esforço mínimo.

@app.get("/accounts/{account_id}")
async def account_overview(
    account_id: str,
    account_service: Annotated(
        AccountService,
        Depends(Factory.get_account_service)
    )
):
    r = await account_service.fetch_profile(account_id)
    if not r:
        raise HTTPException(
            status_code=404, 
            detail="Account not found"
        )
    return r

O terminal permanece fino e legível: valida a entrada, orquestra serviços e mapeia os resultados internos para as respostas HTTP. A lógica de negócios vive fora do ponto final; O manipulador se concentra nas preocupações com HTTP (fiação, códigos de status e formatação).

A fábrica pode simplificar bastante o acesso a diferentes madeireiros no servidor.

class Factory
    @classmethod
    def get_root_logger(cls) -> logging.Logger:
        return logging.getLogger("app")

    @classmethod
    def get_request_logger(
        cls, 
        request: Request
     ) -> logging.LoggerAdapter:
        base = cls.get_root_logger()
        return logging.LoggerAdapter(
            base,
            {
              "path": request.url.path, 
              "method": request.method,
              "corr_id": request.state.corr_id,
            }
        )

Agora, todos os tipos de madeireiros podem ser facilmente instanciados por meio da fábrica em pontos de extremidade:

acc_ser_dep = Annotated(
    AccountService,
    Depends(Factory.get_account_service),
)
log_dep = Annotated(
    LoggerAdapter,
    Depends(Factory.get_request_logger),
)

@app.get("/accounts/{account_id}")
async def account_overview(
    account_id: str,
    account_service: acc_ser_dep,
    logger: log_dep,
):
    r = await account_service.fetch_profile(account_id)
    if not r:
        logger.warning(
            "Account not found",
            extra={"account_id": account_id},
        )
        raise HTTPException(404, "Account not found")
    return r

Um madeireiro escondido para solicitação precisa de um ID de correlação para costurar os registros de uma solicitação em uma história coerente. É comum gerar o ID dentro dos pontos de extremidade, mas isso leva à repetição. É mais eficiente usar um pequeno middleware que define o ID uma vez por solicitação e pode ser estendido para incluir user_idAssim, firm_ide outro contexto:

@app.middleware("http")
async def corr_middleware(request: Request, call_next):
    _id = request.headers.get("X-Request-ID") or str(uuid4())
    request.state.corr_id = _id

    response = await call_next(request)
    response.headers("X-Request-ID") = _id
    return response

Voltando ao exemplo da fábrica, houve uma etapa fácil de acalmar que simplifica a arquitetura de injeção de dependência.

settings: Settings = request.app.state.settings

Esta linha assume que as configurações são inicializadas na inicialização (main.py):

@asynccontextmanager
async def lifespan(app: FastAPI):
    app.state.settings = Settings()
    yield
    await Factory.aclose()


app = FastAPI(lifespan=lifespan)

Objetos anexados a app.state são escondidos ao longo da vida para cada processo de trabalhador do FASTAPI e devem ser fechados na fase de desligamento.

Os objetos de vida útil podem persistir em muitas solicitações e viver a vida inteira do servidor. Usando app.state não é a única maneira de segurá -los (singletons, objetos em cache com chaves etc.), mas é a opção mais conveniente no FASTAPI.

Uma “sessão” escovida de solicitação, que agrega um estado específico de solicitação exigido pela lógica de negócios (por exemplo, correlação_id, user_id), morrerá logo após a conclusão da solicitação.

Ao longo da vida útil do servidor, podemos usar objetos anexados ao app.state (pool db, clientes HTTP etc.). Quando o servidor é desligado, acionamos o fechamento desses objetos e temos uma lógica especial para isso.

class Factory:
    @classmethod
    async def aclose(cls):
        await app.state.db_pool.aclose()
        await app.state.external_http_client.aclose()
        ...

O fato de um objeto ter sido coletado com lixo não significa que as conexões correspondentes tenham sido fechadas. É por isso que temos await Factory.aclose() na vida útil do servidor.

O mesmo padrão pode ser aplicado a clientes escondidos por solicitação por meio da fábrica.

class Factory:
    @classmethod
    async def get_account_service(cls):
        settings: Settings = request.app.state.settings

        account_service = AccountService(
            client=await cls.get_http_client(settings),
        )

        try:
            yield account_service
        finally:
            await account_service.client.aclose()

No exemplo acima account_service é excluído apenas após o fechamento das conexões do cliente HTTP, para que seus soquetes sejam liberados imediatamente.

As configurações são um bom exemplo de um objeto de vida útil que as solicitações podem acessar simultaneamente via app.state.settings. Idealmente, os objetos de vida útil devem ser somente leitura e não devem fazer referência a objetos escondidos por solicitações (para que eles possam ser coletados de lixo).

Caso contrário, existe um risco de vazamentos de memória e condições de corrida. Em geral, qualquer coisa compartilhada entre solicitações deve ter guardas de simultaneidade incorporadas (Thread/Async-Safe).

Para compartilhar o estado entre diferentes solicitações ou até trabalhadores, o armazenamento externo é uma opção mais segura e escalável. Mas o que acontece se duas solicitações do mesmo trabalhador tentarem escrever nas configurações ao mesmo tempo? Vamos considerar este exemplo:

cfg = request.app.state.settings
old = cfg.threshold
await some_async_call()  # yields, other requests run here
request.app.state.settings.threshold = old + 1

Este código deixará o estado compartilhado inconsistente. Se for absolutamente necessário atualizar o estado, um bloqueio deve ser usado

app.state.settings_lock = asyncio.Lock()

async def update_settings(app, patch: dict):
    async with app.state.settings_lock:
        settings = app.state.settings
        app.state.settings = settings.model_copy(
            update=patch,
        )

Com esta função, as configurações podem ser atualizadas com segurança:

await update_settings(request.app,{"threshold":old+1})

Vamos considerar um exemplo sem await some_async_call()

cfg = request.app.state.settings
old = cfg.threshold
request.app.state.settings.threshold = old + 1

O código é executado como um único bloco no thread de loop de eventos. Isso garante a integridade do estado, mas aumenta a latência do servidor. O FASTAPI não move automaticamente o trabalho síncrono do thread de loop de eventos para ThreadPool. Isso deve ser feito explicitamente por um desenvolvedor.

@app.get("/update-state")
async def update_state():
    # moves to threadpool
    await anyio.to_thread.run_sync(
        update_settings_sync
    )
    return {"ok": True}

Há mais uma maneira de fazê -lo via depende, mas não vale a pena uma exploração detalhada, porque ter vários threads gravar em um estado compartilhado não é o design certo.

Ambos os exemplos de trabalho síncrono de descarga funcionarão bem quando não houver atualizações de estado compartilhadas.

@app.post("/update-state")
async def update_state(
    result: dict = Depends(update_settings_dep)
):
    return result

Uma sólida compreensão de como o loop de eventos interage com o código síncrono e assíncrono ajuda um desenvolvedor a encontrar a solução mais eficaz.

A exaustão do laço de eventos geralmente ocorre quando assíncrono Os pontos de extremidade têm desempenho pesado sincronização trabalhar. Veja os exemplos abaixo:

def cpu_heavy(n: int) -> float:
    # Python CPU; never yields
    s = 0.0
    for i in range(n):
        s += math.sqrt(i)
    return s

@app.get("/cpu")
async def cpu(n: int = 10_000_000):
    # heavy CPU on the loop, requests on the worker stall
    return {"sum": cpu_heavy(n)}

@app.get("/sleep")
async def sleep(ms: int = 500):
    # blocking sleep holds the loop
    time.sleep(ms / 1000)
    return {"slept_ms": ms}

@app.get("/io")
async def io():
    # stalling call until the socket completes
    r = requests.get(
        "
        timeout=5,
    )
    return {"status": r.status_code}

A exaustão do laço de eventos é muito difícil de diagnosticar. Um desenvolvedor teria que ver quais solicitações estavam acontecendo na época da solicitação paralisada e percorrer a lógica para identificar possíveis culpados.

Normalmente, uma função Special Loop_Lag é criada no servidor (via LifeSpan) para relatar atrasos estendidos à OpenElemetria, mas esse sinal por si só raramente identifica a causa raiz.

async def lag_probe(
    interval: float = 0.5,
    warn_ms: float = 100.0
):
    loop = asyncio.get_running_loop()
    next_t = loop.time() + interval

    while True:
        await asyncio.sleep(interval)
        now = loop.time()
        lag_ms = max(0.0, (now - next_t) * 1000.0)

        if lag_ms > warn_ms:
            print(f"event_loop_lag_ms={lag_ms:.1f}")

        next_t += interval

Um anti-padrão típico em FASTAPI está tendo um ponto de extremidade para chamar outro terminal. O motivo usual é evitar a duplicação do código que já foi compartimentado em outros lugares e papel sobre a ausência de uma interface de serviço adequada. Esse padrão adiciona latência e sobrecarga desnecessárias – serialização extra, autenticação e log.

Se esse anti-padrão for aplicado sistematicamente entre os pontos de extremidade, ele ficará rapidamente fora de controle, amplificando a carga e a latência. Torna -se mais difícil estabelecer a política de escala com base no número de solicitações recebidas devido à sua natureza exponencial. Escusado será dizer que, com o tempo, mesmo os pequenos erros de design erram a bola de neve em grandes problemas.

Outro anti-padrão está girando vários trabalhadores em um servidor que depende de estruturas pesadas na memória (LLMS, grandes modelos pydantic, caches grandes, etc.). Cada trabalhador é um processo separado com seu próprio loop de eventos; portanto, esses objetos são replicados por trabalhador, aumentando a memória e o consumo de CPU. Se um trabalhador passar pelos limites do contêiner, o pod será morto a OOM (por exemplo, por Kubernetes) antes que o Gunicorn possa reciclar esse trabalhador.