code-framework 1.0.0b1

Async-first Python 3.13 framework bundling an HTTP client with OAuth 2.0, Redis caching, Flagr feature flags, and OpenTelemetry observability into one internal dependency.

code-framework

code-framework is a Python 3.13, async-first framework that bundles five building blocks commonly needed by backend services into one internal dependency:

• code_framework/http — an async HTTP client (HttpClient, wrapping httpx.AsyncClient) with an OAuth 2.0 client-credentials helper (OAuthClient), thin request helper functions, and ASGI middleware for binding a per-request client to a contextvars context.
• code_framework/cache — an async Redis cache client (CacheClient) plus a @cache cache-aside decorator with distributed-lock thundering-herd protection and TTL helpers.
• code_framework/flags — an async Flagr feature-flag client (FlagrClient) with optional cache-aside evaluation.
• code_framework/observability — structured JSON logging plus OpenTelemetry distributed tracing, wired up with one call (setup_logger).
• code_framework/exceptions — a standardized HTTP status-code exception hierarchy (FrameworkHttpError + subclasses) shared by the request helpers and OAuthClient.

All I/O is async/await. TLS 1.2+ is enforced by default everywhere a network connection is opened. The package is published to PyPI via GitHub Actions and is meant to be consumed as a dependency, not vendored. Proprietary — all rights reserved (see LICENSE); published on PyPI for internal distribution convenience, not as an open-source release.

---

Installation

Base install only requires httpx — redis and the OpenTelemetry stack are optional extras, since not every consumer needs caching or tracing:

[project]
dependencies = [
    "code-framework",
]

Pull in only what you use via extras:

uv add code-framework                     # http + exceptions only
uv add "code-framework[cache]"            # + CacheClient (redis)
uv add "code-framework[observability]"    # + setup_logger/ObservabilityFilter (OTel SDK)
uv add "code-framework[all]"              # everything

or with pip:

pip install code-framework
pip install "code-framework[cache,observability]"

code_framework.cache.CacheClient and code_framework.observability.setup_logger / ObservabilityFilter are resolved lazily (PEP 562) — importing code_framework.http or code_framework.flags never requires redis or the OTel packages to be installed; only touching those two specific symbols does, and only then do you need the matching extra.

Requires Python 3.13+.

---

Quickstart: HttpClient

HttpClient (code_framework/http/client.py) is always used as an async context manager — accessing .session outside the async with block raises RuntimeError.

from code_framework.http import HttpClient, get

async def main() -> None:
    async with HttpClient(base_url="https://api.example.com", timeout=30.0) as client:
        response = await get(client, "/users/1", raise_for_status=True)
        print(response.json())

Constructor signature (code_framework/http/client.py):

HttpClient(
    base_url: str = "",
    timeout: float = 30.0,
    response_extractor: Callable[[httpx.Response], dict[str, Any]] | None = None,
    root_extractor: Callable[[httpx.Response], dict[str, Any]] | None = None,
)

• base_url / timeout are passed straight through to the underlying httpx.AsyncClient (follow_redirects=True is always on).
• root_extractor and response_extractor both run on every response via an httpx event hook and feed data into the outbound observability context (see Observability below). They're merged in order — root_extractor first, then response_extractor overlays on top (same key wins for response_extractor). Use root_extractor for fields common to every call through this client (wire it up once, e.g. in app.py alongside HttpClientMiddleware); use response_extractor only to add fields specific to one talker's calls — e.g. a request ID pulled from a response header that only this particular API returns.
• TLS: an ssl.SSLContext with minimum_version = ssl.TLSVersion.TLSv1_2 is created internally and passed as verify=.

Request helpers (code_framework/http/requests.py)

Each helper takes any object satisfying ClientProtocol (i.e. anything exposing a .session property — HttpClient included) plus a URL, and forwards *args/**kwargs to the matching httpx.AsyncClient method:

from code_framework.http import get, post, put, patch, delete, head, options

All seven accept raise_for_status: bool = False as a keyword-only convenience — pass raise_for_status=True to call response.raise_for_status() before returning. Every call also emits an http request log line (method, url, status, duration_ms) when outbound HTTP logging is active (see HttpClientMiddleware below) — no setup needed beyond that middleware.

Log level tracks the response status, even without raise_for_status. A plain get(client, url) call (the default, no exception involved) logs at INFO for a 2xx/3xx response, WARNING for a 4xx, and ERROR for a 5xx — via log_level_for_status() (code_framework/exceptions/mapping.py). This means a talker that never opted into raise_for_status=True still gets severity-appropriate logs for a "normal" 404/409/etc. response. Status codes outside the explicitly mapped set (see below) always log at ERROR, on the assumption that an unrecognized status is worth a closer look regardless of its numeric range.

Error handling (code_framework/exceptions)

Request helpers translate failures into a typed exception hierarchy instead of letting raw httpx exceptions leak out:

from code_framework.exceptions import FrameworkHttpError, NotFoundError, TransportError

try:
    response = await get(client, "/users/1", raise_for_status=True)
except NotFoundError:
    ...
except FrameworkHttpError as exc:  # any other mapped 4xx/5xx
    logger.warning("upstream call failed", extra={"key": exc.key, "status": exc.status_code})

• raise_for_status=True + a 4xx/5xx response → response.raise_for_status() raises httpx.HTTPStatusError, which is caught and re-raised (raise ... from exc, original preserved as __cause__) as the matching FrameworkHttpError subclass via exception_for_status() (code_framework/exceptions/mapping.py): BadRequestError(400), UnauthorizedError(401), ForbiddenError(403), NotFoundError(404), ConflictError(409), UnprocessableEntityError(422), TooManyRequestsError(429), InternalServerError(500), BadGatewayError(502), ServiceUnavailableError(503), GatewayTimeoutError(504), or UnknownHttpError for any other status code.
• Network/timeout failures (httpx.HTTPError other than HTTPStatusError) always raise TransportError — regardless of raise_for_status, since there is no response to decide against. This is a behavior change from a plain httpx client: a talker that never set raise_for_status=True still needs to handle TransportError (or the common base FrameworkHttpError) around connection failures/timeouts.
• Every mapped/transport failure also emits an http request failed log line (namespaces http and error, with error.key matching the exception's key) whenever outbound HTTP logging is active — the same gating as the success-path log. The level follows the exception's own log_level class attribute: WARNING for the 4xx classes above, ERROR for the 5xx classes, TransportError, and UnknownHttpError (the last one is ERROR even when the underlying status code is a 4xx the framework doesn't explicitly recognize — an unmapped code is treated as worth escalating, not assumed to be a routine client error).
• Each exception carries status_code, key, message (default_message unless you construct it yourself with a custom one), and log_level (logging.WARNING/logging.ERROR); any extra keyword arguments passed to the constructor land in .troubleshooting.

Binding a client to the request context

code_framework/http/context.py exposes get_request_client, set_request_client, and reset_request_client, backed by a ContextVar. HttpClientMiddleware (code_framework/http/middleware.py) is an ASGI middleware that opens one client per request and binds it automatically:

from code_framework.http import HttpClientMiddleware, get, get_request_client

def common_fields(response: httpx.Response) -> dict[str, Any]:
    return {"request_id": response.headers.get("x-request-id", "")}

app.add_middleware(
    HttpClientMiddleware,
    base_url="https://api.example.com",
    root_extractor=common_fields,   # wired up once, applied to every call through this middleware
)

async def handler():
    client = get_request_client()          # the HttpClient opened for this request
    return await get(client, "/users/1")

Pass client_factory instead of base_url/timeout to bind any other async-context-manager client (a composite client, an OAuth-wrapping client, etc.) — see the docstring in code_framework/http/middleware.py for the exact pattern. root_extractor is ignored when client_factory is supplied (same rule as base_url/timeout) — reuse the same function inside your factory's own HttpClient(...) call, and add a response_extractor alongside it for that talker's call-specific fields, if needed.

---

OAuth 2.0 client-credentials flow

OAuthClient (code_framework/http/oauth.py) implements the client_credentials grant with in-memory token caching (_TokenCache, a private @dataclass with a 30-second expiry buffer):

from code_framework.http import OAuthClient

oauth = OAuthClient(
    token_url="https://auth.example.com/oauth/token",
    client_id="my-client-id",
    client_secret="my-client-secret",
    scope="read:orders",  # optional
)

headers = await oauth.auth_headers()  # {"Authorization": "Bearer <token>"}

• get_token() returns the cached token if still valid, otherwise POSTs to token_url with grant_type=client_credentials (application/x-www-form-urlencoded) and caches the result.
• auth_headers() is a convenience wrapper returning a ready-to-merge header dict.
• Like HttpClient, the internal token request uses an ssl.SSLContext with minimum_version = ssl.TLSVersion.TLSv1_2.
• Errors are mapped to code_framework.exceptions, same as the request helpers — a 4xx/5xx from the token endpoint raises the matching FrameworkHttpError subclass (e.g. UnauthorizedError on 401), network/timeout failures raise TransportError, and a malformed token response (non-JSON body, or JSON missing access_token) raises UnknownHttpError. All chained (from exc) to the original error.
• OAuthClient never logs, by design — unlike requests.py, no ERROR log is emitted on failure here. The token request body carries client_secret and the response carries access_token; to keep those out of any log pipeline entirely, this client raises the standardized exception and nothing else. If you need troubleshooting visibility, catch the raised exception at the call site and log only what you've confirmed is safe (e.g. exc.key/exc.status_code — never the raw response body).

Combine it with HttpClient by merging headers per call, or via a small composite client injected through HttpClientMiddleware's client_factory.

---

Caching (code_framework/cache)

CacheClient (code_framework/cache/client.py) is an async Redis client, also used as an async context manager, backed by a persistent connection pool:

from code_framework.cache import CacheClient

async with CacheClient(
    host="localhost",
    port=6379,
    password="",
    default_ttl=300,
    max_connections=10,
    use_ssl=True,          # default; set False only for local dev without TLS
) as cache:
    await cache.set("my-key", {"hello": "world"}, ttl=60)
    value = await cache.get("my-key")   # raw JSON string, or None on miss/error
    exists = await cache.exists("my-key")
    await cache.delete("my-key")

Notes on behavior (see docstrings in code_framework/cache/client.py for full detail):

• set/append JSON-serialize the value; get returns the raw JSON string — deserialize it yourself (the @cache decorator does this for you).
• Failures are silent by design: get returns None, set/append/delete become no-ops, exists returns False. A cache miss is not an error.
• TLS is on by default (use_ssl=True), enforcing ssl.TLSVersion.TLSv1_2 minimum.
• cache.lock(key, timeout=5.0, blocking_timeout=2.0) returns a Redis distributed lock, usable as async with cache.lock("my-key"):.

The @cache decorator

code_framework/cache/decorator.py provides cache-aside wrapping for async handlers, with double-checked locking to avoid a thundering herd on cache miss:

from code_framework.cache import cache, build_cache_key

@cache(client=cache_client, key=build_cache_key("products", "detail", product_id), ttl=60)
async def get_product(product_id: int) -> dict:
    ...

ttl accepts an int, a zero-arg callable evaluated at each cache miss (e.g. ttl_until_midnight), or None to use the client's default_ttl. Any client or deserialization error falls back to calling the wrapped handler directly.

TTL and locking helpers (code_framework/cache/helpers.py)

from code_framework.cache import build_cache_key, resolve_ttl, ttl_until, ttl_until_midnight, ttl_until_next, safe_lock

• build_cache_key(*parts) — joins parts with : (e.g. build_cache_key("detail", user_id)).
• resolve_ttl(value) — converts an int or timedelta to a clamped, non-negative seconds value.
• ttl_until(target_time, tz=UTC) / ttl_until_midnight(tz=UTC) / ttl_until_next(targets, tz=UTC) — seconds until a daily clock time (or the nearest of several), for clock-based expirations.
• safe_lock(client, key, timeout=5.0, blocking_timeout=2.0) — async context manager that yields True/False for lock acquisition without ever raising; use it when you need explicit locking outside of @cache.

Protocols (code_framework/cache/protocols.py)

CacheClientProtocol (get/set/append/delete/exists/lock) and CacheObserverProtocol (on_hit/on_miss/on_error) let you type-hint against the interface and substitute test doubles or wrappers without inheriting from CacheClient.

---

Feature flags (code_framework/flags)

FlagrClient (code_framework/flags/client.py) evaluates feature flags against a Flagr server, also as an async context manager:

from code_framework.flags import FlagrClient

async with FlagrClient(base_url="https://flagr.example.com") as flags:
    enabled = await flags.is_enabled("pvp_new_mode", entity_id=user_id, entity_context={"rank": "5"})
    variant = await flags.get_variant("checkout_experiment", entity_id=user_id)

• is_enabled returns False, get_variant returns None, on any HTTP/network/parse error — flags degrade gracefully rather than breaking the request.
• entity_context passes arbitrary attributes (e.g. rank, region) matched against segment constraints configured in the Flagr UI.
• Optional transparent caching: pass cache=<CacheClientProtocol> and cache_ttl= (same forms as @cache's ttl) to avoid a Flagr round trip on every evaluation. Only do this when the same flag/entity/context combination repeats — see the docstring in code_framework/flags/client.py for when caching helps vs. when the app should cache the result itself instead.

FeatureFlagProtocol (code_framework/flags/protocols.py) exposes is_enabled/get_variant for typing against the interface instead of the concrete FlagrClient.

---

Observability

code_framework/observability provides structured JSON logging and OpenTelemetry tracing, wired up with a single call at startup:

from code_framework.observability import setup_logger

setup_logger("orders-api")  # local fallback: spans to console, trace_id/span_id in every log

setup_logger(service_name, otlp_endpoint=None, handler=None, level=logging.INFO, headers=None) (code_framework/observability/setup.py):

• Creates an OTel TracerProvider with a Resource tagging service_name, and instruments httpx automatically via HTTPXClientInstrumentor — every outbound HttpClient call gets a span.

• Default is a local, infra-free fallback: otlp_endpoint=None (the default) exports spans to a ConsoleSpanExporter, so trace_id/span_id show up in logs with no collector running — useful for local dev and tests.

• Pass otlp_endpoint to export to a real destination (a self-hosted OTel Collector, a Datadog Agent's OTLP intake, a vendor's SaaS OTLP endpoint, ...) — spans go to {otlp_endpoint}/v1/traces via OTLPSpanExporter. This function never reads environment variables itself; read os.getenv(...) at the call site if you want that driven by config:

  import os
  from code_framework.observability import setup_logger
  
  setup_logger("orders-api", otlp_endpoint=os.getenv("OTLP_ENDPOINT"))
  

• headers are forwarded to every OTLP export request — e.g. {"DD-API-KEY": "..."} for a vendor SaaS OTLP intake that requires an API key. Ignored when otlp_endpoint is None.

• Installs ObservabilityFilter (code_framework/observability/filter.py) and ObservabilityFormatter (code_framework/observability/formatter.py) on the given handler (or a StreamHandler(sys.stderr) by default) and attaches it to the root logger at level.

After calling setup_logger, every logging.getLogger(...).info(...) call emits structured JSON with timestamp, level, key (logger name), message, framework_version, trace_id, span_id, and — when populated — namespaced fields http, cache, flag, error, context.

Per-request context

ObservabilityMiddleware (code_framework/observability/middleware.py) is an ASGI middleware that attaches a structured context to each HTTP request and emits one log per request on completion (method, path, user-agent, status, duration_ms, plus anything set during the request). The level mirrors the same log_level_for_status() rule used by the outbound request helpers: INFO for a 2xx/3xx response this service returns, WARNING for a 4xx, ERROR for a 5xx — so a handled "not found" response logs differently from a normal one without anyone needing to raise or catch anything. If the wrapped app lets an exception propagate uncaught, the log is always ERROR with exc_info=True (populating error.traceback) regardless of status code, and the exception continues propagating after the log is emitted:

from code_framework.observability import ObservabilityMiddleware

app.add_middleware(
    ObservabilityMiddleware,
    context_extractor=lambda headers: {"tenant": headers.get("x-tenant-id", "")},
)

Application code can enrich the current request's context with set_context(message="...", extra={...}) (code_framework/observability/context.py, re-exported from code_framework.observability), which sets the eventual log's message and merges extra into the context namespace. Framework components (CacheClient, FlagrClient, HttpClient's outbound logging) populate their own namespaces (cache, flag, http) automatically via the internal sign_context helper — you don't need to call it yourself.

---

Security notes

TLS 1.2+ (ssl.TLSVersion.TLSv1_2 minimum) is enforced by default in every component that opens a network connection:

• HttpClient (code_framework/http/client.py)
• OAuthClient (code_framework/http/oauth.py)
• CacheClient (code_framework/cache/client.py, when use_ssl=True, the default — pass use_ssl=False only for local development against a non-TLS Redis instance)

Do not weaken these settings when extending the framework.

---

Development workflow

uv sync --all-extras     # install runtime + dev dependencies

make analyse              # ruff format, ruff check --fix, bandit, semgrep, pip-audit
make test                  # uv run pytest (asyncio_mode = "auto", tests live in tests/)
make build                 # uv build

Individual tools (see pyproject.toml for exact config — line length 120, double quotes, ruff rules E, W, F, I, B, C4, UP, S, N, RUF with S101/E501 ignored, bandit at medium severity/confidence with B101 skipped):

uv run ruff format code_framework
uv run ruff check code_framework --fix
uv run bandit -c pyproject.toml -r code_framework
uv run semgrep scan --config=auto code_framework
uv run pip-audit --ignore-vuln GHSA-5239-wwwm-4pmq
uv run pytest

This repository also defines Claude Code skills/agents (.claude/skills/, .claude/agents/) that automate this workflow for contributors using Claude Code — see CLAUDE.md for details.

---

CI/CD and releases

GitHub Actions (.github/workflows/):

• ci.yml — on every push/PR to prd or dev: make analyse then make test.
• beta.yml — analyse, test, then build + uv publish to PyPI under the beta environment. Two ways to trigger it:
  • Push a tag matching v*b* (e.g. v1.0.0b1) directly — the exact version is whatever you tagged.
  • Run it manually (Actions tab → "Beta Release" → "Run workflow") and pick patch / minor / major from the bump dropdown. The workflow computes the next version itself: it bumps the chosen segment off the latest stable tag (0.0.0 if none exists yet — so a minor bump on this repo's first-ever release produces v0.1.0b1), then finds the highest existing beta already cut for that target and continues the b counter (or starts at b1 if none exists), creates that tag, pushes it, and runs the same analyse/test/publish pipeline against it.
• release.yml — same two-trigger shape as beta.yml, but for stable releases: push a tag matching v[0-9]+.[0-9]+.[0-9]+ directly, or run it manually with a bump choice. The computed tag has no b suffix, and the workflow refuses to run (fails fast) if that exact version was already tagged before — it will never silently overwrite a published release.
• Both beta.yml and release.yml delegate their actual analyse → test → build → publish steps to a shared reusable workflow (_release-pipeline.yml), so there's one place that defines what "run the pipeline" means regardless of which trigger kicked it off.

Publishing uses OIDC trusted publishing (id-token: write) — no PyPI API tokens are stored as secrets. The published version itself is never hand-edited — [tool.hatch.version] source = "vcs" derives it from whichever git tag triggered the run.