API Reference

Complete reference for all fastapi-topaz public exports.

---

Configuration

TopazConfig

Configuration for Topaz authorization. Create once at app startup, use to generate authorization dependencies.

PARAMETER	DESCRIPTION
authorizer_options	Connection settings for Topaz authorizer TYPE: AuthorizerOptions
policy_path_root	Root package name for policies (e.g., "myapp") TYPE: str
identity_provider	Function to extract user identity from request TYPE: Callable[[Request], Identity]
policy_instance_name	Name of policy instance to evaluate TYPE: str
policy_instance_label	Label for policy instance (defaults to name) TYPE: str | None DEFAULT: None
resource_context_provider	Function to provide additional context TYPE: Callable[[Request], ResourceContext] | None DEFAULT: None
policy_path_normalizer	Optional callable to transform generated policy paths (e.g., replace hyphens with underscores for valid Rego identifiers) TYPE: Callable[[str], str] | None DEFAULT: None
default_policy	Optional fallback policy path evaluated when no explicit policy file exists and no policy group matches. All requests still go through Topaz — this only changes which policy is evaluated. TYPE: str | None DEFAULT: None
policy_groups	Optional ordered list of :class:PolicyGroup entries. The first group whose url_pattern matches the route template wins. TYPE: list[PolicyGroup] | None DEFAULT: None
decision_cache	Optional cache for authorization decisions TYPE: CacheBackend | None DEFAULT: None
max_concurrent_checks	Max concurrent authorization checks for bulk operations (default: 10) TYPE: int DEFAULT: 10
check_timeout	gRPC deadline in seconds applied to each authorization call (default: 5.0). Set to None to disable the deadline. TYPE: float | None DEFAULT: 5.0
on_error	Response when the authorization call itself fails and the circuit breaker gives no fallback decision, for the middleware and every dependency. Both values fail closed: "deny" (default): 403 Forbidden, the same as a denial. "unavailable": 503 Service Unavailable, with Retry-After set to the circuit breaker's recovery_timeout when one is configured. Clients and monitoring can tell an outage from a denial, but proxies and HTTP clients may retry 503s. TopazMiddleware(on_error=...) overrides it for the middleware. TYPE: OnError DEFAULT: 'deny'
circuit_breaker	Optional circuit breaker for graceful degradation TYPE: CircuitBreaker | None DEFAULT: None
audit_logger	Optional audit logger for authorization decisions TYPE: AuditLogger | None DEFAULT: None
metrics	Optional Prometheus metrics collector. When combined with circuit_breaker and no user-provided on_state_change callback, circuit transitions and the state gauge are recorded automatically (a user-provided callback takes precedence). TYPE: PrometheusMetrics | None DEFAULT: None
tracing	Optional OpenTelemetry tracing TYPE: OTelTracing | None DEFAULT: None
insecure	Use a plaintext (non-TLS) gRPC channel to the authorizer. Local development only — never enable against a production Topaz. TYPE: bool DEFAULT: False
expose_deny_reason	Include the evaluated policy path and check source in 403 response bodies. This leaks policy structure to clients — enable for development and debugging only, never in production. TYPE: bool DEFAULT: False

Source code in src/fastapi_topaz/config.py

def __init__(
    self,
    *,
    authorizer_options: AuthorizerOptions,
    policy_path_root: str,
    identity_provider: Callable[[Request], Identity],
    policy_instance_name: str,
    policy_instance_label: str | None = None,
    resource_context_provider: Callable[[Request], ResourceContext] | None = None,
    policy_path_normalizer: Callable[[str], str] | None = None,
    default_policy: str | None = None,
    policy_groups: list[PolicyGroup] | None = None,
    decision_cache: CacheBackend | None = None,
    max_concurrent_checks: int = 10,
    check_timeout: float | None = 5.0,
    on_error: OnError = "deny",
    circuit_breaker: CircuitBreaker | None = None,
    audit_logger: AuditLogger | None = None,
    metrics: PrometheusMetrics | None = None,
    tracing: OTelTracing | None = None,
    insecure: bool = False,
    expose_deny_reason: bool = False,
):
    self.authorizer_options = authorizer_options
    self.policy_path_root = policy_path_root
    self.identity_provider = identity_provider
    self.policy_instance_name = policy_instance_name
    self.policy_instance_label = policy_instance_label or policy_instance_name
    self.resource_context_provider = resource_context_provider
    self.policy_path_normalizer = policy_path_normalizer
    self.default_policy = default_policy  # validated by the property setter
    self.policy_groups = policy_groups or []  # validated by the property setter
    self.decision_cache = decision_cache
    if max_concurrent_checks < 1:
        raise ValueError(f"max_concurrent_checks must be >= 1, got {max_concurrent_checks}")
    self.max_concurrent_checks = max_concurrent_checks
    self.check_timeout = check_timeout
    self.on_error = on_error  # validated by the property setter
    self.circuit_breaker = circuit_breaker
    self.audit_logger = audit_logger
    self.metrics = metrics
    self.tracing = tracing
    self.insecure = insecure
    self.expose_deny_reason = expose_deny_reason
    self._authorizer = SharedAuthorizerClient(authorizer_options, insecure=insecure)
    # asyncio primitives are created lazily on first use: on Python 3.9
    # they bind the event loop active at creation time, and configs are
    # typically created at module import, outside any loop
    self._semaphore: asyncio.Semaphore | None = None
    # Stale cache for circuit breaker fallback (stores entries beyond normal TTL)
    self._stale_cache: dict[str, tuple[bool, float]] = {}
    self._stale_cache_lock: asyncio.Lock | None = None

    # Configure connection pool with authorizer options

    # Auto-wire circuit breaker metrics: record transitions and the state
    # gauge unless the user installed their own on_state_change callback
    if self.metrics and self.circuit_breaker and self.circuit_breaker.on_state_change is None:
        state_values = {"closed": 0, "open": 1, "half_open": 2}
        metrics = self.metrics

        def _record_circuit_metrics(old_state: str, new_state: str, reason: str) -> None:
            metrics.record_circuit_transition(old_state, new_state)
            metrics.set_circuit_state(state_values.get(new_state, 0))

        self.circuit_breaker.on_state_change = _record_circuit_metrics

check_decision async

check_decision(request: Request, policy_path: str, decision: str, resource_context: ResourceContext | None = None, source: str = 'dependency', policy_resolution_source: str | None = None) -> bool

Check an authorization decision, using cache if available.

This is the core authorization check method that handles caching, circuit breaker logic, and can be used directly for custom authorization logic.

When audit_logger is configured, every call emits exactly one audit event, regardless of caller (middleware, dependency, or manual API).

Source code in src/fastapi_topaz/config.py

async def check_decision(
    self,
    request: Request,
    policy_path: str,
    decision: str,
    resource_context: ResourceContext | None = None,
    source: str = "dependency",
    policy_resolution_source: str | None = None,
) -> bool:
    """
    Check an authorization decision, using cache if available.

    This is the core authorization check method that handles caching,
    circuit breaker logic, and can be used directly for custom authorization logic.

    When ``audit_logger`` is configured, every call emits exactly one audit
    event, regardless of caller (middleware, dependency, or manual API).
    """
    identity = self.identity_provider(request)
    start_time = time.monotonic()
    cached_result = False
    result: bool = False
    span = None
    check_error: Exception | None = None

    # Start tracing span
    if self.tracing:
        span = self.tracing.start_auth_span(
            source=source,
            check_type="policy",
            policy_path=policy_path,
            identity_value=identity.value if identity else None,
        )

    try:
        # Check fresh cache first
        identity_value = identity.value or ""
        scope = self._cache_scope(identity)
        if self.decision_cache:
            cached = await self.decision_cache.get(
                identity_value, policy_path, decision, resource_context, **scope
            )
            if cached is not None:
                logger.debug(f"Cache HIT: {policy_path}, decision={decision}")
                cached_result = True
                if self.metrics:
                    self.metrics.record_cache_hit(source)
                result = cached
                return result
            else:
                if self.metrics:
                    self.metrics.record_cache_miss(source)

        # Check circuit breaker - should we attempt the call?
        admission: Admission | None = None
        if self.circuit_breaker:
            admission = await self.circuit_breaker.admit()
            if admission is None:
                # Circuit is open, use fallback
                stale_cached = await self._get_stale_cached(
                    identity_value, policy_path, decision, resource_context, **scope
                )
                logger.warning(
                    f"Circuit OPEN, using fallback for {policy_path} "
                    f"(stale_cache={'hit' if stale_cached is not None else 'miss'})"
                )

                result = await self.circuit_breaker.get_fallback_decision(
                    request,
                    policy_path,
                    dict(resource_context) if resource_context else {},
                    stale_cached,
                    ConnectionError("Circuit breaker open"),
                )

                if self.metrics:
                    self.metrics.record_fallback(
                        "circuit_open",
                        stale_cached is not None,
                        "allowed" if result else "denied",
                    )

                if self.circuit_breaker.on_fallback:
                    try:
                        self.circuit_breaker.on_fallback(
                            request, policy_path, stale_cached, result
                        )
                    except Exception as e:
                        logger.error(f"Error in on_fallback callback: {e}")

                return result

        # Make the authorization call over the shared channel
        topaz_start = time.monotonic()
        try:
            decisions_result = await self._call_authorizer(
                identity, policy_path, (decision,), resource_context, admission
            )
        except Exception as e:
            # Check if this is a failure that should trip the circuit breaker
            if not (self.circuit_breaker and self.circuit_breaker.is_failure_exception(e)):
                raise
            if self.metrics:
                self.metrics.record_error(type(e).__name__)
            if self.tracing and span:
                self.tracing.record_error(span, e)
                span = None  # Don't end span twice

            # Try fallback
            stale_cached = await self._get_stale_cached(
                identity_value, policy_path, decision, resource_context, **scope
            )
            logger.warning(
                f"Topaz call failed ({type(e).__name__}), using fallback for {policy_path}"
            )

            result = await self.circuit_breaker.get_fallback_decision(
                request,
                policy_path,
                dict(resource_context) if resource_context else {},
                stale_cached,
                e,
            )

            if self.metrics:
                self.metrics.record_fallback(
                    "error",
                    stale_cached is not None,
                    "allowed" if result else "denied",
                )

            if self.circuit_breaker.on_fallback:
                try:
                    self.circuit_breaker.on_fallback(request, policy_path, stale_cached, result)
                except Exception as cb_error:
                    logger.error(f"Error in on_fallback callback: {cb_error}")

            # Fallback produced a decision — not a propagating error
            return result

        result = decisions_result.get(decision, False)
        topaz_latency = time.monotonic() - topaz_start

        if self.metrics:
            self.metrics.record_topaz_latency(topaz_latency)

        # Cache writes stay outside the inner try: a backend error must not
        # turn into a fallback decision
        if self.decision_cache:
            await self.decision_cache.set(
                identity_value, policy_path, decision, resource_context, result, **scope
            )
            if self.metrics:
                self.metrics.set_cache_size(self.decision_cache.size())

        # Store in stale cache for circuit breaker fallback
        await self._set_stale_cached(
            identity_value, policy_path, decision, resource_context, result, **scope
        )

        return result

    except Exception as e:
        check_error = e
        if self.metrics:
            self.metrics.record_error(type(e).__name__)
        if self.tracing and span:
            self.tracing.record_error(span, e)
            span = None  # Don't end span twice
        raise

    finally:
        latency_seconds = time.monotonic() - start_time
        latency_ms = latency_seconds * 1000
        result_decision = "allowed" if result else "denied"

        # Record metrics
        if self.metrics:
            self.metrics.record_auth_request(
                source=source,
                decision=result_decision,
                check_type="policy",
                policy_path=policy_path,
            )
            self.metrics.record_latency(latency_seconds, source, cached_result, policy_path)

        # End tracing span
        if self.tracing and span:
            self.tracing.end_auth_span(
                span,
                decision=result_decision,
                cached=cached_result,
                latency_ms=latency_ms,
                resource_context=dict(resource_context) if resource_context else None,
            )

        # Audit logging — single emission point for all sources
        await self._emit_audit_event(
            request,
            policy_path,
            result,
            source,
            cached_result,
            latency_ms,
            resource_context,
            policy_resolution_source=policy_resolution_source,
            reason="authorizer_error" if check_error is not None else None,
        )

is_allowed async

is_allowed(request: Request, policy_path: str, resource_context: ResourceContext | None = None, decision: str = 'allowed', source: str = 'manual') -> bool

Check if an action is allowed without raising an exception.

This is a non-raising alternative to require_policy_allowed that returns True/False instead of raising HTTPException(403). Useful for UI patterns where you need to check permissions without blocking (e.g., showing/hiding edit or delete buttons).

PARAMETER	DESCRIPTION
request	The FastAPI request object TYPE: Request
policy_path	Full policy path (e.g., "webapp.PUT.documents") TYPE: str
resource_context	Optional resource context dict TYPE: ResourceContext | None DEFAULT: None
decision	Decision to check (default: "allowed") TYPE: str DEFAULT: 'allowed'

RETURNS	DESCRIPTION
bool	True if allowed, False otherwise

RAISES	DESCRIPTION
Exception	Authorizer errors the circuit breaker does not turn into a fallback decision (no breaker configured, or a non-failure code such as INVALID_ARGUMENT) propagate to the caller.

Example

@app.get("/documents/{id}")
async def get_document(id: int, request: Request):
    doc = await fetch_document(id)
    can_edit = await config.is_allowed(
        request,
        policy_path="myapp.PUT.documents",
        resource_context={"id": str(id)},
    )
    return {"document": doc, "can_edit": can_edit}

Source code in src/fastapi_topaz/config.py

async def is_allowed(
    self,
    request: Request,
    policy_path: str,
    resource_context: ResourceContext | None = None,
    decision: str = "allowed",
    source: str = "manual",
) -> bool:
    """
    Check if an action is allowed without raising an exception.

    This is a non-raising alternative to require_policy_allowed that returns
    True/False instead of raising HTTPException(403). Useful for UI patterns
    where you need to check permissions without blocking (e.g., showing/hiding
    edit or delete buttons).

    Args:
        request: The FastAPI request object
        policy_path: Full policy path (e.g., "webapp.PUT.documents")
        resource_context: Optional resource context dict
        decision: Decision to check (default: "allowed")

    Returns:
        True if allowed, False otherwise

    Raises:
        Exception: Authorizer errors the circuit breaker does not turn into a
            fallback decision (no breaker configured, or a non-failure code
            such as INVALID_ARGUMENT) propagate to the caller.

    Example:
        ```python
        @app.get("/documents/{id}")
        async def get_document(id: int, request: Request):
            doc = await fetch_document(id)
            can_edit = await config.is_allowed(
                request,
                policy_path="myapp.PUT.documents",
                resource_context={"id": str(id)},
            )
            return {"document": doc, "can_edit": can_edit}
        ```
    """
    ctx = self._policy_resource_context(request, resource_context)
    return await self.check_decision(request, policy_path, decision, ctx, source=source)

check_relation async

check_relation(request: Request, object_type: str, object_id: str, relation: str, subject_type: str = 'user', source: str = 'manual') -> bool

Check a ReBAC relation without raising an exception.

This is a non-raising alternative to require_rebac_allowed that returns True/False instead of raising HTTPException(403). Useful for checking if a user has a specific relationship with an object.

PARAMETER	DESCRIPTION
request	The FastAPI request object TYPE: Request
object_type	Type of object (e.g., "document", "folder") TYPE: str
object_id	ID of the object to check TYPE: str
relation	Relation to check (e.g., "can_read", "can_write", "can_delete") TYPE: str
subject_type	Subject type (default: "user") TYPE: str DEFAULT: 'user'

RETURNS	DESCRIPTION
bool	True if the relation exists, False otherwise

RAISES	DESCRIPTION
Exception	Authorizer errors the circuit breaker does not turn into a fallback decision (no breaker configured, or a non-failure code such as INVALID_ARGUMENT) propagate to the caller.

Example

@app.get("/documents/{id}")
async def get_document(id: int, request: Request):
    doc = await fetch_document(id)
    can_delete = await config.check_relation(
        request,
        object_type="document",
        object_id=str(id),
        relation="can_delete",
    )
    return {"document": doc, "can_delete": can_delete}

Source code in src/fastapi_topaz/config.py

async def check_relation(
    self,
    request: Request,
    object_type: str,
    object_id: str,
    relation: str,
    subject_type: str = "user",
    source: str = "manual",
) -> bool:
    """
    Check a ReBAC relation without raising an exception.

    This is a non-raising alternative to require_rebac_allowed that returns
    True/False instead of raising HTTPException(403). Useful for checking
    if a user has a specific relationship with an object.

    Args:
        request: The FastAPI request object
        object_type: Type of object (e.g., "document", "folder")
        object_id: ID of the object to check
        relation: Relation to check (e.g., "can_read", "can_write", "can_delete")
        subject_type: Subject type (default: "user")

    Returns:
        True if the relation exists, False otherwise

    Raises:
        Exception: Authorizer errors the circuit breaker does not turn into a
            fallback decision (no breaker configured, or a non-failure code
            such as INVALID_ARGUMENT) propagate to the caller.

    Example:
        ```python
        @app.get("/documents/{id}")
        async def get_document(id: int, request: Request):
            doc = await fetch_document(id)
            can_delete = await config.check_relation(
                request,
                object_type="document",
                object_id=str(id),
                relation="can_delete",
            )
            return {"document": doc, "can_delete": can_delete}
        ```
    """
    resource_ctx: ResourceContext = {}
    if self.resource_context_provider:
        resource_ctx.update(self.resource_context_provider(request))

    resource_ctx.update(
        {
            "object_type": object_type,
            "object_id": object_id,
            "relation": relation,
            "subject_type": subject_type,
        }
    )

    policy_path = f"{self.policy_path_root}.check"
    return await self.check_decision(
        request, policy_path, "allowed", resource_ctx, source=source
    )

check_relations async

check_relations(request: Request, object_type: str, object_id: str, relations: list[str], subject_type: str = 'user', source: str = 'manual', batch: bool = False) -> dict[str, bool]

Check multiple ReBAC relations at once without raising exceptions.

This method checks multiple relations concurrently and returns a dict mapping relation names to boolean results. Useful for fetching all permissions for an object in a single call (e.g., to populate a permissions object in an API response).

PARAMETER	DESCRIPTION
request	The FastAPI request object TYPE: Request
object_type	Type of object (e.g., "document", "folder") TYPE: str
object_id	ID of the object to check TYPE: str
relations	List of relations to check (e.g., ["can_read", "can_write", "can_delete"]) TYPE: list[str]
subject_type	Subject type (default: "user") TYPE: str DEFAULT: 'user'
batch	Evaluate all relations in a single Topaz call instead of one call per relation. Requires the {root}.check Rego package to define one rule per relation name (e.g. can_read, can_write) reading input.resource.object_type / object_id / subject_type — no relation key is sent. Default False (per-relation fan-out, unchanged behaviour). TYPE: bool DEFAULT: False

RETURNS	DESCRIPTION
dict[str, bool]	Dict mapping relation names to boolean results

RAISES	DESCRIPTION
Exception	Authorizer errors the circuit breaker does not turn into a fallback decision (no breaker configured, or a non-failure code such as INVALID_ARGUMENT) propagate to the caller.

Example

@app.get("/documents/{id}")
async def get_document(id: int, request: Request):
    doc = await fetch_document(id)
    permissions = await config.check_relations(
        request,
        object_type="document",
        object_id=str(id),
        relations=["can_read", "can_write", "can_delete", "can_share"],
    )
    # permissions = {"can_read": True, "can_write": True, "can_delete": False, "can_share": False}
    return {"document": doc, "permissions": permissions}

Source code in src/fastapi_topaz/config.py

async def check_relations(
    self,
    request: Request,
    object_type: str,
    object_id: str,
    relations: list[str],
    subject_type: str = "user",
    source: str = "manual",
    batch: bool = False,
) -> dict[str, bool]:
    """
    Check multiple ReBAC relations at once without raising exceptions.

    This method checks multiple relations concurrently and returns a dict
    mapping relation names to boolean results. Useful for fetching all
    permissions for an object in a single call (e.g., to populate a
    permissions object in an API response).

    Args:
        request: The FastAPI request object
        object_type: Type of object (e.g., "document", "folder")
        object_id: ID of the object to check
        relations: List of relations to check (e.g., ["can_read", "can_write", "can_delete"])
        subject_type: Subject type (default: "user")
        batch: Evaluate all relations in a single Topaz call instead of
            one call per relation. Requires the ``{root}.check`` Rego
            package to define one rule per relation name (e.g.
            ``can_read``, ``can_write``) reading
            ``input.resource.object_type`` / ``object_id`` /
            ``subject_type`` — no ``relation`` key is sent. Default False
            (per-relation fan-out, unchanged behaviour).

    Returns:
        Dict mapping relation names to boolean results

    Raises:
        Exception: Authorizer errors the circuit breaker does not turn into a
            fallback decision (no breaker configured, or a non-failure code
            such as INVALID_ARGUMENT) propagate to the caller.

    Example:
        ```python
        @app.get("/documents/{id}")
        async def get_document(id: int, request: Request):
            doc = await fetch_document(id)
            permissions = await config.check_relations(
                request,
                object_type="document",
                object_id=str(id),
                relations=["can_read", "can_write", "can_delete", "can_share"],
            )
            # permissions = {"can_read": True, "can_write": True, "can_delete": False, "can_share": False}
            return {"document": doc, "permissions": permissions}
        ```
    """
    if batch:
        resource_ctx: ResourceContext = {}
        if self.resource_context_provider:
            resource_ctx.update(self.resource_context_provider(request))
        resource_ctx.update(
            {
                "object_type": object_type,
                "object_id": object_id,
                "subject_type": subject_type,
            }
        )
        return await self._check_decisions_batch(
            request,
            f"{self.policy_path_root}.check",
            relations,
            resource_ctx,
            source=source,
        )

    async def check_single_relation(rel: str) -> tuple[str, bool]:
        async with self._get_semaphore():
            result = await self.check_relation(
                request,
                object_type=object_type,
                object_id=object_id,
                relation=rel,
                subject_type=subject_type,
                source=source,
            )
        return rel, result

    results = await asyncio.gather(*[check_single_relation(rel) for rel in relations])
    return dict(results)

invalidate_cache async

invalidate_cache(identity_value: str | None = None, policy_path: str | None = None, object_id: str | None = None) -> int

Invalidate cached decisions matching all provided criteria (AND).

Use after permission changes (e.g. revoking a user's access) so cached decisions do not keep serving stale grants until TTL expiry.

Delegates to decision_cache.invalidate(...) when the configured backend implements it; backends without an invalidate method are skipped. The stale fallback cache is always cleared entirely.

PARAMETER	DESCRIPTION
identity_value	Invalidate entries for this identity TYPE: str | None DEFAULT: None
policy_path	Invalidate entries for this policy path TYPE: str | None DEFAULT: None
object_id	Invalidate entries for this object ID TYPE: str | None DEFAULT: None

RETURNS	DESCRIPTION
int	Number of entries removed from the decision cache (0 when the
int	backend has no invalidate).

RAISES	DESCRIPTION
ValueError	If no criterion is provided.

Source code in src/fastapi_topaz/config.py

async def invalidate_cache(
    self,
    identity_value: str | None = None,
    policy_path: str | None = None,
    object_id: str | None = None,
) -> int:
    """Invalidate cached decisions matching all provided criteria (AND).

    Use after permission changes (e.g. revoking a user's access) so cached
    decisions do not keep serving stale grants until TTL expiry.

    Delegates to ``decision_cache.invalidate(...)`` when the configured
    backend implements it; backends without an ``invalidate`` method are
    skipped. The stale fallback cache is always cleared entirely.

    Args:
        identity_value: Invalidate entries for this identity
        policy_path: Invalidate entries for this policy path
        object_id: Invalidate entries for this object ID

    Returns:
        Number of entries removed from the decision cache (0 when the
        backend has no ``invalidate``).

    Raises:
        ValueError: If no criterion is provided.
    """
    if identity_value is None and policy_path is None and object_id is None:
        raise ValueError("invalidate_cache() requires at least one criterion")

    removed = 0
    invalidate = getattr(self.decision_cache, "invalidate", None)
    if invalidate is not None:
        removed = await invalidate(
            identity_value=identity_value,
            policy_path=policy_path,
            object_id=object_id,
        )

    # efficiency: stale-cache entries do not carry key components, so
    # selective matching is impossible — clear it all; correctness of
    # revocation beats fallback availability
    async with self._get_stale_cache_lock():
        self._stale_cache.clear()

    return removed

health async

health(ping: bool = False) -> dict[str, Any]

Report authorization readiness for health/readiness endpoints.

PARAMETER	DESCRIPTION
ping	Also call the authorizer's Info RPC to verify reachability. Ping failures are reported in the result, never raised. TYPE: bool DEFAULT: False

RETURNS	DESCRIPTION
dict[str, Any]	Dict with keys: healthy (False when the circuit breaker is
dict[str, Any]	open or a requested ping failed), authorizer_url,
dict[str, Any]	check_timeout, circuit_breaker (status dict or None),
dict[str, Any]	cache_size (int or None), and ping ({"ok", "error"}
dict[str, Any]	or None when not requested).

Example

@app.get("/health/authz")
async def authz_health():
    return await config.health(ping=True)

Source code in src/fastapi_topaz/config.py

async def health(self, ping: bool = False) -> dict[str, Any]:
    """Report authorization readiness for health/readiness endpoints.

    Args:
        ping: Also call the authorizer's Info RPC to verify reachability.
            Ping failures are reported in the result, never raised.

    Returns:
        Dict with keys: ``healthy`` (False when the circuit breaker is
        open or a requested ping failed), ``authorizer_url``,
        ``check_timeout``, ``circuit_breaker`` (status dict or None),
        ``cache_size`` (int or None), and ``ping`` (``{"ok", "error"}``
        or None when not requested).

    Example:
        ```python
        @app.get("/health/authz")
        async def authz_health():
            return await config.health(ping=True)
        ```
    """
    healthy = True

    circuit_status = None
    if self.circuit_breaker:
        status = self.circuit_breaker.status()
        circuit_status = asdict(status)
        if status.is_open:
            healthy = False

    ping_result: dict[str, Any] | None = None
    if ping:
        try:
            await self._authorizer.info(timeout=self.check_timeout)
            ping_result = {"ok": True, "error": None}
        except Exception as e:
            ping_result = {"ok": False, "error": f"{type(e).__name__}: {e}"}
            healthy = False

    return {
        "healthy": healthy,
        "authorizer_url": self.authorizer_options.url,
        "check_timeout": self.check_timeout,
        "circuit_breaker": circuit_status,
        "cache_size": self.decision_cache.size() if self.decision_cache else None,
        "ping": ping_result,
    }

policy_path_for

policy_path_for(method: str, route_path: str) -> str

Generate the policy path for a given HTTP method and route path.

Useful for debugging, testing, or previewing what policy path will be generated for a given route.

PARAMETER	DESCRIPTION
method	HTTP method (e.g., "GET", "POST") TYPE: str
route_path	URL path pattern (e.g., "/documents/{id}") TYPE: str

RETURNS	DESCRIPTION
str	The policy path that would be used for authorization

Example

config.policy_path_for("GET", "/documents/{id}") "myapp.GET.documents.__id"

Source code in src/fastapi_topaz/config.py

def policy_path_for(self, method: str, route_path: str) -> str:
    """
    Generate the policy path for a given HTTP method and route path.

    Useful for debugging, testing, or previewing what policy path
    will be generated for a given route.

    Args:
        method: HTTP method (e.g., "GET", "POST")
        route_path: URL path pattern (e.g., "/documents/{id}")

    Returns:
        The policy path that would be used for authorization

    Example:
        >>> config.policy_path_for("GET", "/documents/{id}")
        "myapp.GET.documents.__id"
    """
    return _resolve_policy_path(
        self.policy_path_root, method, route_path, self.policy_path_normalizer
    )

---

DecisionCache

Simple in-memory TTL cache for authorization decisions.

PARAMETER	DESCRIPTION
ttl_seconds	Time-to-live for cache entries (default: 60 seconds) TYPE: float DEFAULT: 60.0
max_size	Maximum number of entries to cache (default: 1000) TYPE: int DEFAULT: 1000

get async

get(identity_value: str, policy_path: str, decision: str, resource_context: ResourceContext | None, *, identity_type: str = '', policy_instance: str = '') -> bool | None

Get a cached decision, or None if not cached or expired.

Source code in src/fastapi_topaz/cache.py

async def get(
    self,
    identity_value: str,
    policy_path: str,
    decision: str,
    resource_context: ResourceContext | None,
    *,
    identity_type: str = "",
    policy_instance: str = "",
) -> bool | None:
    """Get a cached decision, or None if not cached or expired."""
    key = self._make_key(
        identity_value,
        policy_path,
        decision,
        resource_context,
        identity_type=identity_type,
        policy_instance=policy_instance,
    )
    async with self._lock:
        entry = self._cache.get(key)
        if entry is None:
            return None
        if time.monotonic() > entry.expires_at:
            del self._cache[key]
            return None
        # Re-insert to mark as recently used (LRU)
        del self._cache[key]
        self._cache[key] = entry
        return entry.value

set async

set(identity_value: str, policy_path: str, decision: str, resource_context: ResourceContext | None, value: bool, *, identity_type: str = '', policy_instance: str = '') -> None

Cache a decision.

Source code in src/fastapi_topaz/cache.py

async def set(
    self,
    identity_value: str,
    policy_path: str,
    decision: str,
    resource_context: ResourceContext | None,
    value: bool,
    *,
    identity_type: str = "",
    policy_instance: str = "",
) -> None:
    """Cache a decision."""
    key = self._make_key(
        identity_value,
        policy_path,
        decision,
        resource_context,
        identity_type=identity_type,
        policy_instance=policy_instance,
    )
    async with self._lock:
        # Evict oldest entries if cache is full
        if len(self._cache) >= self.max_size:
            # Remove expired entries first
            now = time.monotonic()
            expired = [k for k, v in self._cache.items() if v.expires_at < now]
            for k in expired:
                del self._cache[k]
            # If still full, remove oldest 10%
            if len(self._cache) >= self.max_size:
                to_remove = list(self._cache.keys())[: max(1, self.max_size // 10)]
                for k in to_remove:
                    del self._cache[k]

        object_id = resource_context.get("object_id") if resource_context else None
        self._cache[key] = CacheEntry(
            value=value,
            expires_at=time.monotonic() + self.ttl_seconds,
            identity_value=identity_value,
            policy_path=policy_path,
            object_id=str(object_id) if object_id is not None else None,
        )

invalidate async

invalidate(identity_value: str | None = None, policy_path: str | None = None, object_id: str | None = None) -> int

Remove cached decisions matching all provided criteria (AND).

PARAMETER	DESCRIPTION
identity_value	Remove entries for this identity TYPE: str | None DEFAULT: None
policy_path	Remove entries for this policy path TYPE: str | None DEFAULT: None
object_id	Remove entries whose resource context had this object_id TYPE: str | None DEFAULT: None

RETURNS	DESCRIPTION
int	Number of entries removed.

RAISES	DESCRIPTION
ValueError	If no criterion is provided (use :meth:clear to remove everything).

Source code in src/fastapi_topaz/cache.py

async def invalidate(
    self,
    identity_value: str | None = None,
    policy_path: str | None = None,
    object_id: str | None = None,
) -> int:
    """Remove cached decisions matching all provided criteria (AND).

    Args:
        identity_value: Remove entries for this identity
        policy_path: Remove entries for this policy path
        object_id: Remove entries whose resource context had this object_id

    Returns:
        Number of entries removed.

    Raises:
        ValueError: If no criterion is provided (use :meth:`clear` to
            remove everything).
    """
    if identity_value is None and policy_path is None and object_id is None:
        raise ValueError("invalidate() requires at least one criterion; use clear() instead")

    async with self._lock:
        matching = [
            key
            for key, entry in self._cache.items()
            if (identity_value is None or entry.identity_value == identity_value)
            and (policy_path is None or entry.policy_path == policy_path)
            and (object_id is None or entry.object_id == object_id)
        ]
        for key in matching:
            del self._cache[key]
        return len(matching)

size

size() -> int

Current number of cached entries (may include expired ones).

Source code in src/fastapi_topaz/cache.py

def size(self) -> int:
    """Current number of cached entries (may include expired ones)."""
    return len(self._cache)

clear async

clear() -> None

Clear all cached entries.

Source code in src/fastapi_topaz/cache.py

async def clear(self) -> None:
    """Clear all cached entries."""
    async with self._lock:
        self._cache.clear()

---

Authorization Dependencies

require_policy_allowed

Async dependency that raises HTTPException(403) if policy denies access. When the authorizer call itself fails, raises 403 or 503 per config.on_error.

PARAMETER	DESCRIPTION
config	Topaz configuration TYPE: TopazConfig
policy_path	Full policy path (e.g., "webapp.POST.api.documents") TYPE: str
decision	Decision to check (default: "allowed") TYPE: str DEFAULT: 'allowed'
resource_context	Optional resource context dict TYPE: ResourceContext | None DEFAULT: None

RETURNS	DESCRIPTION
Callable[[Request], Awaitable[None]]	Async dependency function for FastAPI

Example

@router.post("/documents")
async def create_document(
    _: None = Depends(require_policy_allowed(topaz_config, "webapp.POST.api.documents")),
):
    ...

Source code in src/fastapi_topaz/dependencies.py

def require_policy_allowed(
    config: TopazConfig,
    policy_path: str,
    decision: str = "allowed",
    resource_context: ResourceContext | None = None,
) -> Callable[[Request], Awaitable[None]]:
    """
    Async dependency that raises HTTPException(403) if policy denies access.
    When the authorizer call itself fails, raises 403 or 503 per ``config.on_error``.

    Args:
        config: Topaz configuration
        policy_path: Full policy path (e.g., "webapp.POST.api.documents")
        decision: Decision to check (default: "allowed")
        resource_context: Optional resource context dict

    Returns:
        Async dependency function for FastAPI

    Example:
        ```python
        @router.post("/documents")
        async def create_document(
            _: None = Depends(require_policy_allowed(topaz_config, "webapp.POST.api.documents")),
        ):
            ...
        ```
    """

    async def dependency(request: Request) -> None:
        await _check_policy_and_raise(config, request, policy_path, decision, resource_context)

    return dependency

---

require_policy_auto

Async dependency that auto-generates policy path from route and raises HTTPException(403) if denied. When the authorizer call itself fails, raises 403 or 503 per config.on_error.

The policy path is automatically derived from the HTTP method and route path pattern: - GET /documents -> {root}.GET.documents - POST /documents -> {root}.POST.documents - GET /documents/{id} -> {root}.GET.documents.__id - PUT /users/{user_id}/docs/{doc_id} -> {root}.PUT.users.__user_id.docs.__doc_id

PARAMETER	DESCRIPTION
config	Topaz configuration TYPE: TopazConfig
decision	Decision to check (default: "allowed") TYPE: str DEFAULT: 'allowed'
resource_context	Optional resource context dict TYPE: ResourceContext | None DEFAULT: None

RETURNS	DESCRIPTION
Callable[[Request], Awaitable[None]]	Async dependency function for FastAPI

Example

@router.get("/documents/{id}")
async def get_document(
    id: int,
    _: None = Depends(require_policy_auto(topaz_config)),
):
    # Policy path auto-generated as "myapp.GET.documents.__id"
    ...

Source code in src/fastapi_topaz/dependencies.py

def require_policy_auto(
    config: TopazConfig,
    decision: str = "allowed",
    resource_context: ResourceContext | None = None,
) -> Callable[[Request], Awaitable[None]]:
    """
    Async dependency that auto-generates policy path from route and raises HTTPException(403) if denied.
    When the authorizer call itself fails, raises 403 or 503 per ``config.on_error``.

    The policy path is automatically derived from the HTTP method and route path pattern:
    - GET /documents -> {root}.GET.documents
    - POST /documents -> {root}.POST.documents
    - GET /documents/{id} -> {root}.GET.documents.__id
    - PUT /users/{user_id}/docs/{doc_id} -> {root}.PUT.users.__user_id.docs.__doc_id

    Args:
        config: Topaz configuration
        decision: Decision to check (default: "allowed")
        resource_context: Optional resource context dict

    Returns:
        Async dependency function for FastAPI

    Example:
        ```python
        @router.get("/documents/{id}")
        async def get_document(
            id: int,
            _: None = Depends(require_policy_auto(topaz_config)),
        ):
            # Policy path auto-generated as "myapp.GET.documents.__id"
            ...
        ```
    """

    async def dependency(request: Request) -> None:
        # Route path pattern with include_router() prefixes; never fall back to
        # the un-prefixed route path, which would check the wrong policy
        route_path = resolve_route_path(request.scope)
        if route_path is None:
            logger.error(
                "Could not resolve the route path for %s %s", request.method, request.url.path
            )
            raise HTTPException(
                status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
                detail="Unable to determine route for policy path auto-resolution",
            )

        method = request.method

        # Generate policy path
        policy_path = _resolve_policy_path(
            config.policy_path_root,
            method,
            route_path,
            config.policy_path_normalizer,
        )

        await _check_policy_and_raise(config, request, policy_path, decision, resource_context)

    return dependency

---

require_rebac_allowed

Async dependency that raises HTTPException(403) if ReBAC check fails. When the authorizer call itself fails, raises 403 or 503 per config.on_error.

PARAMETER	DESCRIPTION
config	Topaz configuration TYPE: TopazConfig
object_type	Type of object (e.g., "document", "folder") TYPE: str
relation	Relation to check (e.g., "can_write", "can_delete") TYPE: str
object_id	Static ID, callable to extract from request, or None (uses path param "id") TYPE: str | Callable[[Request], str] | None DEFAULT: None
subject_type	Subject type (default: "user") TYPE: str DEFAULT: 'user'

RETURNS	DESCRIPTION
Callable[[Request], Awaitable[None]]	Async dependency function for FastAPI

Example

@router.put("/documents/{id}")
async def update_document(
    id: int,
    _: None = Depends(require_rebac_allowed(topaz_config, "document", "can_write")),
):
    ...

Source code in src/fastapi_topaz/dependencies.py

def require_rebac_allowed(
    config: TopazConfig,
    object_type: str,
    relation: str,
    object_id: str | Callable[[Request], str] | None = None,
    subject_type: str = "user",
) -> Callable[[Request], Awaitable[None]]:
    """
    Async dependency that raises HTTPException(403) if ReBAC check fails.
    When the authorizer call itself fails, raises 403 or 503 per ``config.on_error``.

    Args:
        config: Topaz configuration
        object_type: Type of object (e.g., "document", "folder")
        relation: Relation to check (e.g., "can_write", "can_delete")
        object_id: Static ID, callable to extract from request, or None (uses path param "id")
        subject_type: Subject type (default: "user")

    Returns:
        Async dependency function for FastAPI

    Example:
        ```python
        @router.put("/documents/{id}")
        async def update_document(
            id: int,
            _: None = Depends(require_rebac_allowed(topaz_config, "document", "can_write")),
        ):
            ...
        ```
    """
    _reject_empty_static_object_id(object_id)

    async def dependency(request: Request) -> None:
        # Resolve object_id
        if callable(object_id):
            obj_id = object_id(request)
            _require_object_id(obj_id, request, "<callable object_id>")
        elif object_id is not None:
            obj_id = object_id
        else:
            # Default: extract from path params
            obj_id = str(request.path_params.get("id", ""))
            _require_object_id(obj_id, request, "id")

        # Start with resource context from provider (includes document data, user info, etc.)
        resource_ctx: ResourceContext = {}
        if config.resource_context_provider:
            resource_ctx.update(config.resource_context_provider(request))

        # Add ReBAC-specific fields
        resource_ctx.update(
            {
                "object_type": object_type,
                "object_id": obj_id,
                "relation": relation,
                "subject_type": subject_type,
            }
        )

        policy_path = f"{config.policy_path_root}.check"

        try:
            allowed = await config.check_decision(request, policy_path, "allowed", resource_ctx)
        except Exception as exc:
            _raise_authz_error(config, policy_path, "rebac", exc)

        if not allowed:
            _raise_rebac_denied(config, relation, object_type, obj_id)

    return dependency

---

get_authorized_resource

Async dependency that checks authorization, then fetches the resource. Returns resource or raises 403/404. When the authorizer call itself fails, raises 403 or 503 per config.on_error.

Authorization runs first, so a denied request gets 403 whether or not the resource exists (no existence oracle) and the fetcher is never called.

PARAMETER	DESCRIPTION
config	Topaz configuration TYPE: TopazConfig
resource_fetcher	Function that takes (request) and returns resource or None. Coroutine functions and objects with an async __call__ are awaited; sync functions run in a threadpool so blocking I/O does not stall the event loop, and an awaitable they return (e.g. from lambda r: fetch(r)) is awaited too. TYPE: Callable[[Request], T | None | Awaitable[T | None]]
object_type	Type of object (e.g., "document") TYPE: str
relation	Relation to check (e.g., "can_write") TYPE: str
object_id	Static ID, callable, or None (uses path param "id") TYPE: str | Callable[[Request], str] | None DEFAULT: None
subject_type	Subject type (default: "user") TYPE: str DEFAULT: 'user'

RETURNS	DESCRIPTION
Callable[[Request], Awaitable[T]]	Async dependency function that returns the authorized resource

Example

def fetch_document(request: Request, db: Session) -> Document | None:
    doc_id = request.path_params["id"]
    return db.query(Document).filter(Document.id == doc_id).first()

@router.put("/documents/{id}")
async def update_document(
    document: Document = Depends(
        get_authorized_resource(topaz_config, fetch_document, "document", "can_write")
    ),
):
    # document is pre-fetched and authorized
    ...

Source code in src/fastapi_topaz/dependencies.py

def get_authorized_resource(
    config: TopazConfig,
    resource_fetcher: Callable[[Request], T | None | Awaitable[T | None]],
    object_type: str,
    relation: str,
    object_id: str | Callable[[Request], str] | None = None,
    subject_type: str = "user",
) -> Callable[[Request], Awaitable[T]]:
    """
    Async dependency that checks authorization, then fetches the resource.
    Returns resource or raises 403/404.
    When the authorizer call itself fails, raises 403 or 503 per ``config.on_error``.

    Authorization runs first, so a denied request gets 403 whether or not the
    resource exists (no existence oracle) and the fetcher is never called.

    Args:
        config: Topaz configuration
        resource_fetcher: Function that takes (request) and returns resource or
            None. Coroutine functions and objects with an async ``__call__``
            are awaited; sync functions run in a threadpool so blocking I/O
            does not stall the event loop, and an awaitable they return (e.g.
            from ``lambda r: fetch(r)``) is awaited too.
        object_type: Type of object (e.g., "document")
        relation: Relation to check (e.g., "can_write")
        object_id: Static ID, callable, or None (uses path param "id")
        subject_type: Subject type (default: "user")

    Returns:
        Async dependency function that returns the authorized resource

    Example:
        ```python
        def fetch_document(request: Request, db: Session) -> Document | None:
            doc_id = request.path_params["id"]
            return db.query(Document).filter(Document.id == doc_id).first()

        @router.put("/documents/{id}")
        async def update_document(
            document: Document = Depends(
                get_authorized_resource(topaz_config, fetch_document, "document", "can_write")
            ),
        ):
            # document is pre-fetched and authorized
            ...
        ```
    """
    _reject_empty_static_object_id(object_id)
    # An object with async __call__ is a coroutine function only through __call__
    call = getattr(resource_fetcher, "__call__", None)
    is_async_fetcher = inspect.iscoroutinefunction(resource_fetcher) or (
        inspect.iscoroutinefunction(call)
    )

    async def dependency(request: Request) -> T:
        # Resolve object_id
        if callable(object_id):
            obj_id = object_id(request)
            _require_object_id(obj_id, request, "<callable object_id>")
        elif object_id is not None:
            obj_id = object_id
        else:
            obj_id = str(request.path_params.get("id", ""))
            _require_object_id(obj_id, request, "id")

        # Check authorization
        resource_ctx: ResourceContext = {
            "object_type": object_type,
            "object_id": obj_id,
            "relation": relation,
            "subject_type": subject_type,
        }

        policy_path = f"{config.policy_path_root}.check"

        try:
            allowed = await config.check_decision(request, policy_path, "allowed", resource_ctx)
        except Exception as exc:
            _raise_authz_error(config, policy_path, "rebac", exc)

        if not allowed:
            _raise_rebac_denied(config, relation, object_type, obj_id)

        # Fetch only after authorization succeeded
        if is_async_fetcher:
            resource = await cast(Callable[[Request], Awaitable["T | None"]], resource_fetcher)(
                request
            )
        else:
            resource = await run_in_threadpool(functools.partial(resource_fetcher, request))
        # A sync wrapper such as lambda r: fetch(r) returns a coroutine
        if inspect.isawaitable(resource):
            resource = await resource

        if resource is None:
            raise HTTPException(
                status_code=status.HTTP_404_NOT_FOUND,
                detail=f"{object_type.capitalize()} not found",
            )

        return cast(T, resource)

    return dependency

---

filter_authorized_resources

Async dependency that returns an async filter function to remove unauthorized resources. When the authorizer call itself fails, raises 403 or 503 per config.on_error.

Uses concurrent authorization checks (controlled by config.max_concurrent_checks) and caching (if config.decision_cache is set) for optimal performance.

PARAMETER	DESCRIPTION
config	Topaz configuration TYPE: TopazConfig
object_type	Type of object (e.g., "document") TYPE: str
relation	Relation to check (e.g., "can_read") TYPE: str
id_extractor	Function to extract ID from resource object TYPE: Callable[[Any], str] DEFAULT: lambda obj: str(getattr(obj, 'id', ''))
subject_type	Subject type (default: "user") TYPE: str DEFAULT: 'user'

RETURNS	DESCRIPTION
Callable[[Request], Awaitable[Callable[[list[T]], Awaitable[list[T]]]]]	Async dependency that returns an async filter function

Example

@router.get("/documents")
async def list_documents(
    filter_fn: Callable = Depends(
        filter_authorized_resources(topaz_config, "document", "can_read")
    ),
    db: Session = Depends(get_db),
):
    all_docs = db.query(Document).all()
    authorized_docs = await filter_fn(all_docs)
    return authorized_docs

Source code in src/fastapi_topaz/dependencies.py

def filter_authorized_resources(
    config: TopazConfig,
    object_type: str,
    relation: str,
    id_extractor: Callable[[Any], str] = lambda obj: str(getattr(obj, "id", "")),
    subject_type: str = "user",
) -> Callable[[Request], Awaitable[Callable[[list[T]], Awaitable[list[T]]]]]:
    """
    Async dependency that returns an async filter function to remove unauthorized resources.
    When the authorizer call itself fails, raises 403 or 503 per ``config.on_error``.

    Uses concurrent authorization checks (controlled by config.max_concurrent_checks)
    and caching (if config.decision_cache is set) for optimal performance.

    Args:
        config: Topaz configuration
        object_type: Type of object (e.g., "document")
        relation: Relation to check (e.g., "can_read")
        id_extractor: Function to extract ID from resource object
        subject_type: Subject type (default: "user")

    Returns:
        Async dependency that returns an async filter function

    Example:
        ```python
        @router.get("/documents")
        async def list_documents(
            filter_fn: Callable = Depends(
                filter_authorized_resources(topaz_config, "document", "can_read")
            ),
            db: Session = Depends(get_db),
        ):
            all_docs = db.query(Document).all()
            authorized_docs = await filter_fn(all_docs)
            return authorized_docs
        ```
    """

    async def dependency(request: Request) -> Callable[[list[T]], Awaitable[list[T]]]:
        async def check_single(resource: T) -> tuple[T, bool]:
            """Check authorization for a single resource with semaphore limiting."""
            obj_id = id_extractor(resource)
            if not obj_id:
                raise ValueError(
                    f"id_extractor returned an empty object ID for resource {resource!r}; "
                    f"cannot authorize {relation} on {object_type}"
                )

            resource_ctx: ResourceContext = {
                "object_type": object_type,
                "object_id": obj_id,
                "relation": relation,
                "subject_type": subject_type,
            }

            policy_path = f"{config.policy_path_root}.check"

            # Use semaphore to limit concurrent checks
            async with config._get_semaphore():
                try:
                    allowed = await config.check_decision(
                        request, policy_path, "allowed", resource_ctx
                    )
                except Exception as exc:
                    _raise_authz_error(config, policy_path, "rebac", exc)

            return resource, allowed

        async def filter_fn(resources: list[T]) -> list[T]:
            if not resources:
                return []

            # Run all checks concurrently (limited by semaphore)
            results = await asyncio.gather(*[check_single(r) for r in resources])

            # Filter to only authorized resources
            return [resource for resource, allowed in results if allowed]

        return filter_fn

    return dependency

---

require_rebac_hierarchy

Async dependency for hierarchical ReBAC authorization. When the authorizer call itself fails, raises 403 or 503 per config.on_error.

Checks multiple object/relation pairs in a single dependency, reducing boilerplate for nested resources like /orgs/{org}/projects/{proj}/docs/{doc}.

PARAMETER	DESCRIPTION
config	Topaz configuration TYPE: TopazConfig
checks	List of (object_type, id_source, relation) tuples. id_source can be: - "param_name" -> request.path_params["param_name"] - "header:X-Name" -> request.headers["X-Name"] - "query:name" -> request.query_params["name"] - "static:value" -> literal "value" - callable -> callable(request) TYPE: list[tuple[str, str, str]]
mode	Check mode: - "all" (default): All checks must pass (AND). Fails fast. - "any": At least one check must pass (OR). - "first_match": Return on first success. TYPE: Literal['all', 'any', 'first_match'] DEFAULT: 'all'
subject_type	Subject type (default: "user") TYPE: str DEFAULT: 'user'
optimize	Run checks concurrently when possible (default: True) TYPE: bool DEFAULT: True

RETURNS	DESCRIPTION
Callable[[Request], Awaitable[None]]	Async dependency function for FastAPI

RAISES	DESCRIPTION
ValueError	At factory time, if checks is empty
HTTPException(403)	If authorization fails based on mode semantics

Example

@app.get("/orgs/{org_id}/projects/{proj_id}/docs/{doc_id}")
async def get_doc(
    _=Depends(require_rebac_hierarchy(config, [
        ("organization", "org_id", "member"),
        ("project", "proj_id", "viewer"),
        ("document", "doc_id", "can_read"),
    ])),
):
    ...

Source code in src/fastapi_topaz/dependencies.py

def require_rebac_hierarchy(
    config: TopazConfig,
    checks: list[tuple[str, str, str]],
    mode: Literal["all", "any", "first_match"] = "all",
    subject_type: str = "user",
    optimize: bool = True,
) -> Callable[[Request], Awaitable[None]]:
    """
    Async dependency for hierarchical ReBAC authorization.
    When the authorizer call itself fails, raises 403 or 503 per ``config.on_error``.

    Checks multiple object/relation pairs in a single dependency, reducing
    boilerplate for nested resources like /orgs/{org}/projects/{proj}/docs/{doc}.

    Args:
        config: Topaz configuration
        checks: List of (object_type, id_source, relation) tuples.
            id_source can be:
            - "param_name" -> request.path_params["param_name"]
            - "header:X-Name" -> request.headers["X-Name"]
            - "query:name" -> request.query_params["name"]
            - "static:value" -> literal "value"
            - callable -> callable(request)
        mode: Check mode:
            - "all" (default): All checks must pass (AND). Fails fast.
            - "any": At least one check must pass (OR).
            - "first_match": Return on first success.
        subject_type: Subject type (default: "user")
        optimize: Run checks concurrently when possible (default: True)

    Returns:
        Async dependency function for FastAPI

    Raises:
        ValueError: At factory time, if ``checks`` is empty
        HTTPException(403): If authorization fails based on mode semantics

    Example:
        ```python
        @app.get("/orgs/{org_id}/projects/{proj_id}/docs/{doc_id}")
        async def get_doc(
            _=Depends(require_rebac_hierarchy(config, [
                ("organization", "org_id", "member"),
                ("project", "proj_id", "viewer"),
                ("document", "doc_id", "can_read"),
            ])),
        ):
            ...
        ```
    """
    if not checks:
        # An empty "all" would vacuously allow everything
        raise ValueError("require_rebac_hierarchy() requires at least one check")

    async def dependency(request: Request) -> None:
        try:
            result = await config.check_hierarchy(
                request, checks, mode, subject_type, optimize, source="dependency"
            )
        except ValueError as e:
            logger.error(
                "Could not resolve object ID in hierarchy check for %s %s: %s",
                request.method,
                request.url.path,
                e,
            )
            raise HTTPException(
                status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
                detail="Authorization misconfiguration: could not resolve object ID",
            ) from e
        except Exception as exc:
            _raise_authz_error(config, f"{config.policy_path_root}.check", "rebac", exc)

        if not result.allowed:
            if result.denied_at:
                logger.debug(f"Hierarchy access DENIED at {result.denied_at}")
            else:
                logger.debug("Hierarchy access DENIED: no matching permissions")
            raise HTTPException(
                status_code=status.HTTP_403_FORBIDDEN,
                detail="Forbidden",
            )

    return dependency

---

HierarchyResult

Result of a hierarchy authorization check.

ATTRIBUTE	DESCRIPTION
allowed	Whether the hierarchy check passed TYPE: bool
checks	List of (object_type, object_id, relation, result) tuples TYPE: list[tuple[str, str, str, bool]]
denied_at	Object type where access was denied (mode="all") TYPE: str | None
first_match	Relation that matched first (mode="first_match") TYPE: str | None

as_dict

as_dict() -> dict[str, bool]

Return dict mapping object_type to boolean result.

Source code in src/fastapi_topaz/config.py

def as_dict(self) -> dict[str, bool]:
    """Return dict mapping object_type to boolean result."""
    return {obj_type: result for obj_type, _, _, result in self.checks}

---

Middleware

TopazMiddleware

FastAPI middleware for global authorization (pure ASGI).

Auto-protects all routes by checking policy paths derived from HTTP method and route pattern. Routes are protected unless explicitly excluded.

Policy Resolution Chain (first match wins): 1. Explicit: Route has a .rego file in policies_dir 2. Group: Route matches a :class:PolicyGroup url_pattern 3. Default: config.default_policy is set 4. Generated: Use auto-generated policy path (legacy behaviour)

PARAMETER	DESCRIPTION
app	The FastAPI application TYPE: ASGIApp
config	TopazConfig with authorizer settings TYPE: TopazConfig
exclude_paths	Regex patterns for paths to skip (e.g., [r"^/health$", r"^/docs.*"]) TYPE: list[str] | None DEFAULT: None
exclude_methods	HTTP methods to skip (default: ["OPTIONS", "HEAD"]) TYPE: list[str] | None DEFAULT: None
on_missing_identity	How to handle missing identity: - "deny": Return 401 Unauthorized - "anonymous": Pass anonymous identity to Topaz (let policy decide) TYPE: Literal['deny', 'anonymous'] DEFAULT: 'deny'
on_denied	Optional callback to customize 403 response TYPE: Callable[[Request, str], Response] | None DEFAULT: None
on_error	How to respond when the authorization check itself fails (e.g. authorizer unreachable) and no circuit breaker fallback applies: - None (default): use config.on_error, shared with the dependencies - "deny": Return 403 Forbidden (on_denied is used when set) - "unavailable": Return 503 Service Unavailable, with Retry-After when the config has a circuit breaker TYPE: OnError | None DEFAULT: None
policies_dir	Optional directory to scan for explicit .rego policy files at startup. When provided, the middleware builds a set of known policy paths and uses the resolution chain to decide which policy to evaluate per request. TYPE: str | Path | None DEFAULT: None

Source code in src/fastapi_topaz/middleware.py

def __init__(
    self,
    app: ASGIApp,
    config: TopazConfig,
    exclude_paths: list[str] | None = None,
    exclude_methods: list[str] | None = None,
    on_missing_identity: Literal["deny", "anonymous"] = "deny",
    on_denied: Callable[[Request, str], Response] | None = None,
    on_error: OnError | None = None,
    policies_dir: str | Path | None = None,
) -> None:
    self.app = app
    self.config = config
    self.exclude_paths = [re.compile(p) for p in (exclude_paths or [])]
    self.exclude_methods = set(exclude_methods or _DEFAULT_EXCLUDE_METHODS)
    self.on_missing_identity = on_missing_identity
    self.on_denied = on_denied
    if on_error not in (None, "deny", "unavailable"):
        raise ValueError(f"on_error must be 'deny', 'unavailable' or None, got {on_error!r}")
    self.on_error: OnError | None = on_error

    # --- Resolution chain setup ---
    # Scan explicit policy files
    self._scanned_policies: set[str] | None = None
    if policies_dir is not None:
        self._scanned_policies = scan_policy_files(policies_dir)
        logger.info(
            "Scanned %d policy files from %s", len(self._scanned_policies), policies_dir
        )

    # Pre-compile policy group patterns (avoid per-request re.compile)
    self._compiled_groups = _compile_policy_groups(config.policy_groups)

    # Cache for static route matches: (method, path) -> (route, child_scope).
    # Only routes without path params are cached — parameterized matches
    # produce per-URL child scopes.
    self._route_cache: dict[tuple[str, str], tuple[Any, dict]] = {}

    # Startup warnings for missing policy files
    if self._scanned_policies is not None:
        if config.default_policy and config.default_policy not in self._scanned_policies:
            logger.warning(
                "default_policy %r not found in %s",
                config.default_policy,
                policies_dir,
            )
        for group in config.policy_groups:
            if group.policy_path not in self._scanned_policies:
                logger.warning(
                    "PolicyGroup policy_path %r not found in %s",
                    group.policy_path,
                    policies_dir,
                )

---

skip_middleware

Decorator to mark a route as excluded from authorization middleware.

The decorated endpoint will not be checked by TopazMiddleware, allowing you to implement custom authorization logic.

from fastapi_topaz import skip_middleware

@app.post("/documents/bulk-import")
@skip_middleware
async def bulk_import(
    _=Depends(require_policy_allowed(config, "myapp.admin.bulk_import")),
):
    # Custom policy path, not auto-generated
    ...

Source code in src/fastapi_topaz/middleware.py

def skip_middleware(func: Callable) -> Callable:
    """
    Decorator to mark a route as excluded from authorization middleware.

    The decorated endpoint will not be checked by TopazMiddleware,
    allowing you to implement custom authorization logic.

    ```python
    from fastapi_topaz import skip_middleware

    @app.post("/documents/bulk-import")
    @skip_middleware
    async def bulk_import(
        _=Depends(require_policy_allowed(config, "myapp.admin.bulk_import")),
    ):
        # Custom policy path, not auto-generated
        ...
    ```
    """
    func.__skip_topaz_middleware__ = True  # type: ignore[attr-defined]
    return func

---

SkipMiddleware

Marker dependency to skip authorization middleware for a router or route.

Use as a dependency on a router to exclude all routes from middleware authorization:

from fastapi import APIRouter, Depends
from fastapi_topaz import SkipMiddleware

public_router = APIRouter(
    prefix="/api/public",
    dependencies=[Depends(SkipMiddleware)],
)

@public_router.get("/status")  # Automatically excluded from middleware
async def public_status():
    return {"status": "ok"}

Source code in src/fastapi_topaz/middleware.py

def __init__(self) -> None:
    pass

---

Circuit Breaker

CircuitBreaker

Circuit breaker configuration for Topaz authorization.

Prevents cascading failures when Topaz is unavailable by opening the circuit after a threshold of failures, using fallback strategies, and automatically testing for recovery.

PARAMETER	DESCRIPTION
failure_threshold	Number of consecutive failures before opening circuit TYPE: int DEFAULT: 5
success_threshold	Number of successes in half-open before closing TYPE: int DEFAULT: 2
recovery_timeout	Seconds to wait before transitioning to half-open TYPE: float DEFAULT: 30.0
fallback	Strategy when circuit is open ("cache_then_deny", "cache_then_allow", "deny", "allow", or custom callable) TYPE: FallbackStrategy DEFAULT: 'cache_then_deny'
serve_stale_cache	Whether to serve expired cache entries when open TYPE: bool DEFAULT: True
stale_cache_ttl	Maximum age (seconds) of stale cache to serve TYPE: float DEFAULT: 300.0
failure_exceptions	Exception types that count as failures TYPE: list[type] DEFAULT: (lambda: [ConnectionError, TimeoutError, OSError])()
failure_grpc_codes	gRPC status codes that count as failures (default: UNAVAILABLE, DEADLINE_EXCEEDED, UNKNOWN, INTERNAL, RESOURCE_EXHAUSTED) TYPE: set DEFAULT: (lambda: {UNAVAILABLE, DEADLINE_EXCEEDED, UNKNOWN, INTERNAL, RESOURCE_EXHAUSTED})()
on_state_change	Callback when circuit state changes TYPE: Callable[[str, str, str], None] | None DEFAULT: None
on_fallback	Callback when fallback is used TYPE: Callable[[Request, str, bool | None, bool], None] | None DEFAULT: None
half_open_max_requests	Number of test requests allowed in half-open state TYPE: int DEFAULT: 1

Example

config = TopazConfig(
    ...,
    circuit_breaker=CircuitBreaker(
        failure_threshold=5,
        recovery_timeout=30,
        fallback="cache_then_deny",
    ),
)

state property

state: CircuitState

Get current circuit state.

status

status() -> CircuitStatus

Get current circuit status for health checks.

Source code in src/fastapi_topaz/circuit_breaker.py

def status(self) -> CircuitStatus:
    """Get current circuit status for health checks."""
    return CircuitStatus(
        state=self._state.value,
        failure_count=self._failure_count,
        success_count=self._success_count,
        last_failure_time=self._last_failure_time,
        last_success_time=self._last_success_time,
        open_since=self._open_since,
    )

admit async

admit() -> Admission | None

Let a request through to Topaz, or refuse it.

Returns an :class:Admission when the circuit is closed, or when a half-open probe slot is free (probe=True). Returns None when the circuit is open, or half-open with every probe slot taken, and the fallback should be used. Pass the ticket back when recording the result.

Source code in src/fastapi_topaz/circuit_breaker.py

async def admit(self) -> Admission | None:
    """
    Let a request through to Topaz, or refuse it.

    Returns an :class:`Admission` when the circuit is closed, or when a
    half-open probe slot is free (``probe=True``). Returns ``None`` when the
    circuit is open, or half-open with every probe slot taken, and the
    fallback should be used. Pass the ticket back when recording the result.
    """
    async with self._lock:
        if self._state == CircuitState.CLOSED:
            return Admission(probe=False, epoch=self._half_open_epoch)

        if self._state == CircuitState.OPEN:
            # Check if recovery timeout has passed
            if self._open_since is not None:
                elapsed = time.monotonic() - self._open_since
                if elapsed >= self.recovery_timeout:
                    await self._transition_to(
                        CircuitState.HALF_OPEN, "recovery_timeout_expired"
                    )
                    self._half_open_requests = 1
                    return Admission(probe=True, epoch=self._half_open_epoch)
            return None

        if self._state == CircuitState.HALF_OPEN:
            # Allow limited requests in half-open state
            if self._half_open_requests < self.half_open_max_requests:
                self._half_open_requests += 1
                return Admission(probe=True, epoch=self._half_open_epoch)
            return None

        return None

should_allow_request async

should_allow_request() -> bool

Check if a request should be allowed through to Topaz.

Returns True if the circuit is closed or if we should test in half-open. Returns False if the circuit is open and fallback should be used. Prefer :meth:admit, whose ticket lets the breaker tell probes apart.

Source code in src/fastapi_topaz/circuit_breaker.py

async def should_allow_request(self) -> bool:
    """
    Check if a request should be allowed through to Topaz.

    Returns True if the circuit is closed or if we should test in half-open.
    Returns False if the circuit is open and fallback should be used.
    Prefer :meth:`admit`, whose ticket lets the breaker tell probes apart.
    """
    return await self.admit() is not None

record_success async

record_success(admission: Admission | None = None) -> None

Record a successful authorization call.

PARAMETER	DESCRIPTION
admission	The ticket from :meth:admit for this call. In half-open state only a current probe frees its slot and counts toward success_threshold. TYPE: Admission | None DEFAULT: None

Source code in src/fastapi_topaz/circuit_breaker.py

async def record_success(self, admission: Admission | None = None) -> None:
    """Record a successful authorization call.

    Args:
        admission: The ticket from :meth:`admit` for this call. In
            half-open state only a current probe frees its slot and counts
            toward ``success_threshold``.
    """
    async with self._lock:
        self._last_success_time = time.monotonic()
        self._failure_count = 0

        if self._state == CircuitState.HALF_OPEN and self._is_current_probe(admission):
            self._half_open_requests = max(0, self._half_open_requests - 1)
            self._success_count += 1
            if self._success_count >= self.success_threshold:
                await self._transition_to(CircuitState.CLOSED, "test_succeeded")

record_failure async

record_failure(exception: Exception, admission: Admission | None = None) -> None

Record a failed authorization call.

PARAMETER	DESCRIPTION
exception	The failure. TYPE: Exception
admission	The ticket from :meth:admit. In half-open state only a current probe reopens the circuit; a call admitted earlier (e.g. while closed) carries no evidence about recovery. TYPE: Admission | None DEFAULT: None

Source code in src/fastapi_topaz/circuit_breaker.py

async def record_failure(
    self, exception: Exception, admission: Admission | None = None
) -> None:
    """Record a failed authorization call.

    Args:
        exception: The failure.
        admission: The ticket from :meth:`admit`. In half-open state only a
            current probe reopens the circuit; a call admitted earlier
            (e.g. while closed) carries no evidence about recovery.
    """
    async with self._lock:
        self._last_failure_time = time.monotonic()
        self._failure_count += 1

        logger.warning(f"Circuit breaker recorded failure #{self._failure_count}: {exception}")

        if self._state == CircuitState.CLOSED:
            self._success_count = 0
            if self._failure_count >= self.failure_threshold:
                await self._transition_to(CircuitState.OPEN, "failure_threshold_exceeded")
        elif self._state == CircuitState.HALF_OPEN and self._is_current_probe(admission):
            self._success_count = 0
            await self._transition_to(CircuitState.OPEN, "test_failed")

release_probe

release_probe(admission: Admission | None = None) -> None

Free a half-open probe slot for a call that ended without a verdict.

PARAMETER	DESCRIPTION
admission	The ticket from :meth:admit; only a current probe frees a slot. TYPE: Admission | None DEFAULT: None

Source code in src/fastapi_topaz/circuit_breaker.py

def release_probe(self, admission: Admission | None = None) -> None:
    """Free a half-open probe slot for a call that ended without a verdict.

    Args:
        admission: The ticket from :meth:`admit`; only a current probe
            frees a slot.
    """
    # Sync and lock-free so it is safe while CancelledError unwinds; all
    # mutations run on the event loop thread.
    if (
        self._state == CircuitState.HALF_OPEN
        and self._half_open_requests > 0
        and self._is_current_probe(admission)
    ):
        self._half_open_requests -= 1

get_fallback_decision async

get_fallback_decision(request: Request, policy_path: str, resource_context: dict[str, Any], cached_decision: bool | None, error: Exception) -> bool

Get the fallback decision when circuit is open.

PARAMETER	DESCRIPTION
request	The FastAPI request TYPE: Request
policy_path	The policy path being checked TYPE: str
resource_context	The resource context TYPE: dict[str, Any]
cached_decision	Cached decision (may be stale), or None if not cached TYPE: bool | None
error	The exception that caused the fallback TYPE: Exception

RETURNS	DESCRIPTION
bool	Authorization decision (True/False)

Source code in src/fastapi_topaz/circuit_breaker.py

async def get_fallback_decision(
    self,
    request: Request,
    policy_path: str,
    resource_context: dict[str, Any],
    cached_decision: bool | None,
    error: Exception,
) -> bool:
    """
    Get the fallback decision when circuit is open.

    Args:
        request: The FastAPI request
        policy_path: The policy path being checked
        resource_context: The resource context
        cached_decision: Cached decision (may be stale), or None if not cached
        error: The exception that caused the fallback

    Returns:
        Authorization decision (True/False)
    """
    # Check no_stale_for patterns
    if cached_decision is not None and self.no_stale_for:
        import fnmatch

        for pattern in self.no_stale_for:
            if fnmatch.fnmatch(policy_path, pattern):
                cached_decision = None
                break

    if callable(self.fallback):
        result = self.fallback(request, policy_path, resource_context, cached_decision, error)
        # Support both sync and async callables
        if asyncio.iscoroutine(result):
            result = await result  # type: ignore[misc]
        return result

    strategy = self.fallback

    if strategy == "cache_then_deny":
        if cached_decision is not None:
            return cached_decision
        return False

    if strategy == "cache_then_allow":
        if cached_decision is not None:
            return cached_decision
        return True

    if strategy == "deny":
        return False

    if strategy == "allow":
        return True

    # Default to deny for unknown strategies
    logger.error(f"Unknown fallback strategy: {strategy}, defaulting to deny")
    return False

reset async

reset() -> None

Reset the circuit breaker to initial state.

Source code in src/fastapi_topaz/circuit_breaker.py

async def reset(self) -> None:
    """Reset the circuit breaker to initial state."""
    async with self._lock:
        await self._transition_to(CircuitState.CLOSED, "manual_reset")
        self._failure_count = 0
        self._success_count = 0
        self._last_failure_time = None
        self._last_success_time = None
        self._open_since = None
        self._half_open_requests = 0

is_failure_exception

is_failure_exception(exc: Exception) -> bool

Check if an exception should count as a circuit breaker failure.

gRPC errors (including grpc.aio.AioRpcError) count as failures only when their status code is in failure_grpc_codes, so policy errors like INVALID_ARGUMENT or NOT_FOUND do not trip the breaker.

Source code in src/fastapi_topaz/circuit_breaker.py

def is_failure_exception(self, exc: Exception) -> bool:
    """Check if an exception should count as a circuit breaker failure.

    gRPC errors (including grpc.aio.AioRpcError) count as failures only
    when their status code is in failure_grpc_codes, so policy errors
    like INVALID_ARGUMENT or NOT_FOUND do not trip the breaker.
    """
    if isinstance(exc, grpc.RpcError):
        try:
            return exc.code() in self.failure_grpc_codes
        except Exception:
            return True
    return any(isinstance(exc, exc_type) for exc_type in self.failure_exceptions)

---

Admission

A ticket for one call let through by :meth:CircuitBreaker.admit.

Pass it back to record_success / record_failure / release_probe so that, in half-open state, only the calls admitted as probes for the current half-open period free probe slots and decide recovery.

---

CircuitState

Bases: Enum

Circuit breaker states.

---

CircuitStatus

Current status of the circuit breaker for health checks.

is_open property

is_open: bool

Check if circuit is currently open.

---

Audit Logging

AuditLogger

Audit logger for authorization decisions.

PARAMETER	DESCRIPTION
log_allowed	Log successful authorizations TYPE: bool DEFAULT: True
log_denied	Log denied authorizations TYPE: bool DEFAULT: True
log_skipped	Log skipped/excluded routes TYPE: bool DEFAULT: False
log_unauthenticated	Log 401 events TYPE: bool DEFAULT: True
log_manual_checks	Log is_allowed(), check_relations() TYPE: bool DEFAULT: False
level_allowed	Log level for allowed events TYPE: str DEFAULT: 'INFO'
level_denied	Log level for denied events TYPE: str DEFAULT: 'WARNING'
level_unauthenticated	Log level for 401 events TYPE: str DEFAULT: 'WARNING'
level_skipped	Log level for skipped events TYPE: str DEFAULT: 'DEBUG'
include_resource_context	Include resource context in logs. Off by default: resource context often carries user data (emails, names, document attributes) that should not land in logs unreviewed. TYPE: bool DEFAULT: False
include_request_headers	Include HTTP headers (privacy concern) TYPE: bool DEFAULT: False
handler	Custom async handler for events TYPE: AuditHandler | None DEFAULT: None

Note

The client IP recorded in events is taken from the x-forwarded-for / x-real-ip headers when present, which are client-spoofable unless a trusted reverse proxy strips or sets them. Do not treat the audit client_ip as authenticated for forensics unless your deployment guarantees those headers.

log_decision async

log_decision(request: Request | None, policy_path: str, allowed: bool, *, source: str = 'dependency', check_type: str = 'policy', cached: bool = False, latency_ms: float | None = None, identity_type: str | None = None, identity_value: str | None = None, object_type: str | None = None, object_id: str | None = None, relation: str | None = None, subject_type: str | None = None, resource_context: dict[str, Any] | None = None, policy_resolution_source: str | None = None, reason: str | None = None) -> None

Log an authorization decision.

Source code in src/fastapi_topaz/audit.py

async def log_decision(
    self,
    request: Request | None,
    policy_path: str,
    allowed: bool,
    *,
    source: str = "dependency",
    check_type: str = "policy",
    cached: bool = False,
    latency_ms: float | None = None,
    identity_type: str | None = None,
    identity_value: str | None = None,
    object_type: str | None = None,
    object_id: str | None = None,
    relation: str | None = None,
    subject_type: str | None = None,
    resource_context: dict[str, Any] | None = None,
    policy_resolution_source: str | None = None,
    reason: str | None = None,
) -> None:
    """Log an authorization decision."""
    if source == "manual" and not self.log_manual_checks:
        return
    if allowed and not self.log_allowed:
        return
    if not allowed and not self.log_denied:
        return

    decision = "allowed" if allowed else "denied"
    event_name = f"authorization.{source}.{decision}"
    level = self.level_allowed if allowed else self.level_denied

    event = AuditEvent(
        event=event_name,
        level=level,
        request_id=self._get_request_id(request),
        source=source,
        identity_type=identity_type,
        identity_value=identity_value,
        policy_path=policy_path,
        decision=decision,
        check_type=check_type,
        cached=cached,
        latency_ms=latency_ms,
        method=request.method if request else None,
        path=str(request.url.path) if request else None,
        client_ip=self._get_client_ip(request),
        object_type=object_type,
        object_id=object_id,
        relation=relation,
        subject_type=subject_type,
        resource_context=resource_context if self.include_resource_context else None,
        policy_resolution_source=policy_resolution_source,
        reason=reason,
        request_headers=self._collect_headers(request),
    )

    await self._emit(event)

log_batch_check async

log_batch_check(request: Request | None, object_type: str, object_id: str, results: dict[str, bool], *, latency_ms: float | None = None, identity_value: str | None = None) -> None

Log batch relation check results.

Source code in src/fastapi_topaz/audit.py

async def log_batch_check(
    self,
    request: Request | None,
    object_type: str,
    object_id: str,
    results: dict[str, bool],
    *,
    latency_ms: float | None = None,
    identity_value: str | None = None,
) -> None:
    """Log batch relation check results."""
    if not self.log_manual_checks:
        return

    event = AuditEvent(
        event="authorization.check.relations",
        level=self.level_allowed,
        request_id=self._get_request_id(request),
        source="manual",
        identity_value=identity_value,
        check_type="rebac_batch",
        latency_ms=latency_ms,
        method=request.method if request else None,
        path=str(request.url.path) if request else None,
        client_ip=self._get_client_ip(request),
        object_type=object_type,
        object_id=object_id,
        results=results,
    )

    await self._emit(event)

log_unauthenticated_event async

log_unauthenticated_event(request: Request | None, reason: str = 'missing_identity') -> None

Log unauthenticated access attempt.

Source code in src/fastapi_topaz/audit.py

async def log_unauthenticated_event(
    self,
    request: Request | None,
    reason: str = "missing_identity",
) -> None:
    """Log unauthenticated access attempt."""
    if not self.log_unauthenticated:
        return

    event = AuditEvent(
        event="authorization.middleware.unauthenticated",
        level=self.level_unauthenticated,
        request_id=self._get_request_id(request),
        source="middleware",
        anonymous=True,
        method=request.method if request else None,
        path=str(request.url.path) if request else None,
        client_ip=self._get_client_ip(request),
        reason=reason,
    )

    await self._emit(event)

---

AuditEvent

Structured audit event for authorization decisions.

to_dict

to_dict() -> dict[str, Any]

Convert to structured dict for logging.

Source code in src/fastapi_topaz/audit.py

def to_dict(self) -> dict[str, Any]:
    """Convert to structured dict for logging."""
    data: dict[str, Any] = {
        "timestamp": self.timestamp,
        "level": self.level,
        "event": self.event,
        "source": self.source,
    }
    if self.request_id:
        data["request_id"] = self.request_id

    # Identity block
    if self.identity_value is not None:
        data["identity"] = {
            "type": self.identity_type,
            "value": self.identity_value,
            "anonymous": self.anonymous,
        }
    elif self.anonymous:
        data["identity"] = None

    # Authorization block
    auth: dict[str, Any] = {}
    if self.policy_path:
        auth["policy_path"] = self.policy_path
    if self.decision:
        auth["decision"] = self.decision
    if self.check_type != "policy":
        auth["check_type"] = self.check_type
    if self.cached:
        auth["cached"] = True
    if self.latency_ms is not None:
        auth["latency_ms"] = round(self.latency_ms, 2)
    if self.policy_resolution_source:
        auth["resolution_source"] = self.policy_resolution_source
    if auth:
        data["authorization"] = auth

    # Request block
    req: dict[str, Any] = {}
    if self.method:
        req["method"] = self.method
    if self.path:
        req["path"] = self.path
    if self.route_pattern:
        req["route_pattern"] = self.route_pattern
    if self.client_ip:
        req["ip"] = self.client_ip
    if self.request_headers:
        req["headers"] = self.request_headers
    if req:
        data["request"] = req

    # Resource block (ReBAC)
    if self.object_type or self.object_id or self.relation:
        data["resource"] = {
            k: v
            for k, v in {
                "object_type": self.object_type,
                "object_id": self.object_id,
                "relation": self.relation,
                "subject_type": self.subject_type,
            }.items()
            if v is not None
        }

    # Additional fields
    if self.reason:
        data["reason"] = self.reason
    if self.results:
        data["results"] = self.results
    if self.resource_context:
        data["resource_context"] = self.resource_context

    return data

to_json

to_json() -> str

Convert to JSON string.

Source code in src/fastapi_topaz/audit.py

def to_json(self) -> str:
    """Convert to JSON string."""
    return json.dumps(self.to_dict(), default=str)

---

Observability

PrometheusMetrics

Prometheus metrics collector for authorization decisions.

When two instances share a registry (e.g. two PrometheusMetrics created with the default global REGISTRY), the second instance reuses the collectors already registered under the same metric names instead of raising ValueError: Duplicated timeseries.

PARAMETER	DESCRIPTION
prefix	Metric name prefix (default: "topaz") TYPE: str DEFAULT: 'topaz'
include_policy_path	Add policy_path label (high cardinality) TYPE: bool DEFAULT: False
include_object_type	Add object_type label (medium cardinality) TYPE: bool DEFAULT: False
include_relation	Add relation label (medium cardinality) TYPE: bool DEFAULT: False
latency_buckets	Histogram buckets for latency TYPE: list[float] DEFAULT: (lambda: [0.001, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5])()
registry	Custom prometheus registry (default: global) TYPE: Any DEFAULT: None

record_auth_request

record_auth_request(source: str, decision: str, check_type: str, policy_path: str | None = None) -> None

Record an authorization request.

Source code in src/fastapi_topaz/observability.py

def record_auth_request(
    self,
    source: str,
    decision: str,
    check_type: str,
    policy_path: str | None = None,
) -> None:
    """Record an authorization request."""
    self._initialize()
    if not PROMETHEUS_AVAILABLE or not self._auth_requests:
        return

    labels = {"source": source, "decision": decision, "check_type": check_type}
    if self.include_policy_path and policy_path:
        labels["policy_path"] = policy_path

    self._auth_requests.labels(**labels).inc()

record_cache_hit

record_cache_hit(source: str) -> None

Record a cache hit.

Source code in src/fastapi_topaz/observability.py

def record_cache_hit(self, source: str) -> None:
    """Record a cache hit."""
    self._initialize()
    if not PROMETHEUS_AVAILABLE or not self._cache_hits:
        return
    self._cache_hits.labels(source=source).inc()

record_cache_miss

record_cache_miss(source: str) -> None

Record a cache miss.

Source code in src/fastapi_topaz/observability.py

def record_cache_miss(self, source: str) -> None:
    """Record a cache miss."""
    self._initialize()
    if not PROMETHEUS_AVAILABLE or not self._cache_misses:
        return
    self._cache_misses.labels(source=source).inc()

record_latency

record_latency(latency_seconds: float, source: str, cached: bool, policy_path: str | None = None) -> None

Record authorization latency.

Source code in src/fastapi_topaz/observability.py

def record_latency(
    self,
    latency_seconds: float,
    source: str,
    cached: bool,
    policy_path: str | None = None,
) -> None:
    """Record authorization latency."""
    self._initialize()
    if not PROMETHEUS_AVAILABLE or not self._auth_latency:
        return

    labels = {"source": source, "cached": str(cached).lower()}
    if self.include_policy_path and policy_path:
        labels["policy_path"] = policy_path

    self._auth_latency.labels(**labels).observe(latency_seconds)

record_topaz_latency

record_topaz_latency(latency_seconds: float) -> None

Record actual Topaz call latency.

Source code in src/fastapi_topaz/observability.py

def record_topaz_latency(self, latency_seconds: float) -> None:
    """Record actual Topaz call latency."""
    self._initialize()
    if not PROMETHEUS_AVAILABLE or not self._topaz_latency:
        return
    self._topaz_latency.observe(latency_seconds)

record_error

record_error(error_type: str) -> None

Record an authorization error.

Source code in src/fastapi_topaz/observability.py

def record_error(self, error_type: str) -> None:
    """Record an authorization error."""
    self._initialize()
    if not PROMETHEUS_AVAILABLE or not self._errors:
        return
    self._errors.labels(error_type=error_type).inc()

set_circuit_state

set_circuit_state(state: int) -> None

Set circuit breaker state (0=closed, 1=open, 2=half_open).

Source code in src/fastapi_topaz/observability.py

def set_circuit_state(self, state: int) -> None:
    """Set circuit breaker state (0=closed, 1=open, 2=half_open)."""
    self._initialize()
    if not PROMETHEUS_AVAILABLE or not self._circuit_state:
        return
    self._circuit_state.set(state)

record_circuit_transition

record_circuit_transition(from_state: str, to_state: str) -> None

Record circuit state transition.

Source code in src/fastapi_topaz/observability.py

def record_circuit_transition(self, from_state: str, to_state: str) -> None:
    """Record circuit state transition."""
    self._initialize()
    if not PROMETHEUS_AVAILABLE or not self._circuit_transitions:
        return
    self._circuit_transitions.labels(from_state=from_state, to_state=to_state).inc()

record_fallback

record_fallback(strategy: str, cache_hit: bool, decision: str) -> None

Record circuit breaker fallback.

Source code in src/fastapi_topaz/observability.py

def record_fallback(self, strategy: str, cache_hit: bool, decision: str) -> None:
    """Record circuit breaker fallback."""
    self._initialize()
    if not PROMETHEUS_AVAILABLE or not self._fallback:
        return
    self._fallback.labels(
        strategy=strategy,
        cache_hit=str(cache_hit).lower(),
        decision=decision,
    ).inc()

set_cache_size

set_cache_size(size: int) -> None

Set current cache size.

Source code in src/fastapi_topaz/observability.py

def set_cache_size(self, size: int) -> None:
    """Set current cache size."""
    self._initialize()
    if not PROMETHEUS_AVAILABLE or not self._cache_size:
        return
    self._cache_size.set(size)

---

OTelTracing

OpenTelemetry tracing for authorization decisions.

PARAMETER	DESCRIPTION
trace_all_checks	Trace all authorization checks TYPE: bool DEFAULT: True
trace_cache_operations	Include cache lookup/store spans TYPE: bool DEFAULT: True
include_identity	Add identity to span attributes (privacy risk) TYPE: bool DEFAULT: False
include_policy_path	Add policy_path to spans TYPE: bool DEFAULT: False
include_resource_context	Add full resource context (privacy risk) TYPE: bool DEFAULT: False
span_name_prefix	Prefix for span names TYPE: str DEFAULT: 'topaz'

start_auth_span

start_auth_span(source: str, check_type: str, policy_path: str | None = None, identity_value: str | None = None) -> Any

Start an authorization span.

Source code in src/fastapi_topaz/observability.py

def start_auth_span(
    self,
    source: str,
    check_type: str,
    policy_path: str | None = None,
    identity_value: str | None = None,
) -> Any:
    """Start an authorization span."""
    tracer = self._get_tracer()
    if not tracer or not self.trace_all_checks:
        return None

    attributes = {
        f"{self.span_name_prefix}.source": source,
        f"{self.span_name_prefix}.check_type": check_type,
    }

    if self.include_policy_path and policy_path:
        attributes[f"{self.span_name_prefix}.policy_path"] = policy_path

    if self.include_identity and identity_value:
        attributes[f"{self.span_name_prefix}.identity"] = identity_value

    return tracer.start_span(
        f"{self.span_name_prefix}.authorization",
        attributes=attributes,
    )

end_auth_span

end_auth_span(span: Any, decision: str, cached: bool, latency_ms: float, resource_context: dict | None = None) -> None

End an authorization span with results.

Source code in src/fastapi_topaz/observability.py

def end_auth_span(
    self,
    span: Any,
    decision: str,
    cached: bool,
    latency_ms: float,
    resource_context: dict | None = None,
) -> None:
    """End an authorization span with results."""
    if not span or not OTEL_AVAILABLE:
        return

    span.set_attribute(f"{self.span_name_prefix}.decision", decision)
    span.set_attribute(f"{self.span_name_prefix}.cached", cached)
    span.set_attribute(f"{self.span_name_prefix}.latency_ms", latency_ms)

    if self.include_resource_context and resource_context:
        span.set_attribute(
            f"{self.span_name_prefix}.resource_context",
            str(resource_context),
        )

    span.set_attribute(f"{self.span_name_prefix}.denied", decision == "denied")
    span.set_status(Status(StatusCode.OK))

    span.end()

start_cache_span

start_cache_span(operation: str) -> Any

Start a cache operation span.

Source code in src/fastapi_topaz/observability.py

def start_cache_span(self, operation: str) -> Any:
    """Start a cache operation span."""
    tracer = self._get_tracer()
    if not tracer or not self.trace_cache_operations:
        return None

    return tracer.start_span(f"{self.span_name_prefix}.cache.{operation}")

end_cache_span

end_cache_span(span: Any, hit: bool | None = None) -> None

End a cache operation span.

Source code in src/fastapi_topaz/observability.py

def end_cache_span(self, span: Any, hit: bool | None = None) -> None:
    """End a cache operation span."""
    if not span or not OTEL_AVAILABLE:
        return

    if hit is not None:
        span.set_attribute("hit", hit)

    span.set_status(Status(StatusCode.OK))
    span.end()

start_topaz_span

start_topaz_span() -> Any

Start a Topaz request span.

Source code in src/fastapi_topaz/observability.py

def start_topaz_span(self) -> Any:
    """Start a Topaz request span."""
    tracer = self._get_tracer()
    if not tracer or not self.trace_all_checks:
        return None

    return tracer.start_span(f"{self.span_name_prefix}.topaz.request")

end_topaz_span

end_topaz_span(span: Any, latency_ms: float) -> None

End a Topaz request span.

Source code in src/fastapi_topaz/observability.py

def end_topaz_span(self, span: Any, latency_ms: float) -> None:
    """End a Topaz request span."""
    if not span or not OTEL_AVAILABLE:
        return

    span.set_attribute("latency_ms", latency_ms)
    span.set_status(Status(StatusCode.OK))
    span.end()

record_error

record_error(span: Any, error: Exception) -> None

Record an error on a span.

Source code in src/fastapi_topaz/observability.py

def record_error(self, span: Any, error: Exception) -> None:
    """Record an error on a span."""
    if not span or not OTEL_AVAILABLE:
        return

    span.set_status(Status(StatusCode.ERROR, str(error)))
    span.record_exception(error)
    span.end()

get_current_trace_id

get_current_trace_id() -> str | None

Get current trace ID for correlation.

Source code in src/fastapi_topaz/observability.py

def get_current_trace_id(self) -> str | None:
    """Get current trace ID for correlation."""
    if not OTEL_AVAILABLE:
        return None

    span = trace.get_current_span()  # type: ignore[union-attr]
    if span and span.get_span_context().is_valid:
        return format(span.get_span_context().trace_id, "032x")
    return None

---

Policy Generation

annotate_openapi

Annotate FastAPI routes with their resolved Topaz policy paths in OpenAPI.

Walks every route (including included routers) and resolves each route through the same policy resolution chain as :func:generate_rights_matrix (explicit file when policies_dir is given > policy group > default policy > generated), then merges x-authz-policy and x-authz-source into each route's openapi_extra. Existing openapi_extra keys are preserved. When the methods of one route resolve differently, both keys are written as {method: value} maps sorted by method; otherwise they are strings. Routes TopazMiddleware does not authorize (@skip_middleware, Depends(SkipMiddleware), or the installed middleware's exclude_paths / exclude_methods) get x-authz-source: skipped and no x-authz-policy entry.

Call after route registration and before the first schema build (the schema is cached on first access to app.openapi()). This includes routes added to a router after it was passed to include_router(): on FastAPI >= 0.137 their annotations are lost otherwise.

PARAMETER	DESCRIPTION
app	FastAPI application instance TYPE: FastAPI
config	TopazConfig with policy_path_root (and optional default_policy / policy_groups) TYPE: TopazConfig
policies_dir	Directory containing .rego policy files (optional, enables the explicit tier) TYPE: str | Path | None DEFAULT: None

RETURNS	DESCRIPTION
int	Number of routes annotated.

Source code in src/fastapi_topaz/codegen.py

def annotate_openapi(
    app: FastAPI,
    config: TopazConfig,
    policies_dir: str | Path | None = None,
) -> int:
    """
    Annotate FastAPI routes with their resolved Topaz policy paths in OpenAPI.

    Walks every route (including included routers) and resolves each route through the same policy
    resolution chain as :func:`generate_rights_matrix` (explicit file when
    *policies_dir* is given > policy group > default policy > generated),
    then merges ``x-authz-policy`` and ``x-authz-source`` into each route's
    ``openapi_extra``. Existing ``openapi_extra`` keys are preserved. When
    the methods of one route resolve differently, both keys are written as
    ``{method: value}`` maps sorted by method; otherwise they are strings.
    Routes TopazMiddleware does not authorize (``@skip_middleware``,
    ``Depends(SkipMiddleware)``, or the installed middleware's
    ``exclude_paths`` / ``exclude_methods``) get ``x-authz-source: skipped``
    and no ``x-authz-policy`` entry.

    Call after route registration and before the first schema build (the
    schema is cached on first access to ``app.openapi()``). This includes
    routes added to a router after it was passed to ``include_router()``:
    on FastAPI >= 0.137 their annotations are lost otherwise.

    Args:
        app: FastAPI application instance
        config: TopazConfig with policy_path_root (and optional
            default_policy / policy_groups)
        policies_dir: Directory containing ``.rego`` policy files (optional,
            enables the explicit tier)

    Returns:
        Number of routes annotated.
    """
    resolutions = generate_rights_matrix(app, config, policies_dir=policies_dir)
    by_method_path = {(r.method, r.route_pattern): r for r in resolutions}

    annotated = 0
    for route in iter_routes(app):
        # Only API routes carry openapi_extra; Starlette routes such as the
        # docs endpoints are not part of the schema
        if not isinstance(getattr(route, "original_route", route), APIRoute):
            continue
        api_route = cast(APIRoute, route)
        by_method = {
            method: by_method_path[(method, api_route.path)]
            for method in sorted(api_route.methods or [])
            if (method, api_route.path) in by_method_path
        }
        if not by_method:
            continue
        extra = {**(api_route.openapi_extra or {})}
        pairs = {(r.resolved_policy_path, r.resolution_source) for r in by_method.values()}
        # TopazMiddleware never checks skipped methods: they claim no policy
        checked = {m: r for m, r in by_method.items() if r.resolution_source != "skipped"}
        if len(pairs) == 1:
            policy, source = next(iter(pairs))
            extra["x-authz-source"] = source
            if checked:
                extra["x-authz-policy"] = policy
            else:
                extra.pop("x-authz-policy", None)
        else:
            # Methods resolve differently (e.g. only GET has an explicit file)
            extra["x-authz-source"] = {m: r.resolution_source for m, r in by_method.items()}
            if checked:
                extra["x-authz-policy"] = {m: r.resolved_policy_path for m, r in checked.items()}
            else:
                extra.pop("x-authz-policy", None)
        set_route_attr(route, "openapi_extra", extra)
        annotated += 1

    return annotated

---

Re-exported Types

These types are re-exported from aserto.client for convenience:

from fastapi_topaz import (
    AuthorizerOptions,
    Identity,
    IdentityType,
    ResourceContext,
)