API Reference¶

def authorize_query(
    stmt: Select[Any],
    *,
    actor: ActorLike,
    action: str,
    registry: PolicyRegistry | None = None,
) -> Select[Any]:
    """Apply authorization filters to a SQLAlchemy SELECT statement.

    Looks up registered policies for the statement's entities and the
    given action, evaluates them with the actor, and applies the
    resulting WHERE clauses.

    Args:
        stmt: A SQLAlchemy 2.0 Select statement.
        actor: The user/principal. Must satisfy ActorLike protocol.
        action: The action being performed (e.g., "read", "update").
        registry: Optional custom registry. Defaults to the global registry.

    Returns:
        A new Select with authorization filters applied.

    Example::

        stmt = select(Post).where(Post.category == "tech")
        stmt = authorize_query(stmt, actor=current_user, action="read")
        # SQL: SELECT ... WHERE category = 'tech'
        #      AND (is_published OR author_id = :id)
    """
    target_registry = registry if registry is not None else get_default_registry()

    check_unknown_action(target_registry, action)

    desc_list: list[dict[str, Any]] = stmt.column_descriptions
    for desc in desc_list:
        entity: type | None = desc.get("entity")
        if entity is None:
            continue

        filter_expr = evaluate_policies(target_registry, entity, action, actor)
        stmt = stmt.where(filter_expr)

    return stmt

def can(
    actor: ActorLike,
    action: str,
    resource: DeclarativeBase,
    *,
    registry: PolicyRegistry | None = None,
    session: Session | None = None,
) -> bool:
    """Check if *actor* can perform *action* on a specific resource instance.

    Returns ``True`` if the policy filter matches the resource, ``False``
    otherwise.  The real application database is **never** touched — the
    filter expression is evaluated in-memory by walking the SQLAlchemy
    ColumnElement AST.

    Args:
        actor: The user/principal performing the action.
        action: The action string (e.g., ``"read"``, ``"update"``).
        resource: A mapped SQLAlchemy model instance.
        registry: Optional custom registry.  Defaults to the global registry.
        session: Optional session (reserved for future use).

    Returns:
        ``True`` if access is granted, ``False`` if denied.

    Raises:
        QueryOnlyPolicyError: If any matching policy is marked
            ``query_only=True``.

    Example::

        post = session.get(Post, 1)
        if can(current_user, "read", post):
            return post
    """
    target_registry = registry if registry is not None else get_default_registry()

    check_unknown_action(target_registry, action)

    resource_type = type(resource)

    # Check for query-only policies before attempting in-memory evaluation
    policies = target_registry.lookup(resource_type, action)
    if any(p.query_only for p in policies):
        raise QueryOnlyPolicyError(
            resource_type=resource_type.__name__,
            action=action,
            query_only_policies=[p.name for p in policies if p.query_only],
        )

    filter_expr = evaluate_policies(
        target_registry, resource_type, action, actor, policies=policies
    )

    return eval_expression(filter_expr, resource)

def can_create(
    actor: ActorLike,
    resource: DeclarativeBase,
    *,
    registry: PolicyRegistry | None = None,
    session: Session | None = None,
) -> bool:
    """Check whether *actor* can create *resource* in its current state.

    This is a convenience wrapper around ``can(..., action="create")`` for
    pending or transient ORM instances. Populate the object first, then call
    ``can_create()`` before flushing or committing.

    Args:
        actor: The user/principal attempting the create.
        resource: The pending mapped SQLAlchemy model instance.
        registry: Optional custom registry. Defaults to the global registry.
        session: Optional session (reserved for future use).

    Returns:
        ``True`` if creation is allowed, ``False`` otherwise.
    """
    return can(actor, CREATE, resource, registry=registry, session=session)

def authorize(
    actor: ActorLike,
    action: str,
    resource: DeclarativeBase,
    *,
    registry: PolicyRegistry | None = None,
    message: str | None = None,
    session: Session | None = None,
) -> None:
    """Assert that *actor* is authorized to perform *action* on *resource*.

    Raises :class:`~sqla_authz.exceptions.AuthorizationDenied` when access
    is denied.  Returns ``None`` on success.

    Args:
        actor: The user/principal performing the action.
        action: The action string (e.g., ``"read"``, ``"update"``).
        resource: A mapped SQLAlchemy model instance.
        registry: Optional custom registry.  Defaults to the global registry.
        message: Optional custom error message for the exception.
        session: Optional session (reserved for future use).

    Raises:
        AuthorizationDenied: If the actor is not authorized.
        QueryOnlyPolicyError: If any matching policy is marked
            ``query_only=True``.

    Example::

        authorize(current_user, "update", post)  # raises if denied
    """
    if not can(actor, action, resource, registry=registry, session=session):
        raise AuthorizationDenied(
            actor=actor,
            action=action,
            resource_type=type(resource).__name__,
            message=message,
        )

def authorize_create(
    actor: ActorLike,
    resource: DeclarativeBase,
    *,
    registry: PolicyRegistry | None = None,
    message: str | None = None,
    session: Session | None = None,
) -> None:
    """Assert that *actor* can create *resource* in its current state.

    This is a convenience wrapper around ``authorize(..., action="create")``
    for pending or transient ORM instances.

    Args:
        actor: The user/principal attempting the create.
        resource: The pending mapped SQLAlchemy model instance.
        registry: Optional custom registry. Defaults to the global registry.
        message: Optional custom error message for the exception.
        session: Optional session (reserved for future use).
    """
    authorize(
        actor,
        CREATE,
        resource,
        registry=registry,
        message=message,
        session=session,
    )

def configure(
    *,
    on_missing_policy: OnMissingPolicy | None = None,
    default_action: str | None = None,
    log_policy_decisions: bool | None = None,
    on_unloaded_relationship: OnUnloadedRelationship | None = None,
    strict_mode: bool | None = None,
    on_unprotected_get: OnBypassAction | None = None,
    on_text_query: OnBypassAction | None = None,
    on_skip_authz: OnSkipAuthz | None = None,
    audit_bypasses: bool | None = None,
    intercept_creates: bool | None = None,
    intercept_updates: bool | None = None,
    intercept_deletes: bool | None = None,
    on_write_denied: OnWriteDenied | None = None,
    on_unknown_action: OnUnknownAction | None = None,
) -> AuthzConfig:
    """Update the global configuration by merging overrides.

    Only non-None values are applied. Returns the new global config.

    Args:
        on_missing_policy: Set to ``"deny"`` or ``"raise"``.
        default_action: Set the default action string.
        log_policy_decisions: Enable/disable audit logging of policy decisions.
        on_unloaded_relationship: Set to ``"deny"``, ``"raise"``, or ``"warn"``.
        strict_mode: Enable strict mode (applies convenience defaults).
        on_unprotected_get: Set to ``"ignore"``, ``"warn"``, or ``"raise"``.
        on_text_query: Set to ``"ignore"``, ``"warn"``, or ``"raise"``.
        on_skip_authz: Set to ``"ignore"``, ``"warn"``, or ``"log"``.
        audit_bypasses: Enable/disable bypass audit logging.
        intercept_creates: Enable interception of ORM create/insert flushes.
        intercept_updates: Enable interception of UPDATE statements.
        intercept_deletes: Enable interception of DELETE statements.
        on_write_denied: Set to ``"raise"`` or ``"filter"``.
        on_unknown_action: Set to ``"ignore"``, ``"warn"``, or ``"raise"``.

    Returns:
        The updated global ``AuthzConfig``.

    Example::

        configure(on_missing_policy="raise")
        # Now missing policies raise NoPolicyError instead of denying
    """
    global _global_config
    _global_config = _global_config.merge(
        on_missing_policy=on_missing_policy,
        default_action=default_action,
        log_policy_decisions=log_policy_decisions,
        on_unloaded_relationship=on_unloaded_relationship,
        strict_mode=strict_mode,
        on_unprotected_get=on_unprotected_get,
        on_text_query=on_text_query,
        on_skip_authz=on_skip_authz,
        audit_bypasses=audit_bypasses,
        intercept_creates=intercept_creates,
        intercept_updates=intercept_updates,
        intercept_deletes=intercept_deletes,
        on_write_denied=on_write_denied,
        on_unknown_action=on_unknown_action,
    )
    return _global_config

def predicate(fn: Callable[..., ColumnElement[bool]]) -> Predicate:
    """Decorator/factory that creates a Predicate from a callable.

    Example::

        @predicate
        def is_published(actor: User) -> ColumnElement[bool]:
            return Post.is_published == True

        # Or as a factory:
        is_author = predicate(lambda actor: Post.author_id == actor.id)
    """
    return Predicate(fn, name=getattr(fn, "__name__", "<lambda>"))

def scope(
    applies_to: Sequence[type],
    *,
    actions: Sequence[str] | None = None,
    registry: PolicyRegistry | None = None,
) -> Callable[[F], F]:
    """Decorator that registers a cross-cutting scope filter.

    Scopes are AND'd with OR'd policy results for matching models.
    They enforce invariants like tenant isolation that must not be
    accidentally bypassed by adding a new policy.

    The decorated function receives ``(actor, Model)`` where ``Model``
    is the class currently being filtered, and returns a
    ``ColumnElement[bool]``.

    Args:
        applies_to: List of SQLAlchemy model classes this scope covers.
        actions: Optional list of actions to restrict this scope to.
            If ``None`` (default), the scope applies to all actions.
        registry: Optional custom registry. Defaults to the global registry.

    Returns:
        A decorator that registers the function and returns it unchanged.

    Example::

        @scope(applies_to=[Post, Comment, Document])
        def tenant_scope(actor: User, Model: type) -> ColumnElement[bool]:
            return Model.org_id == actor.org_id

        @scope(applies_to=[Post, Comment], actions=["read"])
        def soft_delete(actor: User, Model: type) -> ColumnElement[bool]:
            return Model.deleted_at.is_(None)
    """
    if not applies_to:
        raise ValueError("applies_to must be a non-empty sequence of model classes")

    def decorator(fn: F) -> F:
        _validate_scope_signature(fn)

        target = registry if registry is not None else get_default_registry()
        scope_reg = ScopeRegistration(
            applies_to=tuple(applies_to),
            fn=fn,
            name=fn.__name__,
            description=fn.__doc__ or "",
            actions=tuple(actions) if actions is not None else None,
        )
        target.register_scope(scope_reg)
        return fn

    return decorator