Decorators¶

Convenient decorators for protecting route handlers with feature flags. These decorators provide a declarative way to control access to endpoints based on flag evaluation.

Overview¶

Two decorators are provided:

@feature_flag: Conditionally execute handlers; returns alternative response when disabled
@require_flag: Require flag to be enabled; raises exception when disabled

Both decorators integrate seamlessly with Litestar’s dependency injection and automatically extract evaluation context from requests when middleware is enabled.

API Reference¶

Decorators for feature flag evaluation.

litestar_flags.decorators.feature_flag(flag_key, *, default=False, default_response=None, context_key=None)[source]¶

Conditionally execute route handlers based on a feature flag.

When the flag is disabled, the handler returns default_response instead of executing the handler function.

Parameters:

flag_key (str) – The feature flag key to evaluate.
default (bool) – Default value if flag is not found.
default_response (Any) – Response to return when flag is disabled.
context_key (str | None) – Optional request attribute to use as targeting key.

Return type:

Callable[[TypeVar(F, bound= Callable[..., Any])], TypeVar(F, bound= Callable[..., Any])]

Returns:

Decorated function.

Example

>>> @get("/new-feature")
>>> @feature_flag("new_feature", default_response={"error": "Not available"})
>>> async def new_feature_endpoint() -> dict:
...     return {"message": "New feature!"}

litestar_flags.decorators.require_flag(flag_key, *, default=False, context_key=None, error_message=None)[source]¶

Require a feature flag to be enabled for the decorated handler.

When the flag is disabled, raises NotAuthorizedException. This is useful for protecting beta or premium features.

Parameters:

flag_key (str) – The feature flag key to evaluate.
default (bool) – Default value if flag is not found.
context_key (str | None) – Optional request attribute to use as targeting key.
error_message (str | None) – Custom error message for the exception.

Return type:

Callable[[TypeVar(F, bound= Callable[..., Any])], TypeVar(F, bound= Callable[..., Any])]

Returns:

Decorated function.

Raises:

NotAuthorizedException – When the flag is disabled.

Example

>>> @get("/beta")
>>> @require_flag("beta_access", error_message="Beta access required")
>>> async def beta_endpoint() -> dict:
...     return {"message": "Welcome to beta!"}

feature_flag Decorator¶

Use @feature_flag when you want to return an alternative response instead of raising an exception.

Basic Usage¶

from litestar import get
from litestar_flags import feature_flag

@get("/new-feature")
@feature_flag("new_feature", default_response={"error": "Feature not available"})
async def new_feature_endpoint() -> dict:
    return {"message": "Welcome to the new feature!"}

With Default Value¶

@get("/beta-feature")
@feature_flag(
    "beta_access",
    default=False,  # Default if flag not found
    default_response={"status": "coming_soon"},
)
async def beta_endpoint() -> dict:
    return {"status": "active", "features": ["beta_1", "beta_2"]}

With Context Key¶

Specify which request attribute to use as the targeting key:

@get("/user/{user_id:str}/premium")
@feature_flag(
    "premium_features",
    context_key="user_id",  # Use path param as targeting key
    default_response={"error": "Premium access required"},
)
async def premium_endpoint(user_id: str) -> dict:
    return {"premium": True, "user_id": user_id}

Parameters¶

Parameter	Type	Description
`flag_key`	`str`	The feature flag key to evaluate
`default`	`bool`	Default value if flag is not found (default: `False`)
`default_response`	`Any`	Response to return when flag is disabled (default: `None`)
`context_key`	`str \| None`	Request attribute to use as targeting key (optional)

require_flag Decorator¶

Use @require_flag when you want to raise an exception for unauthorized access. This is useful for protecting premium or beta features.

Basic Usage¶

from litestar import get
from litestar_flags import require_flag

@get("/admin/dashboard")
@require_flag("admin_access")
async def admin_dashboard() -> dict:
    return {"admin": True, "stats": {...}}

With Custom Error Message¶

@get("/beta")
@require_flag(
    "beta_access",
    error_message="This feature is only available to beta testers",
)
async def beta_endpoint() -> dict:
    return {"beta": True}

With Context Key¶

@get("/org/{org_id:str}/billing")
@require_flag(
    "billing_v2",
    context_key="org_id",
    error_message="New billing is not enabled for your organization",
)
async def billing_endpoint(org_id: str) -> dict:
    return {"billing_version": 2}

Parameters¶

Parameter	Type	Description
`flag_key`	`str`	The feature flag key to evaluate
`default`	`bool`	Default value if flag is not found (default: `False`)
`context_key`	`str \| None`	Request attribute to use as targeting key (optional)
`error_message`	`str \| None`	Custom error message for the exception (optional)

Exception Behavior¶

When the flag is disabled, @require_flag raises NotAuthorizedException (HTTP 403):

# Client receives:
{
    "status_code": 403,
    "detail": "Feature 'beta_access' is not available"
}

Decorator Order¶

When combining with other decorators, place feature flag decorators after the route decorator:

from litestar import get
from litestar_flags import feature_flag, require_flag

# Correct order
@get("/feature")
@feature_flag("my_feature")
async def handler() -> dict:
    ...

# Also correct - multiple flags
@get("/premium-beta")
@require_flag("premium_access")
@require_flag("beta_access")
async def premium_beta_handler() -> dict:
    ...

Context Resolution¶

The decorators resolve evaluation context in this order:

Middleware context: If FeatureFlagsMiddleware is enabled, use extracted context
Context key: If context_key is specified, look up from request: - Path parameters (e.g., /users/{user_id}) - Query parameters (e.g., ?user_id=123) - Headers (e.g., X-User-ID) - User attributes (e.g., request.user.id)
User auth: If authenticated, use request.user.id or request.user.user_id
Default: Use default context if nothing else is available

Integration Examples¶

With Authentication¶

from litestar import get, Request
from litestar.connection import ASGIConnection
from litestar_flags import require_flag

@get("/settings")
@require_flag("settings_v2")
async def settings_endpoint(request: Request) -> dict:
    # The decorator automatically uses request.user.id as targeting key
    return {"user": str(request.user.id), "settings_version": 2}

With Guards¶

from litestar import get
from litestar.guards import Guard
from litestar_flags import require_flag

async def auth_guard(connection: ASGIConnection, _) -> None:
    if not connection.user:
        raise NotAuthorizedException()

@get("/protected", guards=[auth_guard])
@require_flag("protected_feature")
async def protected_endpoint() -> dict:
    return {"protected": True}

With Rate Limiting¶

from litestar import get
from litestar_flags import feature_flag

@get("/api/v2/data")
@feature_flag(
    "api_v2",
    default_response={"error": "API v2 not available", "use": "/api/v1/data"},
)
async def api_v2_endpoint() -> dict:
    return {"version": 2, "data": [...]}

A/B Testing Response¶

from litestar import get
from litestar_flags import FeatureFlagClient, EvaluationContext

@get("/checkout")
async def checkout_endpoint(
    feature_flags: FeatureFlagClient,
    user_id: str,
) -> dict:
    context = EvaluationContext(targeting_key=user_id)
    variant = await feature_flags.get_string_value(
        "checkout_experiment",
        default="control",
        context=context,
    )

    if variant == "treatment":
        return {"checkout_version": "new", "layout": "streamlined"}
    else:
        return {"checkout_version": "classic", "layout": "standard"}