Facade (congressgov)

Everything below is importable directly from congressgov (e.g. from congressgov import Bill, get_client_from_env); it's re-exported from congressgov.services. Service classes like Bill and Member are documented on the Services page instead of repeated here.

Client creation

Build API clients from environment configuration.

create_api_client

create_api_client(*, api_key: str, base_url: str | None = None, request_store: RequestStore | str | bool | None = False, store_fail: bool = False) -> AuthenticatedClient

Return an AuthenticatedClient configured for the Congress.gov API.

Parameters:

Name	Type	Description	Default
api_key	str	Congress.gov API key.	required
base_url	str | None	API base URL (default: https://api.congress.gov/v3).	None
request_store	RequestStore | str | bool | None	When True or a connection string, attach persistent request deduplication. False (default) leaves the client bare.	False
store_fail	bool	Fall through to the network when the store errors.	False

Source code in src/congressgov/services/client_factory.py

def create_api_client(
    *,
    api_key: str,
    base_url: str | None = None,
    request_store: RequestStore | str | bool | None = False,
    store_fail: bool = False,
) -> AuthenticatedClient:
    """
    Return an ``AuthenticatedClient`` configured for the Congress.gov API.

    Args:
        api_key: Congress.gov API key.
        base_url: API base URL (default: ``https://api.congress.gov/v3``).
        request_store: When ``True`` or a connection string, attach persistent
            request deduplication. ``False`` (default) leaves the client bare.
        store_fail: Fall through to the network when the store errors.
    """
    client = AuthenticatedClient(
        base_url=base_url or DEFAULT_API_BASE_URL,
        token=api_key,
        prefix="",
        auth_header_name=CONGRESS_API_AUTH_HEADER,
    )
    return _maybe_attach_store(client, request_store, store_fail=store_fail)

create_stored_api_client

create_stored_api_client(*, api_key: str, base_url: str | None = None, request_store: RequestStore | str | bool | None = True, store_fail: bool = False) -> AuthenticatedClient

Return an authenticated client with request-store deduplication enabled.

Equivalent to create_api_client(..., request_store=True).

Source code in src/congressgov/services/client_factory.py

def create_stored_api_client(
    *,
    api_key: str,
    base_url: str | None = None,
    request_store: RequestStore | str | bool | None = True,
    store_fail: bool = False,
) -> AuthenticatedClient:
    """
    Return an authenticated client with request-store deduplication enabled.

    Equivalent to ``create_api_client(..., request_store=True)``.
    """
    return create_api_client(
        api_key=api_key,
        base_url=base_url,
        request_store=request_store,
        store_fail=store_fail,
    )

create_offline_client

create_offline_client(*, base_url: str | None = None, request_store: RequestStore | str | bool | None = True) -> AuthenticatedClient

Return a client that serves API responses only from the request store.

No network requests are made. Useful for demos and CI when the store was populated by a prior online run.

Source code in src/congressgov/services/client_factory.py

def create_offline_client(
    *,
    base_url: str | None = None,
    request_store: RequestStore | str | bool | None = True,
) -> AuthenticatedClient:
    """
    Return a client that serves API responses only from the request store.

    No network requests are made. Useful for demos and CI when the store was
    populated by a prior online run.
    """
    store = _resolve_request_store(request_store)
    if store is None:
        raise ValueError("Offline mode requires a request store connection")
    client = AuthenticatedClient(
        base_url=base_url or DEFAULT_API_BASE_URL,
        token=OFFLINE_API_KEY,
        prefix="",
        auth_header_name=CONGRESS_API_AUTH_HEADER,
    )
    from congressgov.services.core.request_store import attach_offline_store

    attach_offline_store(client, store)
    return client

get_client_from_env

get_client_from_env(*, api_key_var: str = CONGRESS_API_KEY_ENV, base_url: str | None = None, token: str | None = None, load_dotenv: bool = True, request_store: RequestStore | str | bool | None = True, store_fail: bool = False, offline: bool = False) -> AuthenticatedClient

Create an authenticated client using CONGRESS_API_KEY (or api_key_var).

By default attaches a persistent request store (see CONGRESS_REQUEST_STORE and CONGRESS_WORKSPACE). Pass request_store=False to disable deduplication. Pass offline=True to serve responses only from the store (no API key required).

If load_dotenv is true and python-dotenv is installed, loads a .env file from the current working directory before reading the environment.

Raises:

Type	Description
ValueError	If no API key is available and offline is False.

Source code in src/congressgov/services/client_factory.py

def get_client_from_env(
    *,
    api_key_var: str = CONGRESS_API_KEY_ENV,
    base_url: str | None = None,
    token: str | None = None,
    load_dotenv: bool = True,
    request_store: RequestStore | str | bool | None = True,
    store_fail: bool = False,
    offline: bool = False,
) -> AuthenticatedClient:
    """
    Create an authenticated client using ``CONGRESS_API_KEY`` (or *api_key_var*).

    By default attaches a persistent request store (see ``CONGRESS_REQUEST_STORE``
    and ``CONGRESS_WORKSPACE``). Pass ``request_store=False`` to disable deduplication.
    Pass ``offline=True`` to serve responses only from the store (no API key required).

    If *load_dotenv* is true and ``python-dotenv`` is installed, loads a ``.env``
    file from the current working directory before reading the environment.

    Raises:
        ValueError: If no API key is available and *offline* is False.
    """
    if load_dotenv:
        try:
            from dotenv import load_dotenv as _load_dotenv
        except ImportError:
            pass
        else:
            _load_dotenv()

    if offline:
        return create_offline_client(base_url=base_url, request_store=request_store)

    key = token or os.environ.get(api_key_var)
    if not key:
        raise ValueError(
            f"No API key found. Set {api_key_var} or pass token= explicitly."
        )
    return create_api_client(
        api_key=key,
        base_url=base_url,
        request_store=request_store,
        store_fail=store_fail,
    )

Exceptions

Service-layer exceptions: MiddlewareError (base), with ClientNotFoundError, ValidationError, APIError (and its RateLimitError/TimeoutError subclasses), ExpansionError, RequestStoreError, and UnsupportedApiUrlError covering the specific failure modes below.

from congressgov.services.exceptions import ClientNotFoundError, APIError

if client is None:
    raise ClientNotFoundError("No client available for API calls")

try:
    response = api_call()
except httpx.HTTPStatusError as e:
    raise APIError("API request failed", status_code=e.response.status_code) from e

MiddlewareError

MiddlewareError(message: str, details: Optional[dict[str, Any]] = None)

Bases: Exception

Base class for every congressgov.services exception, so callers can catch all of them with a single except MiddlewareError.

Source code in src/congressgov/services/exceptions.py

def __init__(self, message: str, details: Optional[dict[str, Any]] = None):
    super().__init__(message)
    self.message = message
    self.details = details or {}

ClientNotFoundError

ClientNotFoundError(message: str = 'No client available for API operations', **kwargs: Any)

Bases: MiddlewareError

Raised when a service method needs an API client and none was provided or configured on the instance.

Example

service = Bill() # No client configured service.get(congress=118, bill_type="hr", bill_number=1) ClientNotFoundError: No client available. Provide client parameter or configure service instance.

Source code in src/congressgov/services/exceptions.py

def __init__(self, message: str = "No client available for API operations", **kwargs: Any) -> None:
    super().__init__(message, details=kwargs)

ValidationError

ValidationError(message: str, parameter: Optional[str] = None, value: Optional[Any] = None, constraint: Optional[str] = None, suggestions: Optional[list[str]] = None, valid_values: Optional[list[str]] = None, **kwargs: Any)

Bases: MiddlewareError

Raised for invalid input: missing/malformed/out-of-range parameters, or invalid parameter combinations. Carries suggestions and valid_values so the resulting message can point at likely typo fixes.

Example

service.get(bill_type="invalid") ValidationError: Invalid bill type: 'invalid' Did you mean: 'hr', 's'? Valid values: 'hr', 's', 'hjres', 'sjres', 'hconres', 'sconres', 'hres', 'sres'

Source code in src/congressgov/services/exceptions.py

def __init__(
    self,
    message: str,
    parameter: Optional[str] = None,
    value: Optional[Any] = None,
    constraint: Optional[str] = None,
    suggestions: Optional[list[str]] = None,
    valid_values: Optional[list[str]] = None,
    **kwargs: Any,
) -> None:
    details = kwargs
    if parameter:
        details['parameter'] = parameter
    if value is not None:
        details['value'] = value
    if constraint:
        details['constraint'] = constraint

    self.parameter = parameter
    self.value = value
    self.suggestions = suggestions or []
    self.valid_values = valid_values or []

    super().__init__(message, details=details)

APIError

APIError(message: str = 'API request failed', status_code: Optional[int] = None, url: Optional[str] = None, method: Optional[str] = None, response_body: Optional[str] = None, **kwargs: Any)

Bases: MiddlewareError

Raised when an API request fails, wrapping the HTTP status code, URL, method, and response body for debugging.

Example

try: ... response = client.get("/invalid/endpoint") ... except httpx.HTTPStatusError as e: ... raise APIError("API request failed", status_code=404) from e

Source code in src/congressgov/services/exceptions.py

def __init__(
    self,
    message: str = "API request failed",
    status_code: Optional[int] = None,
    url: Optional[str] = None,
    method: Optional[str] = None,
    response_body: Optional[str] = None,
    **kwargs: Any,
) -> None:
    details = kwargs
    if status_code:
        details['status_code'] = status_code
    if url:
        details['url'] = url
    if method:
        details['method'] = method
    if response_body:
        details['response_body'] = response_body
    super().__init__(message, details=details)

RateLimitError

RateLimitError(message: str = 'API rate limit exceeded', limit: Optional[int] = None, remaining: Optional[int] = None, reset_time: Optional[str] = None, retry_after: Optional[int] = None, **kwargs: Any)

Bases: APIError

Raised when the Congress.gov API rate limit is exceeded. Carries limit, remaining, reset_time, and retry_after for building retry logic.

Example

if rate_limit_exceeded: ... raise RateLimitError( ... "Rate limit exceeded", ... limit=1000, ... remaining=0, ... retry_after=60 ... )

Source code in src/congressgov/services/exceptions.py

def __init__(
    self,
    message: str = "API rate limit exceeded",
    limit: Optional[int] = None,
    remaining: Optional[int] = None,
    reset_time: Optional[str] = None,
    retry_after: Optional[int] = None,
    **kwargs: Any,
) -> None:
    if limit is not None:
        kwargs['limit'] = limit
    if remaining is not None:
        kwargs['remaining'] = remaining
    if reset_time:
        kwargs['reset_time'] = reset_time
    if retry_after:
        kwargs['retry_after'] = retry_after
    super().__init__(message, status_code=429, **kwargs)

TimeoutError

TimeoutError(message: str = 'API request timed out', timeout: Optional[float] = None, elapsed: Optional[float] = None, **kwargs: Any)

Bases: APIError

Raised when an API request times out. Carries timeout (the configured limit) and elapsed (how long it actually ran).

Example

try: ... response = client.get("/slow/endpoint", timeout=5.0) ... except httpx.TimeoutException as e: ... raise TimeoutError("Request timed out", timeout=5.0) from e

Source code in src/congressgov/services/exceptions.py

def __init__(
    self,
    message: str = "API request timed out",
    timeout: Optional[float] = None,
    elapsed: Optional[float] = None,
    **kwargs: Any,
) -> None:
    if timeout is not None:
        kwargs['timeout'] = timeout
    if elapsed is not None:
        kwargs['elapsed'] = elapsed
    super().__init__(message, status_code=408, **kwargs)

UnsupportedApiUrlError

UnsupportedApiUrlError(message: str, *, url: str | None = None)

Bases: MiddlewareError

Raised when a URL cannot be mapped to a known Congress.gov API route.

Source code in src/congressgov/services/exceptions.py

def __init__(self, message: str, *, url: str | None = None) -> None:
    self.url = url
    super().__init__(message)

RequestStoreError

RequestStoreError(message: str, *, request_key: str | None = None, operation: str | None = None, **kwargs: Any)

Bases: MiddlewareError

Raised when the request store cannot read or write persisted responses.

By default the store fails closed (no network fallback). Pass store_fail=True on the client or per-request to fall through to the API.

Source code in src/congressgov/services/exceptions.py

def __init__(
    self,
    message: str,
    *,
    request_key: str | None = None,
    operation: str | None = None,
    **kwargs: Any,
) -> None:
    details = dict(kwargs)
    if request_key:
        details["request_key"] = request_key
    if operation:
        details["operation"] = operation
    super().__init__(message, details=details)

ExpansionError

ExpansionError(message: str, attribute: Optional[str] = None, target_type: Optional[str] = None, failed_attributes: Optional[list[str]] = None, **kwargs: Any)

Bases: MiddlewareError

Raised when expanding a model's related attributes (fetching them from the API) fails, due to missing parameters, an invalid attribute name, or an API failure mid-expansion.

Example

try: ... expanded = service.expand(bill, attributes=["actions"]) ... except ExpansionError as e: ... print(f"Failed to expand: {e.attribute}")

Source code in src/congressgov/services/exceptions.py

def __init__(
    self,
    message: str,
    attribute: Optional[str] = None,
    target_type: Optional[str] = None,
    failed_attributes: Optional[list[str]] = None,
    **kwargs: Any,
) -> None:
    details = kwargs
    if attribute:
        details['attribute'] = attribute
    if target_type:
        details['target_type'] = target_type
    if failed_attributes:
        details['failed_attributes'] = failed_attributes
    super().__init__(message, details=details)

Workspace & storage

Congress.gov project workspace — shared root for all storage lanes.

WorkspacePaths dataclass

WorkspacePaths(root: Path)

Resolved paths under a single workspace root.

legacy_api_db

legacy_api_db() -> Path

Pre-workspace blob store path (.congressgov/request_store.db).

Source code in src/congressgov/services/core/workspace/paths.py

def legacy_api_db(self) -> Path:
    """Pre-workspace blob store path (``.congressgov/request_store.db``)."""
    return self.root / LEGACY_API_DB_NAME

Workspace

Workspace(paths: WorkspacePaths)

Single project workspace for API blobs, datasets, exports, and cache.

Example::

workspace = Workspace.open()
client = get_client_from_env(request_store=workspace.blob_connection())
graph = workspace.open_graph(118)

Source code in src/congressgov/services/core/workspace/__init__.py

def __init__(self, paths: WorkspacePaths) -> None:
    self.paths = paths

blob_connection

blob_connection(name: str = 'api') -> str

Resolve a blob-lane connection via the store registry (and env overrides).

Source code in src/congressgov/services/core/workspace/__init__.py

def blob_connection(self, name: str = "api") -> str:
    """Resolve a blob-lane connection via the store registry (and env overrides)."""
    import os

    if name == "api" and os.environ.get(CONGRESS_REQUEST_STORE_ENV):
        return resolve_blob_connection(workspace=self.paths)
    if name == "api":
        legacy = self.paths.legacy_api_db()
        if legacy.exists() and not self.paths.api_db.exists():
            return f"sqlite:///{legacy.as_posix()}"
    registry = self.registry()
    entry = registry.get_entry(name)
    if entry.lane.value != "blob":
        raise ValueError(f"Store {name!r} is lane {entry.lane.value!r}, not blob")
    return registry.resolve_backend_connection(entry)

write_export

write_export(name: str, content: bytes | str, *, metadata: dict[str, object] | None = None) -> str

Write a generated artifact to the workspace exports lane.

Source code in src/congressgov/services/core/workspace/__init__.py

def write_export(
    self,
    name: str,
    content: bytes | str,
    *,
    metadata: dict[str, object] | None = None,
) -> str:
    """Write a generated artifact to the workspace exports lane."""
    return self.registry().artifact("exports").write(
        name,
        content,
        metadata=metadata,
    )

default_workspace_root

default_workspace_root() -> Path

Return the workspace root from CONGRESS_WORKSPACE or ./.congressgov.

Source code in src/congressgov/services/core/workspace/paths.py

def default_workspace_root() -> Path:
    """Return the workspace root from ``CONGRESS_WORKSPACE`` or ``./.congressgov``."""
    return Path(os.environ.get(CONGRESS_WORKSPACE_ENV, DEFAULT_WORKSPACE_NAME)).expanduser().resolve()

maybe_log_legacy_migration

maybe_log_legacy_migration(paths: WorkspacePaths) -> None

Migrate or log when the legacy flat DB exists but the workspace API path does not.

Source code in src/congressgov/services/core/workspace/resolve.py

def maybe_log_legacy_migration(paths: WorkspacePaths) -> None:
    """Migrate or log when the legacy flat DB exists but the workspace API path does not."""
    if migrate_legacy_blob_if_needed(paths):
        return
    legacy = paths.legacy_api_db()
    if legacy.exists() and not paths.api_db.exists():
        logger.info(
            "Legacy request store found at %s. "
            "Future releases prefer %s under the workspace layout.",
            legacy,
            paths.api_db,
        )

migrate_legacy_blob_if_needed

migrate_legacy_blob_if_needed(paths: WorkspacePaths) -> bool

Copy a legacy flat request_store.db into api/request_store.db.

Returns True when a migration was performed.

Source code in src/congressgov/services/core/workspace/resolve.py

def migrate_legacy_blob_if_needed(paths: WorkspacePaths) -> bool:
    """
    Copy a legacy flat ``request_store.db`` into ``api/request_store.db``.

    Returns True when a migration was performed.
    """
    legacy = paths.legacy_api_db()
    if not legacy.exists() or paths.api_db.exists():
        return False
    paths.api_dir.mkdir(parents=True, exist_ok=True)
    shutil.copy2(legacy, paths.api_db)
    logger.info(
        "Migrated legacy request store from %s to %s",
        legacy,
        paths.api_db,
    )
    return True

resolve_blob_connection

resolve_blob_connection(*, workspace: WorkspacePaths | None = None, override: str | None = None) -> str

Return the blob-lane connection string.

Precedence: 1. Explicit override or CONGRESS_REQUEST_STORE env var 2. Workspace API database (api/request_store.db) 3. Legacy flat .congressgov/request_store.db when it exists

Source code in src/congressgov/services/core/workspace/resolve.py

def resolve_blob_connection(
    *,
    workspace: WorkspacePaths | None = None,
    override: str | None = None,
) -> str:
    """
    Return the blob-lane connection string.

    Precedence:
    1. Explicit *override* or ``CONGRESS_REQUEST_STORE`` env var
    2. Workspace API database (``api/request_store.db``)
    3. Legacy flat ``.congressgov/request_store.db`` when it exists
    """
    explicit = override or os.environ.get(CONGRESS_REQUEST_STORE_ENV)
    if explicit:
        return explicit

    paths = workspace or resolve_workspace()
    if paths.api_db.exists():
        return paths.api_db_url

    legacy = paths.legacy_api_db()
    if legacy.exists():
        logger.info(
            "Using legacy request store at %s. "
            "New workspaces use %s (set CONGRESS_WORKSPACE to relocate).",
            legacy,
            paths.api_db,
        )
        return f"sqlite:///{legacy.as_posix()}"

    return paths.api_db_url

resolve_workspace

resolve_workspace(root: str | Path | None = None) -> WorkspacePaths

Build :class:WorkspacePaths for root or the default workspace.

Source code in src/congressgov/services/core/workspace/resolve.py

def resolve_workspace(root: str | Path | None = None) -> WorkspacePaths:
    """Build :class:`WorkspacePaths` for *root* or the default workspace."""
    if root is None:
        resolved = default_workspace_root()
    else:
        resolved = Path(root).expanduser().resolve()
    return WorkspacePaths(root=resolved)

Request store

Persistent request store for Congress.gov API deduplication.

Every distinct API request is stored on first fetch. Repeat calls with the same canonical key are served from storage unless force_fetch=True.

Quick start::

from congressgov._client import AuthenticatedClient
from congressgov.services.core.request_store import RequestStore, attach_request_store

client = AuthenticatedClient(base_url="https://api.congress.gov/v3", token=API_KEY)
store = RequestStore.open("sqlite:///.congressgov/request_store.db")
attach_request_store(client, store)

Backends: - :memory: — in-memory (tests) - sqlite:///path/to/store.db — persistent SQLite (default)

FileRequestStoreBackend

FileRequestStoreBackend(directory: str | Path)

Persist responses as JSON files under a directory (stdlib only).

Source code in src/congressgov/services/core/request_store/backends/file.py

def __init__(self, directory: str | Path) -> None:
    self.directory = Path(directory)
    self.directory.mkdir(parents=True, exist_ok=True)
    self._lock = threading.RLock()

MemoryRequestStoreBackend

MemoryRequestStoreBackend()

Thread-safe in-memory backend implementing :class:RequestStoreBackend.

Source code in src/congressgov/services/core/request_store/backends/memory.py

def __init__(self) -> None:
    self._data: dict[str, StoredResponse] = {}
    self._lock = threading.RLock()
    self._stats = {"gets": 0, "puts": 0, "hits": 0, "misses": 0}

RequestStoreBackend

Bases: Protocol

Canonical persistence interface for raw API responses.

Implementations may use SQLite, files, Redis, or other engines while exposing the same lookup/store contract.

get

get(key: str) -> StoredResponse | None

Load a stored response by canonical request key.

Source code in src/congressgov/services/core/request_store/backends/base.py

def get(self, key: str) -> StoredResponse | None:
    """Load a stored response by canonical request key."""
    ...

put

put(key: str, response: StoredResponse) -> None

Persist a response under key.

Source code in src/congressgov/services/core/request_store/backends/base.py

def put(self, key: str, response: StoredResponse) -> None:
    """Persist a response under *key*."""
    ...

delete

delete(key: str) -> bool

Remove a stored response. Returns True when a record existed.

Source code in src/congressgov/services/core/request_store/backends/base.py

def delete(self, key: str) -> bool:
    """Remove a stored response. Returns True when a record existed."""
    ...

contains

contains(key: str) -> bool

Return True when key exists in the backend.

Source code in src/congressgov/services/core/request_store/backends/base.py

def contains(self, key: str) -> bool:
    """Return True when *key* exists in the backend."""
    ...

count

count() -> int

Return total stored records.

Source code in src/congressgov/services/core/request_store/backends/base.py

def count(self) -> int:
    """Return total stored records."""
    ...

close

close() -> None

Release backend resources.

Source code in src/congressgov/services/core/request_store/backends/base.py

def close(self) -> None:
    """Release backend resources."""
    ...

get_stats

get_stats() -> dict[str, Any]

Return backend-specific statistics.

Source code in src/congressgov/services/core/request_store/backends/base.py

def get_stats(self) -> dict[str, Any]:
    """Return backend-specific statistics."""
    ...

SQLiteRequestStoreBackend

SQLiteRequestStoreBackend(connection: str | Path)

Persistent SQLite backend implementing :class:RequestStoreBackend.

Source code in src/congressgov/services/core/request_store/backends/sqlite.py

def __init__(self, connection: str | Path) -> None:
    path = str(connection)
    if path.startswith("sqlite:///"):
        path = path[len("sqlite:///") :]
    self._path = Path(path)
    self._path.parent.mkdir(parents=True, exist_ok=True)
    self._lock = threading.RLock()
    self._conn = sqlite3.connect(self._path, check_same_thread=False)
    self._conn.row_factory = sqlite3.Row
    with self._lock:
        apply_migrations(self._conn)

RequestKey dataclass

RequestKey(method: str, url: str, query: tuple[tuple[str, str], ...] = ())

Canonical identity for an API request.

digest

digest() -> str

Stable string used as the storage backend key.

Source code in src/congressgov/services/core/request_store/models.py

def digest(self) -> str:
    """Stable string used as the storage backend key."""
    query_part = "&".join(f"{k}={v}" for k, v in self.query)
    return f"{self.method.upper()}:{self.url}?{query_part}"

StoredResponse dataclass

StoredResponse(status_code: int, headers: dict[str, str], body: bytes, fetched_at: datetime = (lambda: datetime.now(timezone.utc))(), policy: str = 'moderate', request_key: str | None = None, source_updated_at: datetime | None = None)

Raw HTTP response persisted for a :class:RequestKey.

effective_updated_at property

effective_updated_at: datetime

Best available timestamp for change-detection/display purposes.

Note: TTL/staleness math (:func:request_store.policy.estimate_staleness) deliberately uses fetched_at instead, since source_updated_at reflects the API's own (often old) updateDate and would make freshly cached responses look immediately stale.

ChangePolicy

Bases: str, Enum

How quickly a stored response is considered stale.

PolicyConfig dataclass

PolicyConfig(default_policy: ChangePolicy = ChangePolicy.MODERATE, rules: tuple[PolicyRule, ...] = (lambda: DEFAULT_POLICY_RULES)(), overrides: dict[str, ChangePolicy] = dict(), closed_congress_permanent: bool = True)

Resolved policy configuration for the request store.

PolicyRule dataclass

PolicyRule(pattern: Pattern[str], policy: ChangePolicy)

Map URL path patterns to a :class:ChangePolicy.

StalenessEstimate dataclass

StalenessEstimate(policy: ChangePolicy, fetched_at: datetime, ttl_seconds: int | None, age_seconds: float, remaining_seconds: float | None, is_stale: bool, refresh_after: datetime | None, change_likelihood: float)

Estimated freshness of a stored response based on fetch time and policy.

likely_changed property

likely_changed: bool

True when change likelihood exceeds 50%.

RequestStore

RequestStore(backend: RequestStoreBackend | str | None = None, *, config: StoreConfig | None = None, workspace_root: Path | None = None)

Persistent deduplicating store for raw API responses.

Network calls for the same canonical request are blocked when a fresh stored response exists. Use force_fetch=True to bypass the store.

Source code in src/congressgov/services/core/request_store/store.py

def __init__(
    self,
    backend: RequestStoreBackend | str | None = None,
    *,
    config: StoreConfig | None = None,
    workspace_root: Path | None = None,
) -> None:
    from pathlib import Path as _Path

    if backend is None or isinstance(backend, str):
        root = workspace_root
        if root is None:
            from congressgov.services.core.workspace import resolve_workspace

            root = resolve_workspace().root
        self.backend = create_backend(backend, workspace_root=_Path(root))
    else:
        self.backend = backend
    self.config = config or _default_store_config()
    self._sync_gate = _SyncInFlightGate()
    self._async_gate = _AsyncInFlightGate()
    self._stats = {
        "hits": 0,
        "misses": 0,
        "stores": 0,
        "store_errors": 0,
        "inflight_waits": 0,
    }
    self._stats_lock = threading.Lock()

get_sync

get_sync(request, *, force_fetch: bool = False, store_fail: bool | None = None) -> tuple[StoredResponse | None, bool]

Lookup a stored response for a sync httpx request.

Returns (stored_response, is_leader). Followers should wait on the in-flight gate before calling again.

Source code in src/congressgov/services/core/request_store/store.py

def get_sync(
    self,
    request,
    *,
    force_fetch: bool = False,
    store_fail: bool | None = None,
) -> tuple[StoredResponse | None, bool]:
    """
    Lookup a stored response for a sync httpx request.

    Returns ``(stored_response, is_leader)``. Followers should wait on the
    in-flight gate before calling again.
    """
    effective_store_fail = (
        self.config.store_fail if store_fail is None else store_fail
    )
    key_obj = request_key_from_httpx(request)
    key = key_obj.digest()
    is_leader = self._sync_gate.leader_or_wait(key)
    if not is_leader:
        self._record("inflight_waits")
        stored = self._lookup(
            key,
            path=key_obj.url,
            force_fetch=force_fetch,
            store_fail=effective_store_fail,
        )
        return stored, False
    stored = self._lookup(
        key,
        path=key_obj.url,
        force_fetch=force_fetch,
        store_fail=effective_store_fail,
    )
    return stored, True

store

store(request, response, *, store_fail: bool | None = None) -> None

Persist a successful API response.

Source code in src/congressgov/services/core/request_store/store.py

def store(
    self,
    request,
    response,
    *,
    store_fail: bool | None = None,
) -> None:
    """Persist a successful API response."""
    if not self.config.enabled:
        return
    if response.status_code != 200:
        return
    effective_store_fail = (
        self.config.store_fail if store_fail is None else store_fail
    )
    key_obj = request_key_from_httpx(request)
    key = key_obj.digest()
    policy = self.resolve_policy(key_obj.url)
    body = self._response_body_bytes(response)
    stored = StoredResponse(
        status_code=response.status_code,
        headers=normalize_stored_response_headers(
            {key: value for key, value in response.headers.items()}
        ),
        body=body,
        policy=policy.value,
        request_key=key,
        source_updated_at=extract_source_updated_at(body),
    )
    try:
        self.backend.put(key, stored)
        self._record("stores")
    except Exception as exc:
        self._record("store_errors")
        if effective_store_fail:
            return
        raise RequestStoreError(
            "Request store write failed",
            request_key=key,
            operation="put",
        ) from exc

open classmethod

open(connection: str | None = None, **kwargs) -> RequestStore

Open the workspace default on-disk store when connection is omitted.

Source code in src/congressgov/services/core/request_store/store.py

@classmethod
def open(cls, connection: str | None = None, **kwargs) -> RequestStore:
    """Open the workspace default on-disk store when *connection* is omitted."""
    if connection is None:
        from congressgov.services.core.workspace import resolve_blob_connection

        connection = resolve_blob_connection()
    return cls(connection, **kwargs)

StoreConfig dataclass

StoreConfig(enabled: bool = True, store_fail: bool = False, policy: PolicyConfig = load_policy_config())

Runtime configuration for :class:RequestStore.

AsyncOfflineTransport

AsyncOfflineTransport(store: RequestStore)

Bases: AsyncBaseTransport

Async offline-only transport backed by the request store.

Source code in src/congressgov/services/core/request_store/transport.py

def __init__(self, store: RequestStore) -> None:
    self._store = store

AsyncStoringTransport

AsyncStoringTransport(transport: AsyncBaseTransport, store: RequestStore)

Bases: AsyncBaseTransport

Async transport wrapper that reads/writes through a :class:RequestStore.

Source code in src/congressgov/services/core/request_store/transport.py

def __init__(
    self,
    transport: httpx.AsyncBaseTransport,
    store: RequestStore,
) -> None:
    self._transport = transport
    self._store = store

OfflineTransport

OfflineTransport(store: RequestStore)

Bases: BaseTransport

Serve API responses from the request store without network access.

Source code in src/congressgov/services/core/request_store/transport.py

def __init__(self, store: RequestStore) -> None:
    self._store = store

StoringTransport

StoringTransport(transport: BaseTransport, store: RequestStore)

Bases: BaseTransport

Sync transport wrapper that reads/writes through a :class:RequestStore.

Source code in src/congressgov/services/core/request_store/transport.py

def __init__(
    self,
    transport: httpx.BaseTransport,
    store: RequestStore,
) -> None:
    self._transport = transport
    self._store = store

attach_offline_store

attach_offline_store(client: Any, store: RequestStore | str | None = None, *, config: StoreConfig | None = None) -> Any

Attach store in offline-only mode (no network requests).

Source code in src/congressgov/services/core/request_store/client.py

def attach_offline_store(
    client: Any,
    store: RequestStore | str | None = None,
    *,
    config: StoreConfig | None = None,
) -> Any:
    """Attach *store* in offline-only mode (no network requests)."""
    if isinstance(store, str) or store is None:
        request_store = RequestStore(store, config=config or StoreConfig())
    else:
        request_store = store
        if config is not None:
            request_store.config = config

    _CLIENT_STORES[id(client)] = request_store
    client.set_httpx_client(_build_offline_sync_client(client, request_store))
    client.set_async_httpx_client(_build_offline_async_client(client, request_store))
    return client

attach_request_store

attach_request_store(client: Any, store: RequestStore | str | None = None, *, config: StoreConfig | None = None, enabled: bool = True, store_fail: bool = False) -> Any

Wrap client's httpx transport so all API calls pass through store.

Works with AuthenticatedClient and Client from congressgov._client. Call before any API traffic, or after set_httpx_client / set_async_httpx_client.

Parameters:

Name	Type	Description	Default
client	Any	API client instance.	required
store	RequestStore | str | None	:class:RequestStore, connection string, or None for default SQLite.	None
config	StoreConfig | None	Optional store configuration merged with store when provided.	None
enabled	bool	When False, register store metadata without wrapping transport.	True
store_fail	bool	Fall through to network when store read/write fails.	False

Returns:

Type	Description
Any	The same client instance (mutated in place).

Source code in src/congressgov/services/core/request_store/client.py

def attach_request_store(
    client: Any,
    store: RequestStore | str | None = None,
    *,
    config: StoreConfig | None = None,
    enabled: bool = True,
    store_fail: bool = False,
) -> Any:
    """
    Wrap *client*'s httpx transport so all API calls pass through *store*.

    Works with ``AuthenticatedClient`` and ``Client`` from ``congressgov._client``.
    Call before any API traffic, or after ``set_httpx_client`` / ``set_async_httpx_client``.

    Args:
        client: API client instance.
        store: :class:`RequestStore`, connection string, or ``None`` for default SQLite.
        config: Optional store configuration merged with *store* when provided.
        enabled: When False, register store metadata without wrapping transport.
        store_fail: Fall through to network when store read/write fails.

    Returns:
        The same *client* instance (mutated in place).
    """
    if isinstance(store, str) or store is None:
        merged_config = config or StoreConfig(enabled=enabled, store_fail=store_fail)
        request_store = RequestStore(store, config=merged_config)
    elif config is not None:
        request_store = store
        request_store.config = config
        request_store.config.enabled = enabled
        request_store.config.store_fail = store_fail
    else:
        request_store = store
        request_store.config.enabled = enabled
        request_store.config.store_fail = store_fail

    _CLIENT_STORES[id(client)] = request_store

    if not enabled:
        return client

    if client._client is not None:
        _install_sync_transport(client._client, request_store)
    else:
        client.set_httpx_client(_build_sync_httpx_client(client, request_store))

    if client._async_client is not None:
        _install_async_transport(client._async_client, request_store)
    else:
        client.set_async_httpx_client(_build_async_httpx_client(client, request_store))

    return client

build_stored_client

build_stored_client(client_factory: Callable[[], Any], store: RequestStore | str | None = None, **attach_kwargs: Any) -> Any

Create a client via client_factory and attach a request store.

Source code in src/congressgov/services/core/request_store/client.py

def build_stored_client(
    client_factory: Callable[[], Any],
    store: RequestStore | str | None = None,
    **attach_kwargs: Any,
) -> Any:
    """Create a client via *client_factory* and attach a request store."""
    client = client_factory()
    return attach_request_store(client, store, **attach_kwargs)

fetch_options

fetch_options(*, force_fetch: bool = False, store_fail: bool | None = None) -> Iterator[None]

Temporarily override fetch behavior for nested API calls.

Example

with fetch_options(force_fetch=True): bill = bill_service.get(congress=118, bill_type="hr", bill_number=1)

Source code in src/congressgov/services/core/request_store/context.py

@contextmanager
def fetch_options(
    *,
    force_fetch: bool = False,
    store_fail: bool | None = None,
) -> Iterator[None]:
    """
    Temporarily override fetch behavior for nested API calls.

    Example:
        with fetch_options(force_fetch=True):
            bill = bill_service.get(congress=118, bill_type="hr", bill_number=1)
    """
    force_token = _force_fetch.set(force_fetch)
    fail_token = _store_fail.set(store_fail)
    try:
        yield
    finally:
        _force_fetch.reset(force_token)
        _store_fail.reset(fail_token)

create_backend

create_backend(connection: Union[str, Path, None] = None, *, workspace_root: Path | None = None) -> RequestStoreBackend

Create a request-store backend from a connection string.

Supported forms: - None with a workspace_root → the workspace's persistent SQLite blob store (<workspace_root>/api/request_store.db), matching :meth:RequestStore.open - None with no workspace_root or ":memory:" → in-memory backend - sqlite:///path/to/store.db → SQLite file - filesystem path ending in .db → SQLite file - file:///path/to/dir or directory path → JSON file per entry - redis://host:port/db → Redis (requires redis package)

Source code in src/congressgov/services/core/request_store/factory.py

def create_backend(
    connection: Union[str, Path, None] = None,
    *,
    workspace_root: Path | None = None,
) -> RequestStoreBackend:
    """
    Create a request-store backend from a connection string.

    Supported forms:
    - ``None`` with a *workspace_root* → the workspace's persistent SQLite blob
      store (``<workspace_root>/api/request_store.db``), matching
      :meth:`RequestStore.open`
    - ``None`` with no *workspace_root* or ``":memory:"`` → in-memory backend
    - ``sqlite:///path/to/store.db`` → SQLite file
    - filesystem path ending in ``.db`` → SQLite file
    - ``file:///path/to/dir`` or directory path → JSON file per entry
    - ``redis://host:port/db`` → Redis (requires ``redis`` package)
    """
    if connection is None:
        if workspace_root is not None:
            from congressgov.services.core.workspace import WorkspacePaths, resolve_blob_connection

            connection = resolve_blob_connection(workspace=WorkspacePaths(root=workspace_root))
        else:
            connection = ":memory:"
    return _create_backend(str(connection), workspace_root=workspace_root)

print_store_stats

print_store_stats(store: RequestStore) -> None

Print a human-readable summary of store activity.

Source code in src/congressgov/services/core/request_store/introspection.py

def print_store_stats(store: RequestStore) -> None:
    """Print a human-readable summary of store activity."""
    summary = summarize_store(store)
    backend = summary.get("backend", {})
    print("Request store stats:")
    print(f"  hits: {summary.get('hits', 0)}  misses: {summary.get('misses', 0)}")
    print(f"  hit rate: {summary.get('hit_rate', 0.0):.1%}")
    print(f"  records: {backend.get('records', summary.get('records', '?'))}")
    if "stale_records" in summary:
        print(f"  stale records: {summary['stale_records']}")
    if backend.get("path"):
        print(f"  path: {backend['path']}")
    if summary.get("keys_sample"):
        print("  recent keys:")
        for key in summary["keys_sample"]:
            print(f"    - {key}")

summarize_store

summarize_store(store: RequestStore) -> dict[str, Any]

Return combined runtime and backend statistics.

Source code in src/congressgov/services/core/request_store/introspection.py

def summarize_store(store: RequestStore) -> dict[str, Any]:
    """Return combined runtime and backend statistics."""
    summary = store.get_stats()
    summary["keys_sample"] = list_keys(store, limit=5)
    stale_count = 0
    iter_fn = getattr(store.backend, "iter_records", None)
    if iter_fn is not None:
        for record in iter_fn():
            path = (record.request_key or "").split(":", 1)[-1].split("?", 1)[0]
            policy = store.config.policy.resolve(path)
            estimate = estimate_staleness(record, policy=policy)
            if estimate.is_stale:
                stale_count += 1
        summary["stale_records"] = stale_count
    return summary

normalize_query

normalize_query(params: dict[str, Any] | list[tuple[str, Any]]) -> tuple[tuple[str, str], ...]

Return sorted, normalized query pairs for key generation.

Source code in src/congressgov/services/core/request_store/keys.py

def normalize_query(params: dict[str, Any] | list[tuple[str, Any]]) -> tuple[tuple[str, str], ...]:
    """Return sorted, normalized query pairs for key generation."""
    if isinstance(params, dict):
        items = params.items()
    else:
        items = params
    normalized: list[tuple[str, str]] = []
    for key, value in items:
        if key in _STRIP_PARAMS or value is None:
            continue
        normalized.append((str(key), str(value)))
    return tuple(sorted(normalized))

request_key_digest

request_key_digest(request: Request) -> str

Digest string for storage lookup.

Source code in src/congressgov/services/core/request_store/keys.py

def request_key_digest(request: httpx.Request) -> str:
    """Digest string for storage lookup."""
    return request_key_from_httpx(request).digest()

request_key_from_httpx

request_key_from_httpx(request: Request) -> RequestKey

Build a :class:RequestKey from an outgoing httpx request.

Source code in src/congressgov/services/core/request_store/keys.py

def request_key_from_httpx(request: httpx.Request) -> RequestKey:
    """Build a :class:`RequestKey` from an outgoing httpx request."""
    parsed = urlparse(str(request.url))
    path = parsed.path or "/"
    query = normalize_query(parse_qsl(parsed.query, keep_blank_values=True))
    return RequestKey(method=request.method, url=path, query=query)

current_congress

current_congress(*, year: int | None = None) -> int

Estimate the current Congress number from the calendar year.

Source code in src/congressgov/services/core/request_store/policy.py

def current_congress(*, year: int | None = None) -> int:
    """Estimate the current Congress number from the calendar year."""
    moment = year if year is not None else datetime.now(timezone.utc).year
    return (moment - 1789) // 2 + 1

estimate_refresh_after

estimate_refresh_after(stored: StoredResponse, *, policy: ChangePolicy | None = None) -> datetime | None

Estimate when a stored entry becomes eligible for automatic refresh.

Returns None for permanent entries or when already stale.

Source code in src/congressgov/services/core/request_store/policy.py

def estimate_refresh_after(
    stored: StoredResponse,
    *,
    policy: ChangePolicy | None = None,
) -> datetime | None:
    """
    Estimate when a stored entry becomes eligible for automatic refresh.

    Returns ``None`` for permanent entries or when already stale.
    """
    return estimate_staleness(stored, policy=policy).refresh_after

estimate_staleness

estimate_staleness(stored: StoredResponse, *, policy: ChangePolicy | None = None, now: datetime | None = None) -> StalenessEstimate

Estimate how likely a stored response has changed since it was fetched.

change_likelihood ramps from 0.0 at fetch time to 1.0 at TTL expiry (sigmoid curve). PERMANENT entries always report 0.0 unless already forced stale by external logic.

Source code in src/congressgov/services/core/request_store/policy.py

def estimate_staleness(
    stored: StoredResponse,
    *,
    policy: ChangePolicy | None = None,
    now: datetime | None = None,
) -> StalenessEstimate:
    """
    Estimate how likely a stored response has changed since it was fetched.

    ``change_likelihood`` ramps from 0.0 at fetch time to 1.0 at TTL expiry
    (sigmoid curve). ``PERMANENT`` entries always report 0.0 unless already
    forced stale by external logic.
    """
    resolved = policy or ChangePolicy(stored.policy)
    ttl = POLICY_TTLS.get(resolved)
    moment = now or datetime.now(timezone.utc)
    # TTL/age is always anchored to when *we* fetched the response, not the API's
    # own `updateDate` (`source_updated_at`), which is frequently a stale historical
    # timestamp on the source record and would otherwise make freshly stored
    # responses look immediately expired. `source_updated_at` remains available on
    # `StoredResponse` for change-detection use cases, just not for TTL math.
    fetched = stored.fetched_at
    if fetched.tzinfo is None:
        fetched = fetched.replace(tzinfo=timezone.utc)
    age_seconds = max(0.0, (moment - fetched).total_seconds())

    if ttl is None:
        return StalenessEstimate(
            policy=resolved,
            fetched_at=fetched,
            ttl_seconds=None,
            age_seconds=age_seconds,
            remaining_seconds=None,
            is_stale=False,
            refresh_after=None,
            change_likelihood=0.0,
        )

    remaining = max(0.0, ttl - age_seconds)
    stale = age_seconds > ttl
    progress = min(1.0, age_seconds / ttl) if ttl else 0.0
    # Smooth likelihood curve — low early, rises sharply near TTL.
    likelihood = progress ** 2

    refresh_after = None
    if not stale:
        refresh_after = datetime.fromtimestamp(fetched.timestamp() + ttl, tz=timezone.utc)

    return StalenessEstimate(
        policy=resolved,
        fetched_at=fetched,
        ttl_seconds=ttl,
        age_seconds=age_seconds,
        remaining_seconds=remaining if not stale else 0.0,
        is_stale=stale,
        refresh_after=refresh_after,
        change_likelihood=likelihood if not stale else 1.0,
    )

is_stale

is_stale(stored: StoredResponse, *, now: datetime | None = None, policy: ChangePolicy | None = None) -> bool

Return True when a stored response should be refreshed automatically.

PERMANENT entries never go stale unless force_fetch is used.

Source code in src/congressgov/services/core/request_store/policy.py

def is_stale(
    stored: StoredResponse,
    *,
    now: datetime | None = None,
    policy: ChangePolicy | None = None,
) -> bool:
    """
    Return True when a stored response should be refreshed automatically.

    ``PERMANENT`` entries never go stale unless ``force_fetch`` is used.
    """
    return estimate_staleness(stored, policy=policy, now=now).is_stale

load_policy_config

load_policy_config(path: str | Path | None = None) -> PolicyConfig

Load :class:PolicyConfig from path or the bundled default JSON file.

Supports .json, .yaml, and .yml extensions.

Source code in src/congressgov/services/core/request_store/policy_loader.py

def load_policy_config(path: str | Path | None = None) -> PolicyConfig:
    """
    Load :class:`PolicyConfig` from *path* or the bundled default JSON file.

    Supports ``.json``, ``.yaml``, and ``.yml`` extensions.
    """
    policy_path = Path(path) if path is not None else _DEFAULT_POLICY_FILE
    return policy_config_from_dict(_load_raw(policy_path))