Configuration¶

The FeatureFlagsConfig dataclass provides configuration options for the feature flags plugin, including storage backend selection and middleware settings.

Overview¶

Configuration is passed to the FeatureFlagsPlugin on initialization and controls:

Which storage backend to use (memory, database, or Redis)
Connection settings for external backends
Default evaluation context
Middleware and context extraction settings
Dependency injection key for the client

Quick Example¶

from litestar import Litestar
from litestar_flags import FeatureFlagsPlugin, FeatureFlagsConfig

# Simple in-memory configuration
config = FeatureFlagsConfig(backend="memory")

# Database configuration
config = FeatureFlagsConfig(
    backend="database",
    connection_string="postgresql+asyncpg://user:pass@localhost/db",
)

# Redis configuration
config = FeatureFlagsConfig(
    backend="redis",
    redis_url="redis://localhost:6379",
)

app = Litestar(
    plugins=[FeatureFlagsPlugin(config=config)],
)

API Reference¶

Configuration for the feature flags plugin.

class litestar_flags.config.FeatureFlagsConfig[source]¶

Bases: object

Configuration for the feature flags plugin.

backend¶: Storage backend type (“memory”, “database”, “redis”).

connection_string¶: Database connection string (when backend=”database”).

table_prefix¶: Prefix for database tables (when backend=”database”).

redis_url¶: Redis connection URL (when backend=”redis”).

redis_prefix¶: Prefix for Redis keys (when backend=”redis”).

default_context¶: Default evaluation context.

enable_middleware¶: Whether to enable the context extraction middleware.

context_extractor¶: Custom function to extract context from requests.

client_dependency_key¶: Key for dependency injection of the client.

enable_health_endpoint¶: Whether to register a health check endpoint.

health_endpoint_path¶: Path for the health check endpoint.

enable_resilience¶: Whether to enable circuit breaker and retry patterns.

circuit_breaker¶: Optional circuit breaker configuration.

retry_policy¶: Optional retry policy configuration.

Example

>>> config = FeatureFlagsConfig(
...     backend="database",
...     connection_string="postgresql+asyncpg://user:pass@localhost/db",
...     enable_health_endpoint=True,
... )

Parameters:

backend (Literal['memory', 'database', 'redis'])
connection_string (str | None)
table_prefix (str)
redis_url (str | None)
redis_prefix (str)
default_context (EvaluationContext | None)
enable_middleware (bool)
context_extractor (Callable[[Request], EvaluationContext] | None)
client_dependency_key (str)
enable_health_endpoint (bool)
health_endpoint_path (str)
enable_resilience (bool)
circuit_breaker (CircuitBreaker | None)
retry_policy (RetryPolicy | None)
extra (dict[str, Any])

backend: Literal['memory', 'database', 'redis']¶

connection_string: str | None¶

table_prefix: str¶

redis_url: str | None¶

redis_prefix: str¶

default_context: EvaluationContext | None¶

enable_middleware: bool¶

context_extractor: Callable[[Request], EvaluationContext] | None¶

client_dependency_key: str¶

enable_health_endpoint: bool¶

health_endpoint_path: str¶

enable_resilience: bool¶

circuit_breaker: CircuitBreaker | None¶

retry_policy: RetryPolicy | None¶

extra: dict[str, Any]¶

__init__(backend='memory', connection_string=None, table_prefix='ff_', redis_url=None, redis_prefix='feature_flags:', default_context=None, enable_middleware=False, context_extractor=None, client_dependency_key='feature_flags', enable_health_endpoint=False, health_endpoint_path='/flags/health', enable_resilience=False, circuit_breaker=None, retry_policy=None, extra=<factory>)¶

Parameters:

backend (Literal['memory', 'database', 'redis'])
connection_string (str | None)
table_prefix (str)
redis_url (str | None)
redis_prefix (str)
default_context (EvaluationContext | None)
enable_middleware (bool)
context_extractor (Callable[[Request], EvaluationContext] | None)
client_dependency_key (str)
enable_health_endpoint (bool)
health_endpoint_path (str)
enable_resilience (bool)
circuit_breaker (CircuitBreaker | None)
retry_policy (RetryPolicy | None)
extra (dict[str, Any])

Return type:

None

Configuration Options¶

Backend Selection¶

The backend parameter accepts one of three values:

"memory": In-memory storage (default). No persistence, ideal for development.
"database": SQLAlchemy-based persistent storage. Requires connection_string.
"redis": Redis-based distributed storage. Requires redis_url.

# Memory backend (default)
config = FeatureFlagsConfig(backend="memory")

# Database backend
config = FeatureFlagsConfig(
    backend="database",
    connection_string="postgresql+asyncpg://user:pass@localhost/db",
    table_prefix="ff_",  # Optional table prefix
)

# Redis backend
config = FeatureFlagsConfig(
    backend="redis",
    redis_url="redis://localhost:6379/0",
    redis_prefix="feature_flags:",  # Optional key prefix
)

Database Settings¶

When using backend="database":

Parameter	Required	Description
`connection_string`	Yes	SQLAlchemy async connection string
`table_prefix`	No	Prefix for database table names (default: `"ff_"`)

config = FeatureFlagsConfig(
    backend="database",
    connection_string="postgresql+asyncpg://user:pass@localhost/mydb",
    table_prefix="feature_flags_",
)

Redis Settings¶

When using backend="redis":

Parameter	Required	Description
`redis_url`	Yes	Redis connection URL
`redis_prefix`	No	Prefix for Redis keys (default: `"feature_flags:"`)

config = FeatureFlagsConfig(
    backend="redis",
    redis_url="redis://:password@localhost:6379/0",
    redis_prefix="myapp:flags:",
)

Middleware Configuration¶

Enable automatic context extraction from requests:

from litestar_flags import FeatureFlagsConfig, EvaluationContext

def my_context_extractor(request) -> EvaluationContext:
    """Extract evaluation context from the request."""
    user_id = None
    if hasattr(request, "user") and request.user:
        user_id = str(request.user.id)

    return EvaluationContext(
        targeting_key=user_id,
        user_id=user_id,
        ip_address=request.client.host if request.client else None,
    )

config = FeatureFlagsConfig(
    backend="memory",
    enable_middleware=True,
    context_extractor=my_context_extractor,
)

Default Context¶

Set a default evaluation context that applies to all evaluations:

from litestar_flags import FeatureFlagsConfig, EvaluationContext

config = FeatureFlagsConfig(
    backend="memory",
    default_context=EvaluationContext(
        environment="production",
        app_version="2.1.0",
    ),
)

Dependency Key¶

Customize the dependency injection key for the client:

config = FeatureFlagsConfig(
    backend="memory",
    client_dependency_key="flags",  # Default is "feature_flags"
)

# In your route handler:
@get("/")
async def handler(flags: FeatureFlagClient) -> dict:
    # Use 'flags' instead of 'feature_flags'
    pass

Plugin Reference¶

Litestar plugin for feature flags.

class litestar_flags.plugin.FeatureFlagsPlugin[source]¶

Bases: InitPlugin

Litestar plugin for feature flags.

Registers the feature flag client as a dependency and sets up lifecycle hooks for initialization and cleanup.

Example

>>> from litestar import Litestar
>>> from litestar_flags import FeatureFlagsPlugin, FeatureFlagsConfig
>>>
>>> config = FeatureFlagsConfig(backend="memory")
>>> app = Litestar(
...     route_handlers=[...],
...     plugins=[FeatureFlagsPlugin(config=config)],
... )

Parameters:: config (FeatureFlagsConfig | None)

__init__(config=None)[source]¶

Initialize the plugin.

Parameters:: config (FeatureFlagsConfig | None) – Plugin configuration. Defaults to memory backend.
Return type:: None

property config: FeatureFlagsConfig¶: Get the plugin configuration.

property client: FeatureFlagClient | None¶: Get the feature flag client (available after startup).

on_app_init(app_config)[source]¶

Configure the application with feature flags support.

Parameters:: app_config (AppConfig) – The application configuration.
Return type:: AppConfig
Returns:: Modified application configuration.

Plugin Usage¶

Basic Setup¶

from litestar import Litestar, get
from litestar_flags import (
    FeatureFlagsPlugin,
    FeatureFlagsConfig,
    FeatureFlagClient,
)

config = FeatureFlagsConfig(backend="memory")
plugin = FeatureFlagsPlugin(config=config)

@get("/")
async def handler(feature_flags: FeatureFlagClient) -> dict:
    enabled = await feature_flags.get_boolean_value("my-feature")
    return {"enabled": enabled}

app = Litestar(
    route_handlers=[handler],
    plugins=[plugin],
)

Accessing the Client¶

The client is available in three ways:

Dependency Injection (recommended):

@get("/")
async def handler(feature_flags: FeatureFlagClient) -> dict:
    enabled = await feature_flags.is_enabled("my-feature")
    return {"enabled": enabled}

From Application State:

@get("/")
async def handler(request: Request) -> dict:
    client = request.app.state.feature_flags
    enabled = await client.is_enabled("my-feature")
    return {"enabled": enabled}

From Plugin Instance:

plugin = FeatureFlagsPlugin(config=config)
# After app startup:
client = plugin.client