varco-fastapi

FastAPI integration and HTTP client utilities for varco.

Provides structured HTTP connection configuration, TLS trust-store management, JWT authority, and HTTP middleware wiring on top of FastAPI and httpx. Requires varco-core.

Install

pip install varco-fastapi

HTTP connection settings

HttpConnectionSettings is a structured config object that produces kwargs for httpx.AsyncClient (or httpx.Client).

Unlike the Postgres/Redis/Kafka settings, there is no fixed env-var prefix. A service typically calls many different external HTTP APIs — a hardcoded HTTP_ prefix would only allow one of them to be configured from env vars at a time. Instead, you supply a prefix when loading from env:

payment = HttpConnectionSettings.from_env(prefix="PAYMENT_API_")
notify  = HttpConnectionSettings.from_env(prefix="NOTIF_API_")

Plain connection (no auth, no TLS)

import httpx
from varco_fastapi.connection import HttpConnectionSettings

# Direct construction — no env vars read
conn = HttpConnectionSettings(base_url="https://api.example.com/v1", timeout=10.0)

async with httpx.AsyncClient(**conn.to_httpx_kwargs()) as client:
    response = await client.get("/users")

From environment variables (multi-client)

# Payment API
PAYMENT_API_BASE_URL=https://pay.example.com/v1
PAYMENT_API_TIMEOUT=5.0

# Notification API
NOTIF_API_BASE_URL=https://notify.example.com
NOTIF_API_TIMEOUT=10.0

payment = HttpConnectionSettings.from_env(prefix="PAYMENT_API_")
notify  = HttpConnectionSettings.from_env(prefix="NOTIF_API_")

async with httpx.AsyncClient(**payment.to_httpx_kwargs()) as client:
    await client.post("/charge", json={"amount": 9.99})

You can also configure via host and port instead of a full URL:

MY_SVC_HOST=api.example.com
MY_SVC_PORT=8080
# effective base_url → "http://api.example.com:8080"

With Basic authentication

from varco_core.connection import BasicAuthConfig

conn = HttpConnectionSettings(
    base_url="https://api.example.com",
    auth=BasicAuthConfig(username="svc-user", password="secret"),
)
# to_httpx_kwargs() includes auth=("svc-user", "secret") automatically

async with httpx.AsyncClient(**conn.to_httpx_kwargs()) as client:
    response = await client.get("/protected")

From env:

MY_SVC_BASE_URL=https://api.example.com
MY_SVC_AUTH__TYPE=basic
MY_SVC_AUTH__USERNAME=svc-user
MY_SVC_AUTH__PASSWORD=secret

conn = HttpConnectionSettings.from_env(prefix="MY_SVC_")

With OAuth2 static bearer token

from varco_core.connection import OAuth2Config

conn = HttpConnectionSettings(
    base_url="https://api.example.com",
    auth=OAuth2Config(token="eyJhbGciOiJSUzI1NiJ9..."),
)
# OAuth2 is NOT injected into kwargs automatically — httpx has no built-in
# OAuth2 flow.  Add the Authorization header via a middleware or event hook:
async with httpx.AsyncClient(**conn.to_httpx_kwargs()) as client:
    response = await client.get(
        "/protected",
        headers={"Authorization": f"Bearer {conn.auth.token}"},
    )

Note: For OAuth2 client-credentials flows (token refresh), use an httpx event hook or middleware — HttpConnectionSettings is a pure config object and does not manage token lifecycle.

With TLS / SSL (custom CA)

from varco_core.connection import SSLConfig
from pathlib import Path

ssl = SSLConfig(ca_cert=Path("/etc/ssl/api-ca.pem"), verify=True)
conn = HttpConnectionSettings.with_ssl(
    ssl,
    base_url="https://secure-api.example.com",
)
# to_httpx_kwargs()["verify"] → ssl.SSLContext built from the CA cert

async with httpx.AsyncClient(**conn.to_httpx_kwargs()) as client:
    response = await client.get("/data")

From env:

MY_SVC_BASE_URL=https://secure-api.example.com
MY_SVC_SSL__CA_CERT=/etc/ssl/api-ca.pem
MY_SVC_SSL__VERIFY=true

conn = HttpConnectionSettings.from_env(prefix="MY_SVC_")

Disable TLS verification (dev / testing only)

ssl = SSLConfig(verify=False, check_hostname=False)
conn = HttpConnectionSettings.with_ssl(ssl, base_url="https://localhost:8443")
# to_httpx_kwargs()["verify"] → False

With mTLS (client certificates)

ssl = SSLConfig(
    ca_cert=Path("/etc/ssl/ca.pem"),
    client_cert=Path("/etc/ssl/client.crt"),
    client_key=Path("/etc/ssl/client.key"),
)
conn = HttpConnectionSettings.with_ssl(ssl, base_url="https://mtls-api.example.com")
async with httpx.AsyncClient(**conn.to_httpx_kwargs()) as client:
    response = await client.get("/secure")

Bridge to TrustStore (legacy ClientProfile)

trust_store = conn.to_trust_store()   # None when ssl is not set
# use with ClientProfile.production(trust_store=trust_store)

Connection settings reference

All field names below assume a prefix of MY_SVC_ — replace it with your own.

Service-free (generic) REST servers

Use GenericRouter when the server has no AsyncService or repository — for example a data-transformation pipeline, an API gateway, or computed analytics routes. All cross-cutting features (middleware, telemetry, auth, authorization) work identically.

from varco_fastapi.router.presets import GenericRouter
from varco_fastapi.router.endpoint import route
from varco_fastapi.auth import JwtBearerAuth
from varco_fastapi.auth.guard import require_scopes, require_roles, allow_anonymous

class ReportRouter(GenericRouter):
    _prefix = "/reports"
    _auth = JwtBearerAuth(...)

    # Requires scope — denies 403 if caller does not have "reports:read"
    @route("GET", "/summary", requires=require_scopes("reports:read"))
    async def get_summary(self, ctx) -> dict:
        return {"total": 42}

    # Requires role
    @route("DELETE", "/cache", requires=require_roles("admin"))
    async def purge_cache(self, ctx) -> None: ...

    # Public endpoint — allow_anonymous bypasses auth checks entirely
    @route("GET", "/status", requires=allow_anonymous())
    async def status(self, ctx) -> dict:
        return {"ok": True}

app = create_varco_app(routers=[ReportRouter])

Available guard helpers (varco_fastapi.auth.guard):

Custom @route handlers — full FastAPI parameters

A custom @route method may declare any parameter a normal FastAPI endpoint can, and FastAPI parses, validates, coerces and injects it — Query(...), Body(...) (Pydantic models), Depends(...), Request/Response/BackgroundTasks, and type-coerced path params. The return annotation drives the OpenAPI response model. ctx/auth/context still receive the router's AuthContext, and any RouteGuard still runs before the handler.

from fastapi import Body, Depends, Query, Request
from pydantic import BaseModel

from varco_core.auth.base import AuthContext

class SummaryFilter(BaseModel):
    since: str | None = None

class ReportRouter(GenericRouter):
    _prefix = "/reports"
    _auth = JwtBearerAuth(...)

    @route("POST", "/{report_id}/summary", requires=require_scopes("reports:read"))
    async def summary(
        self,
        report_id: int,                       # typed path param — coerced to int
        ctx: AuthContext,                     # injected from _auth
        window: int = Query(30, ge=1, le=365),  # validated query param (422 on bad input)
        filters: SummaryFilter = Body(...),   # Pydantic request body
        repo: Repo = Depends(get_repo),       # arbitrary FastAPI dependency
        request: Request = None,              # raw request if you want it
    ) -> SummaryResponse:                     # → OpenAPI response model
        ...

Under the hood build_router() synthesizes a wrapper whose __signature__ mirrors the method, so FastAPI drives all parsing natively — no manual request handling needed.

Composite / all-in-one deployment

Combine several independently-built services into a single deployable process without changing any of them. Each service keeps its own DIContainer, database, environment, middleware, and /docs — they are mounted side by side under path prefixes.

from varco_fastapi import create_composite_app, ServiceMount

from orders_service.app import app as orders_app      # its own create_varco_app()
from billing_service.app import app as billing_app     # its own container + DB

composite = create_composite_app([
    ServiceMount("/orders", orders_app),
    ServiceMount("/billing", billing_app),
])
# uvicorn composite:composite
#   /orders/...   → orders service (own docs at /orders/docs)
#   /billing/...  → billing service (own docs at /billing/docs)
#   /health       → aggregate health across both services (503 if any is down)
#   /             → landing page listing each service

create_composite_app installs a CompositeLifespan that drives each sub-app's own lifespan — Starlette does not propagate lifespan into mounted apps, so this is what actually starts each service's DB pool / event bus / outbox relay. Startup is fail-fast (one broken service aborts the whole process); shutdown is LIFO.

All services share one os.environ. Runtime isolation is automatic; the only hazard is build-time env-name collisions. Either namespace env vars per service (ORDERS_DB_URL, BILLING_DB_URL) or use build_service for a scoped overlay:

from varco_fastapi import build_service

orders = build_service("/orders", create_orders_app,
                        env={"DATABASE_URL": "postgresql+asyncpg://.../orders"})
billing = build_service("/billing", create_billing_app,
                        env={"DATABASE_URL": "postgresql+asyncpg://.../billing"})
composite = create_composite_app([orders, billing])

See technical_docs/features/composite-deployment.md and examples/23-composite-all-in-one/ for the full reference.

Related packages

Links

• Repository: https://github.com/edoardoscarpaci/varco
• Issue tracker: https://github.com/edoardoscarpaci/varco/issues