Storage Backends¶

Storage backends provide persistence for feature flags. The library includes three built-in backends with different characteristics and use cases.

Overview¶

All storage backends implement the StorageBackend protocol, ensuring a consistent interface regardless of the underlying storage mechanism.

Available Backends:

Backend	Use Case	Requirements
`MemoryStorageBackend`	Development, testing, single-instance	None (built-in)
`DatabaseStorageBackend`	Production with SQL database	`litestar-flags[database]`
`RedisStorageBackend`	Distributed deployments	`litestar-flags[redis]`

StorageBackend Protocol¶

All backends implement this protocol:

Protocols and interfaces for feature flag storage backends.

class litestar_flags.protocols.StorageBackend[source]¶

Bases: Protocol

Protocol for feature flag storage backends.

All storage backend implementations must implement this protocol. Methods are async to support both sync and async backends.

Implementations:

MemoryStorageBackend: In-memory storage for development/testing
DatabaseStorageBackend: SQLAlchemy-based persistent storage
RedisStorageBackend: Redis-based distributed storage

async get_flag(key)[source]¶

Retrieve a single flag by key.

Parameters:: key (str) – The unique flag key.
Return type:: FeatureFlag | None
Returns:: The FeatureFlag if found, None otherwise.

async get_flags(keys)[source]¶

Retrieve multiple flags by keys.

Parameters:: keys (Sequence[str]) – Sequence of flag keys to retrieve.
Return type:: dict[str, FeatureFlag]
Returns:: Dictionary mapping flag keys to FeatureFlag objects.

async get_all_active_flags()[source]¶

Retrieve all active flags.

Return type:: list[FeatureFlag]
Returns:: List of all FeatureFlag objects with ACTIVE status.

async get_override(flag_id, entity_type, entity_id)[source]¶

Retrieve entity-specific override.

Parameters:

flag_id (UUID) – The flag’s UUID.
entity_type (str) – Type of entity (e.g., “user”, “organization”).
entity_id (str) – The entity’s identifier.

Return type:

FlagOverride | None

Returns:

The FlagOverride if found, None otherwise.

async create_flag(flag)[source]¶

Create a new flag.

Parameters:: flag (FeatureFlag) – The flag to create.
Return type:: FeatureFlag
Returns:: The created flag with any generated fields populated.

async update_flag(flag)[source]¶

Update an existing flag.

Parameters:: flag (FeatureFlag) – The flag with updated values.
Return type:: FeatureFlag
Returns:: The updated flag.

async delete_flag(key)[source]¶

Delete a flag by key.

Parameters:: key (str) – The unique flag key.
Return type:: bool
Returns:: True if the flag was deleted, False if not found.

async health_check()[source]¶

Check storage backend health.

Return type:: bool
Returns:: True if the backend is healthy, False otherwise.

async close()[source]¶

Close any open connections and clean up resources.

Return type:: None

async get_scheduled_changes(flag_id=None, pending_only=True)[source]¶

Get scheduled changes, optionally filtered by flag and status.

Parameters:

flag_id (UUID | None) – If provided, filter to changes for this flag only.
pending_only (bool) – If True, only return changes not yet executed.

Return type:

list[ScheduledFlagChange]

Returns:

List of scheduled changes matching the criteria.

async create_scheduled_change(change)[source]¶

Create a new scheduled change.

Parameters:: change (ScheduledFlagChange) – The scheduled change to create.
Return type:: ScheduledFlagChange
Returns:: The created scheduled change with any generated fields populated.

async update_scheduled_change(change)[source]¶

Update a scheduled change (e.g., mark as executed).

Parameters:: change (ScheduledFlagChange) – The scheduled change with updated values.
Return type:: ScheduledFlagChange
Returns:: The updated scheduled change.

async get_time_schedules(flag_id=None)[source]¶

Get time schedules for a flag or all flags.

Parameters:: flag_id (UUID | None) – If provided, filter to schedules for this flag only.
Return type:: list[TimeSchedule]
Returns:: List of time schedules matching the criteria.

async create_time_schedule(schedule)[source]¶

Create a new time schedule.

Parameters:: schedule (TimeSchedule) – The time schedule to create.
Return type:: TimeSchedule
Returns:: The created time schedule with any generated fields populated.

async delete_time_schedule(schedule_id)[source]¶

Delete a time schedule.

Parameters:: schedule_id (UUID) – The UUID of the time schedule to delete.
Return type:: bool
Returns:: True if the schedule was deleted, False if not found.

async get_rollout_phases(flag_id)[source]¶

Get rollout phases for a flag.

Parameters:: flag_id (UUID) – The UUID of the flag.
Return type:: list[RolloutPhase]
Returns:: List of rollout phases for the flag, ordered by phase number.

async create_rollout_phase(phase)[source]¶

Create a new rollout phase.

Parameters:: phase (RolloutPhase) – The rollout phase to create.
Return type:: RolloutPhase
Returns:: The created rollout phase with any generated fields populated.

async get_segment(segment_id)[source]¶

Retrieve a segment by ID.

Parameters:: segment_id (UUID) – The UUID of the segment.
Return type:: Segment | None
Returns:: The Segment if found, None otherwise.

async get_segment_by_name(name)[source]¶

Retrieve a segment by name.

Parameters:: name (str) – The unique segment name.
Return type:: Segment | None
Returns:: The Segment if found, None otherwise.

async get_all_segments()[source]¶

Retrieve all segments.

Return type:: list[Segment]
Returns:: List of all Segment objects.

async get_child_segments(parent_id)[source]¶

Retrieve all child segments of a parent segment.

Parameters:: parent_id (UUID) – The UUID of the parent segment.
Return type:: list[Segment]
Returns:: List of child Segment objects.

async create_segment(segment)[source]¶

Create a new segment.

Parameters:: segment (Segment) – The segment to create.
Return type:: Segment
Returns:: The created segment with any generated fields populated.

async update_segment(segment)[source]¶

Update an existing segment.

Parameters:: segment (Segment) – The segment with updated values.
Return type:: Segment
Returns:: The updated segment.

async delete_segment(segment_id)[source]¶

Delete a segment by ID.

Parameters:: segment_id (UUID) – The UUID of the segment to delete.
Return type:: bool
Returns:: True if the segment was deleted, False if not found.

async get_environment(slug)[source]¶

Retrieve an environment by slug.

Parameters:: slug (str) – The unique URL-safe identifier for the environment.
Return type:: Environment | None
Returns:: The Environment if found, None otherwise.

async get_environment_by_id(env_id)[source]¶

Retrieve an environment by ID.

Parameters:: env_id (UUID) – The UUID of the environment.
Return type:: Environment | None
Returns:: The Environment if found, None otherwise.

async get_all_environments()[source]¶

Retrieve all environments.

Return type:: list[Environment]
Returns:: List of all Environment objects.

async get_child_environments(parent_id)[source]¶

Retrieve all child environments of a parent environment.

Parameters:: parent_id (UUID) – The UUID of the parent environment.
Return type:: list[Environment]
Returns:: List of child Environment objects.

async create_environment(env)[source]¶

Create a new environment.

Parameters:: env (Environment) – The environment to create.
Return type:: Environment
Returns:: The created environment with any generated fields populated.

async update_environment(env)[source]¶

Update an existing environment.

Parameters:: env (Environment) – The environment with updated values.
Return type:: Environment
Returns:: The updated environment.

async delete_environment(slug)[source]¶

Delete an environment by slug.

Parameters:: slug (str) – The unique URL-safe identifier of the environment to delete.
Return type:: bool
Returns:: True if the environment was deleted, False if not found.

async get_environment_flag(env_id, flag_id)[source]¶

Retrieve environment-specific flag configuration.

Parameters:

env_id (UUID) – The UUID of the environment.
flag_id (UUID) – The UUID of the feature flag.

Return type:

EnvironmentFlag | None

Returns:

The EnvironmentFlag if found, None otherwise.

async get_environment_flags(env_id)[source]¶

Retrieve all flag configurations for an environment.

Parameters:: env_id (UUID) – The UUID of the environment.
Return type:: list[EnvironmentFlag]
Returns:: List of EnvironmentFlag objects for the specified environment.

async get_flag_environments(flag_id)[source]¶

Retrieve all environment configurations for a flag.

Parameters:: flag_id (UUID) – The UUID of the feature flag.
Return type:: list[EnvironmentFlag]
Returns:: List of EnvironmentFlag objects for the specified flag.

async create_environment_flag(env_flag)[source]¶

Create a new environment-specific flag configuration.

Parameters:: env_flag (EnvironmentFlag) – The environment flag configuration to create.
Return type:: EnvironmentFlag
Returns:: The created EnvironmentFlag with any generated fields populated.

async update_environment_flag(env_flag)[source]¶

Update an existing environment-specific flag configuration.

Parameters:: env_flag (EnvironmentFlag) – The environment flag configuration with updated values.
Return type:: EnvironmentFlag
Returns:: The updated EnvironmentFlag.

async delete_environment_flag(env_id, flag_id)[source]¶

Delete an environment-specific flag configuration.

Parameters:

env_id (UUID) – The UUID of the environment.
flag_id (UUID) – The UUID of the feature flag.

Return type:

bool

Returns:

True if the configuration was deleted, False if not found.

__init__(*args, **kwargs)¶

MemoryStorageBackend¶

In-memory storage for development and testing. Data is not persisted between application restarts.

In-memory storage backend for feature flags.

class litestar_flags.storage.memory.MemoryStorageBackend[source]¶

Bases: object

In-memory storage backend for development and testing.

This backend stores all data in memory and is not persistent. Ideal for development, testing, and simple single-instance deployments.

Example

>>> storage = MemoryStorageBackend()
>>> await storage.create_flag(flag)
>>> flag = await storage.get_flag("my-feature")

__init__()[source]¶

Initialize the in-memory storage.

Return type:: None

async get_flag(key)[source]¶

Retrieve a single flag by key.

Parameters:: key (str) – The unique flag key.
Return type:: FeatureFlag | None
Returns:: The FeatureFlag if found, None otherwise.

async get_flags(keys)[source]¶

Retrieve multiple flags by keys.

Parameters:: keys (Sequence[str]) – Sequence of flag keys to retrieve.
Return type:: dict[str, FeatureFlag]
Returns:: Dictionary mapping flag keys to FeatureFlag objects.

async get_all_active_flags()[source]¶

Retrieve all active flags.

Return type:: list[FeatureFlag]
Returns:: List of all FeatureFlag objects with ACTIVE status.

async get_override(flag_id, entity_type, entity_id)[source]¶

Retrieve entity-specific override.

Parameters:

flag_id (UUID) – The flag’s UUID.
entity_type (str) – Type of entity (e.g., “user”, “organization”).
entity_id (str) – The entity’s identifier.

Return type:

FlagOverride | None

Returns:

The FlagOverride if found and not expired, None otherwise.

async get_overrides_for_entity(entity_type, entity_id)[source]¶

Retrieve all overrides for an entity.

Parameters:

entity_type (str) – Type of entity (e.g., “user”, “organization”).
entity_id (str) – The entity’s identifier.

Return type:

list[FlagOverride]

Returns:

List of non-expired overrides for the entity.

async create_flag(flag)[source]¶

Create a new flag.

Parameters:: flag (FeatureFlag) – The flag to create.
Return type:: FeatureFlag
Returns:: The created flag.
Raises:: ValueError – If a flag with the same key already exists.

async update_flag(flag)[source]¶

Update an existing flag.

Parameters:: flag (FeatureFlag) – The flag with updated values.
Return type:: FeatureFlag
Returns:: The updated flag.
Raises:: ValueError – If the flag does not exist.

async delete_flag(key)[source]¶

Delete a flag by key.

Parameters:: key (str) – The unique flag key.
Return type:: bool
Returns:: True if the flag was deleted, False if not found.

async create_override(override)[source]¶

Create a new override.

Parameters:: override (FlagOverride) – The override to create.
Return type:: FlagOverride
Returns:: The created override.

async delete_override(flag_id, entity_type, entity_id)[source]¶

Delete an override.

Parameters:

flag_id (UUID) – The flag’s UUID.
entity_type (str) – Type of entity.
entity_id (str) – The entity’s identifier.

Return type:

bool

Returns:

True if the override was deleted, False if not found.

async health_check()[source]¶

Check storage backend health.

Return type:: bool
Returns:: Always True for in-memory storage.

async close()[source]¶

Close the storage backend.

Clears all data from memory.

Return type:: None

async get_scheduled_changes(flag_id=None, pending_only=True)[source]¶

Get scheduled changes, optionally filtered by flag and status.

Parameters:

flag_id (UUID | None) – If provided, filter to changes for this flag only.
pending_only (bool) – If True, only return changes not yet executed.

Return type:

list[ScheduledFlagChange]

Returns:

List of scheduled changes matching the criteria.

async create_scheduled_change(change)[source]¶

Create a new scheduled change.

Parameters:: change (ScheduledFlagChange) – The scheduled change to create.
Return type:: ScheduledFlagChange
Returns:: The created scheduled change.

async update_scheduled_change(change)[source]¶

Update a scheduled change (e.g., mark as executed).

Parameters:: change (ScheduledFlagChange) – The scheduled change with updated values.
Return type:: ScheduledFlagChange
Returns:: The updated scheduled change.
Raises:: ValueError – If the scheduled change does not exist.

async get_time_schedules(flag_id=None)[source]¶

Get time schedules for a flag or all flags.

Parameters:: flag_id (UUID | None) – If provided, filter to schedules for this flag only.
Return type:: list[TimeSchedule]
Returns:: List of time schedules matching the criteria.

async create_time_schedule(schedule)[source]¶

Create a new time schedule.

Parameters:: schedule (TimeSchedule) – The time schedule to create.
Return type:: TimeSchedule
Returns:: The created time schedule.

async delete_time_schedule(schedule_id)[source]¶

Delete a time schedule.

Parameters:: schedule_id (UUID) – The UUID of the time schedule to delete.
Return type:: bool
Returns:: True if the schedule was deleted, False if not found.

async get_rollout_phases(flag_id)[source]¶

Get rollout phases for a flag.

Parameters:: flag_id (UUID) – The UUID of the flag.
Return type:: list[RolloutPhase]
Returns:: List of rollout phases for the flag, ordered by phase number.

async create_rollout_phase(phase)[source]¶

Create a new rollout phase.

Parameters:: phase (RolloutPhase) – The rollout phase to create.
Return type:: RolloutPhase
Returns:: The created rollout phase.

async get_segment(segment_id)[source]¶

Retrieve a segment by ID.

Parameters:: segment_id (UUID) – The UUID of the segment.
Return type:: Segment | None
Returns:: The Segment if found, None otherwise.

async get_segment_by_name(name)[source]¶

Retrieve a segment by name.

Parameters:: name (str) – The unique segment name.
Return type:: Segment | None
Returns:: The Segment if found, None otherwise.

async get_all_segments()[source]¶

Retrieve all segments.

Return type:: list[Segment]
Returns:: List of all Segment objects.

async get_child_segments(parent_id)[source]¶

Retrieve all child segments of a parent segment.

Parameters:: parent_id (UUID) – The UUID of the parent segment.
Return type:: list[Segment]
Returns:: List of child Segment objects.

async create_segment(segment)[source]¶

Create a new segment.

Parameters:: segment (Segment) – The segment to create.
Return type:: Segment
Returns:: The created segment.
Raises:: ValueError – If a segment with the same name already exists.

async update_segment(segment)[source]¶

Update an existing segment.

Parameters:: segment (Segment) – The segment with updated values.
Return type:: Segment
Returns:: The updated segment.
Raises:: ValueError – If the segment does not exist or name conflict occurs.

async delete_segment(segment_id)[source]¶

Delete a segment by ID.

Parameters:: segment_id (UUID) – The UUID of the segment to delete.
Return type:: bool
Returns:: True if the segment was deleted, False if not found.

async get_environment(slug)[source]¶

Retrieve an environment by slug.

Parameters:: slug (str) – The unique environment slug.
Return type:: Environment | None
Returns:: The Environment if found, None otherwise.

async get_environment_by_id(env_id)[source]¶

Retrieve an environment by ID.

Parameters:: env_id (UUID) – The UUID of the environment.
Return type:: Environment | None
Returns:: The Environment if found, None otherwise.

async get_all_environments()[source]¶

Retrieve all environments.

Return type:: list[Environment]
Returns:: List of all Environment objects.

async get_child_environments(parent_id)[source]¶

Retrieve all child environments of a parent environment.

Parameters:: parent_id (UUID) – The UUID of the parent environment.
Return type:: list[Environment]
Returns:: List of child Environment objects.

async create_environment(env)[source]¶

Create a new environment.

Parameters:: env (Environment) – The environment to create.
Return type:: Environment
Returns:: The created environment.
Raises:: ValueError – If an environment with the same slug already exists.

async update_environment(env)[source]¶

Update an existing environment.

Parameters:: env (Environment) – The environment with updated values.
Return type:: Environment
Returns:: The updated environment.
Raises:: ValueError – If the environment does not exist or slug conflict occurs.

async delete_environment(slug)[source]¶

Delete an environment by slug.

Also deletes all related environment flags.

Parameters:: slug (str) – The unique environment slug.
Return type:: bool
Returns:: True if the environment was deleted, False if not found.

async get_environment_flag(env_id, flag_id)[source]¶

Retrieve an environment-specific flag configuration.

Parameters:

env_id (UUID) – The environment’s UUID.
flag_id (UUID) – The flag’s UUID.

Return type:

EnvironmentFlag | None

Returns:

The EnvironmentFlag if found, None otherwise.

async get_environment_flags(env_id)[source]¶

Retrieve all flag configurations for an environment.

Parameters:: env_id (UUID) – The environment’s UUID.
Return type:: list[EnvironmentFlag]
Returns:: List of EnvironmentFlag objects for the environment.

async get_flag_environments(flag_id)[source]¶

Retrieve all environment configurations for a flag.

Parameters:: flag_id (UUID) – The flag’s UUID.
Return type:: list[EnvironmentFlag]
Returns:: List of EnvironmentFlag objects for the flag.

async create_environment_flag(env_flag)[source]¶

Create a new environment flag configuration.

Parameters:: env_flag (EnvironmentFlag) – The environment flag configuration to create.
Return type:: EnvironmentFlag
Returns:: The created environment flag.
Raises:: ValueError – If the environment flag already exists.

async update_environment_flag(env_flag)[source]¶

Update an existing environment flag configuration.

Parameters:: env_flag (EnvironmentFlag) – The environment flag with updated values.
Return type:: EnvironmentFlag
Returns:: The updated environment flag.
Raises:: ValueError – If the environment flag does not exist.

async delete_environment_flag(env_id, flag_id)[source]¶

Delete an environment flag configuration.

Parameters:

env_id (UUID) – The environment’s UUID.
flag_id (UUID) – The flag’s UUID.

Return type:

bool

Returns:

True if the environment flag was deleted, False if not found.

Usage¶

from litestar_flags import MemoryStorageBackend, FeatureFlagClient

# Create storage
storage = MemoryStorageBackend()

# Create client
client = FeatureFlagClient(storage=storage)

# Or via plugin configuration
from litestar_flags import FeatureFlagsConfig

config = FeatureFlagsConfig(backend="memory")

When to Use¶

Local development
Unit and integration testing
Simple single-instance deployments
Prototyping and experimentation

DatabaseStorageBackend¶

SQLAlchemy-based persistent storage using Advanced-Alchemy. Supports PostgreSQL, MySQL, SQLite, and other SQLAlchemy-compatible databases.

Database storage backend using Advanced-Alchemy.

class litestar_flags.storage.database.DatabaseStorageBackend[source]¶

Bases: object

Database storage backend using Advanced-Alchemy.

This backend stores feature flags and related data in a relational database using SQLAlchemy async operations.

Example

>>> storage = await DatabaseStorageBackend.create(
...     connection_string="postgresql+asyncpg://user:pass@localhost/db"
... )
>>> flag = await storage.get_flag("my-feature")

Parameters:

engine (AsyncEngine)
session_maker (async_sessionmaker[AsyncSession])

__init__(engine, session_maker)[source]¶

Initialize the database storage backend.

Parameters:

engine (AsyncEngine) – The SQLAlchemy async engine.
session_maker (async_sessionmaker[AsyncSession]) – The session maker factory.

Return type:

None

async classmethod create(connection_string, table_prefix='ff_', create_tables=True, **engine_kwargs)[source]¶

Create a new database storage backend.

Parameters:

connection_string (str) – Database connection string.
table_prefix (str) – Prefix for table names (not currently used).
create_tables (bool) – Whether to create tables on startup.
**engine_kwargs (dict) – Additional arguments for create_async_engine.

Return type:

DatabaseStorageBackend

Returns:

Configured DatabaseStorageBackend instance.

async get_flag(key)[source]¶

Retrieve a single flag by key.

Parameters:: key (str) – The unique flag key.
Return type:: FeatureFlag | None
Returns:: The FeatureFlag if found, None otherwise.

async get_flags(keys)[source]¶

Retrieve multiple flags by keys.

Parameters:: keys (Sequence[str]) – Sequence of flag keys to retrieve.
Return type:: dict[str, FeatureFlag]
Returns:: Dictionary mapping flag keys to FeatureFlag objects.

async get_all_active_flags()[source]¶

Retrieve all active flags.

Return type:: list[FeatureFlag]
Returns:: List of all FeatureFlag objects with ACTIVE status.

async get_override(flag_id, entity_type, entity_id)[source]¶

Retrieve entity-specific override.

Parameters:

flag_id (UUID) – The flag’s UUID.
entity_type (str) – Type of entity (e.g., “user”, “organization”).
entity_id (str) – The entity’s identifier.

Return type:

FlagOverride | None

Returns:

The FlagOverride if found, None otherwise.

async create_flag(flag)[source]¶

Create a new flag.

Parameters:: flag (FeatureFlag) – The flag to create.
Return type:: FeatureFlag
Returns:: The created flag with any generated fields populated.

async update_flag(flag)[source]¶

Update an existing flag.

Parameters:: flag (FeatureFlag) – The flag with updated values.
Return type:: FeatureFlag
Returns:: The updated flag.

async delete_flag(key)[source]¶

Delete a flag by key.

Parameters:: key (str) – The unique flag key.
Return type:: bool
Returns:: True if the flag was deleted, False if not found.

async create_override(override)[source]¶

Create a new override.

Parameters:: override (FlagOverride) – The override to create.
Return type:: FlagOverride
Returns:: The created override.

async delete_override(flag_id, entity_type, entity_id)[source]¶

Delete an override.

Parameters:

flag_id (UUID) – The flag’s UUID.
entity_type (str) – Type of entity.
entity_id (str) – The entity’s identifier.

Return type:

bool

Returns:

True if the override was deleted, False if not found.

async health_check()[source]¶

Check storage backend health.

Return type:: bool
Returns:: True if the backend is healthy, False otherwise.

async close()[source]¶

Close database connections.

Return type:: None

async get_scheduled_changes(flag_id=None, pending_only=True)[source]¶

Get scheduled changes, optionally filtered by flag and status.

Parameters:

flag_id (UUID | None) – If provided, filter to changes for this flag only.
pending_only (bool) – If True, only return changes not yet executed.

Return type:

list[ScheduledFlagChange]

Returns:

List of scheduled changes matching the criteria.

async create_scheduled_change(change)[source]¶

Create a new scheduled change.

Parameters:: change (ScheduledFlagChange) – The scheduled change to create.
Return type:: ScheduledFlagChange
Returns:: The created scheduled change with any generated fields populated.

async update_scheduled_change(change)[source]¶

Update a scheduled change (e.g., mark as executed).

Parameters:: change (ScheduledFlagChange) – The scheduled change with updated values.
Return type:: ScheduledFlagChange
Returns:: The updated scheduled change.

async get_time_schedules(flag_id=None)[source]¶

Get time schedules for a flag or all flags.

Parameters:: flag_id (UUID | None) – If provided, filter to schedules for this flag only.
Return type:: list[TimeSchedule]
Returns:: List of time schedules matching the criteria.

async create_time_schedule(schedule)[source]¶

Create a new time schedule.

Parameters:: schedule (TimeSchedule) – The time schedule to create.
Return type:: TimeSchedule
Returns:: The created time schedule with any generated fields populated.

async delete_time_schedule(schedule_id)[source]¶

Delete a time schedule.

Parameters:: schedule_id (UUID) – The UUID of the time schedule to delete.
Return type:: bool
Returns:: True if the schedule was deleted, False if not found.

async get_rollout_phases(flag_id)[source]¶

Get rollout phases for a flag.

Parameters:: flag_id (UUID) – The UUID of the flag.
Return type:: list[RolloutPhase]
Returns:: List of rollout phases for the flag, ordered by phase number.

async create_rollout_phase(phase)[source]¶

Create a new rollout phase.

Parameters:: phase (RolloutPhase) – The rollout phase to create.
Return type:: RolloutPhase
Returns:: The created rollout phase with any generated fields populated.

class litestar_flags.storage.database.FeatureFlagRepository[source]¶

Bases: SQLAlchemyAsyncRepository[FeatureFlag]

Repository for feature flag CRUD operations.

Parameters:

statement (Optional[Select])
session (Union[AsyncSession, async_scoped_session[AsyncSession]])
auto_expunge (bool)
auto_refresh (bool)
auto_commit (bool)
order_by (Union[list[Union[tuple[Union[str, InstrumentedAttribute[Any]], bool], UnaryExpression[Any]]], tuple[Union[str, InstrumentedAttribute[Any]], bool], UnaryExpression[Any], None])
error_messages (Union[ErrorMessages, type[Empty], None])
load (Union[Sequence[Union[_AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any], Sequence[Union[_AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any]]]]], _AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any], ExecutableOption, Sequence[ExecutableOption], None])
execution_options (Optional[dict[str, Any]])
wrap_exceptions (bool)
uniquify (Optional[bool])
count_with_window_function (Optional[bool])
kwargs (Any)

model_type¶: alias of FeatureFlag

async get_by_key(key)[source]¶

Get a flag by its unique key.

Parameters:: key (str) – The flag key.
Return type:: FeatureFlag | None
Returns:: The flag if found, None otherwise.

async get_by_keys(keys)[source]¶

Get multiple flags by their keys.

Parameters:: keys (Sequence[str]) – The flag keys.
Return type:: list[FeatureFlag]
Returns:: List of found flags.

async get_active_flags()[source]¶

Get all active flags.

Return type:: list[FeatureFlag]
Returns:: List of active flags.

class litestar_flags.storage.database.FlagOverrideRepository[source]¶

Bases: SQLAlchemyAsyncRepository[FlagOverride]

Repository for flag override CRUD operations.

Parameters:

statement (Optional[Select])
session (Union[AsyncSession, async_scoped_session[AsyncSession]])
auto_expunge (bool)
auto_refresh (bool)
auto_commit (bool)
order_by (Union[list[Union[tuple[Union[str, InstrumentedAttribute[Any]], bool], UnaryExpression[Any]]], tuple[Union[str, InstrumentedAttribute[Any]], bool], UnaryExpression[Any], None])
error_messages (Union[ErrorMessages, type[Empty], None])
load (Union[Sequence[Union[_AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any], Sequence[Union[_AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any]]]]], _AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any], ExecutableOption, Sequence[ExecutableOption], None])
execution_options (Optional[dict[str, Any]])
wrap_exceptions (bool)
uniquify (Optional[bool])
count_with_window_function (Optional[bool])
kwargs (Any)

model_type¶: alias of FlagOverride

async get_override(flag_id, entity_type, entity_id)[source]¶

Get an override for a specific entity.

Parameters:

flag_id (UUID) – The flag’s UUID.
entity_type (str) – Type of entity.
entity_id (str) – The entity’s identifier.

Return type:

FlagOverride | None

Returns:

The override if found, None otherwise.

class litestar_flags.storage.database.RolloutPhaseRepository[source]¶

Bases: SQLAlchemyAsyncRepository[RolloutPhase]

Repository for rollout phase CRUD operations.

Parameters:

statement (Optional[Select])
session (Union[AsyncSession, async_scoped_session[AsyncSession]])
auto_expunge (bool)
auto_refresh (bool)
auto_commit (bool)
order_by (Union[list[Union[tuple[Union[str, InstrumentedAttribute[Any]], bool], UnaryExpression[Any]]], tuple[Union[str, InstrumentedAttribute[Any]], bool], UnaryExpression[Any], None])
error_messages (Union[ErrorMessages, type[Empty], None])
load (Union[Sequence[Union[_AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any], Sequence[Union[_AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any]]]]], _AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any], ExecutableOption, Sequence[ExecutableOption], None])
execution_options (Optional[dict[str, Any]])
wrap_exceptions (bool)
uniquify (Optional[bool])
count_with_window_function (Optional[bool])
kwargs (Any)

model_type¶: alias of RolloutPhase

async get_phases_for_flag(flag_id)[source]¶

Get all rollout phases for a flag.

Parameters:: flag_id (UUID) – The flag’s UUID.
Return type:: list[RolloutPhase]
Returns:: List of rollout phases, ordered by phase_number.

async get_pending_phases(flag_id)[source]¶

Get pending (not yet executed) rollout phases for a flag.

Parameters:: flag_id (UUID) – The flag’s UUID.
Return type:: list[RolloutPhase]
Returns:: List of pending rollout phases, ordered by phase_number.

class litestar_flags.storage.database.ScheduledFlagChangeRepository[source]¶

Bases: SQLAlchemyAsyncRepository[ScheduledFlagChange]

Repository for scheduled flag change CRUD operations.

Parameters:

statement (Optional[Select])
session (Union[AsyncSession, async_scoped_session[AsyncSession]])
auto_expunge (bool)
auto_refresh (bool)
auto_commit (bool)
order_by (Union[list[Union[tuple[Union[str, InstrumentedAttribute[Any]], bool], UnaryExpression[Any]]], tuple[Union[str, InstrumentedAttribute[Any]], bool], UnaryExpression[Any], None])
error_messages (Union[ErrorMessages, type[Empty], None])
load (Union[Sequence[Union[_AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any], Sequence[Union[_AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any]]]]], _AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any], ExecutableOption, Sequence[ExecutableOption], None])
execution_options (Optional[dict[str, Any]])
wrap_exceptions (bool)
uniquify (Optional[bool])
count_with_window_function (Optional[bool])
kwargs (Any)

model_type¶: alias of ScheduledFlagChange

async get_pending_changes(flag_id=None)[source]¶

Get pending (not yet executed) scheduled changes.

Parameters:: flag_id (UUID | None) – If provided, filter to changes for this flag only.
Return type:: list[ScheduledFlagChange]
Returns:: List of pending scheduled changes, ordered by scheduled_at.

async get_all_changes(flag_id=None)[source]¶

Get all scheduled changes (pending and executed).

Parameters:: flag_id (UUID | None) – If provided, filter to changes for this flag only.
Return type:: list[ScheduledFlagChange]
Returns:: List of scheduled changes, ordered by scheduled_at.

class litestar_flags.storage.database.TimeScheduleRepository[source]¶

Bases: SQLAlchemyAsyncRepository[TimeSchedule]

Repository for time schedule CRUD operations.

Parameters:

statement (Optional[Select])
session (Union[AsyncSession, async_scoped_session[AsyncSession]])
auto_expunge (bool)
auto_refresh (bool)
auto_commit (bool)
order_by (Union[list[Union[tuple[Union[str, InstrumentedAttribute[Any]], bool], UnaryExpression[Any]]], tuple[Union[str, InstrumentedAttribute[Any]], bool], UnaryExpression[Any], None])
error_messages (Union[ErrorMessages, type[Empty], None])
load (Union[Sequence[Union[_AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any], Sequence[Union[_AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any]]]]], _AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any], ExecutableOption, Sequence[ExecutableOption], None])
execution_options (Optional[dict[str, Any]])
wrap_exceptions (bool)
uniquify (Optional[bool])
count_with_window_function (Optional[bool])
kwargs (Any)

model_type¶: alias of TimeSchedule

async get_schedules_for_flag(flag_id)[source]¶

Get all time schedules for a flag.

Parameters:: flag_id (UUID) – The flag’s UUID.
Return type:: list[TimeSchedule]
Returns:: List of time schedules for the flag.

async get_enabled_schedules(flag_id=None)[source]¶

Get all enabled time schedules.

Parameters:: flag_id (UUID | None) – If provided, filter to schedules for this flag only.
Return type:: list[TimeSchedule]
Returns:: List of enabled time schedules.

Installation¶

uv (Recommended)

uv add litestar-flags[database]

pip

pip install litestar-flags[database]

Usage¶

from litestar_flags import FeatureFlagsConfig

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

# MySQL
config = FeatureFlagsConfig(
    backend="database",
    connection_string="mysql+aiomysql://user:pass@localhost/mydb",
)

# SQLite (for development)
config = FeatureFlagsConfig(
    backend="database",
    connection_string="sqlite+aiosqlite:///flags.db",
)

Direct Usage¶

For advanced use cases, you can create the backend directly:

from litestar_flags.storage.database import DatabaseStorageBackend

storage = await DatabaseStorageBackend.create(
    connection_string="postgresql+asyncpg://user:pass@localhost/mydb",
    table_prefix="ff_",
    create_tables=True,  # Auto-create tables on startup
)

# Use with client
client = FeatureFlagClient(storage=storage)

# Don't forget to close when done
await storage.close()

When to Use¶

Production deployments requiring persistence
Multi-instance deployments with shared database
When you need transactional consistency
Integration with existing database infrastructure

RedisStorageBackend¶

Redis-based distributed storage for high-performance, distributed deployments.

Redis storage backend for feature flags.

class litestar_flags.storage.redis.RedisStorageBackend[source]¶

Bases: object

Redis storage backend for distributed feature flags.

This backend stores feature flags in Redis, making it suitable for distributed deployments where multiple application instances need to share flag state.

Data is stored as JSON strings with the following key patterns: - Flags: {prefix}flag:{key} - Overrides: {prefix}override:{flag_id}:{entity_type}:{entity_id} - Flag index: {prefix}flags (SET of all flag keys)

Example

>>> storage = await RedisStorageBackend.create(
...     url="redis://localhost:6379",
...     prefix="ff:"
... )
>>> flag = await storage.get_flag("my-feature")

Parameters:

redis (Redis)
prefix (str)

__init__(redis, prefix='feature_flags:')[source]¶

Initialize the Redis storage backend.

Parameters:

redis (Redis) – The Redis client instance.
prefix (str) – Key prefix for all stored data.

Return type:

None

async classmethod create(url, prefix='feature_flags:', **redis_kwargs)[source]¶

Create a new Redis storage backend.

Parameters:

url (str) – Redis connection URL.
prefix (str) – Key prefix for all stored data.
**redis_kwargs (Any) – Additional arguments for Redis.from_url().

Return type:

RedisStorageBackend

Returns:

Configured RedisStorageBackend instance.

async get_flag(key)[source]¶

Retrieve a single flag by key.

Parameters:: key (str) – The unique flag key.
Return type:: FeatureFlag | None
Returns:: The FeatureFlag if found, None otherwise.

async get_flags(keys)[source]¶

Retrieve multiple flags by keys.

Parameters:: keys (Sequence[str]) – Sequence of flag keys to retrieve.
Return type:: dict[str, FeatureFlag]
Returns:: Dictionary mapping flag keys to FeatureFlag objects.

async get_all_active_flags()[source]¶

Retrieve all active flags.

Return type:: list[FeatureFlag]
Returns:: List of all FeatureFlag objects with ACTIVE status.

async get_override(flag_id, entity_type, entity_id)[source]¶

Retrieve entity-specific override.

Parameters:

flag_id (UUID) – The flag’s UUID.
entity_type (str) – Type of entity (e.g., “user”, “organization”).
entity_id (str) – The entity’s identifier.

Return type:

FlagOverride | None

Returns:

The FlagOverride if found and not expired, None otherwise.

async create_flag(flag)[source]¶

Create a new flag.

Parameters:: flag (FeatureFlag) – The flag to create.
Return type:: FeatureFlag
Returns:: The created flag.

async update_flag(flag)[source]¶

Update an existing flag.

Parameters:: flag (FeatureFlag) – The flag with updated values.
Return type:: FeatureFlag
Returns:: The updated flag.

async delete_flag(key)[source]¶

Delete a flag by key.

Parameters:: key (str) – The unique flag key.
Return type:: bool
Returns:: True if the flag was deleted, False if not found.

async create_override(override)[source]¶

Create a new override.

Parameters:: override (FlagOverride) – The override to create.
Return type:: FlagOverride
Returns:: The created override.

async delete_override(flag_id, entity_type, entity_id)[source]¶

Delete an override.

Parameters:

flag_id (UUID) – The flag’s UUID.
entity_type (str) – Type of entity.
entity_id (str) – The entity’s identifier.

Return type:

bool

Returns:

True if the override was deleted, False if not found.

async health_check()[source]¶

Check storage backend health.

Return type:: bool
Returns:: True if the backend is healthy, False otherwise.

async close()[source]¶

Close Redis connections.

Return type:: None

async get_scheduled_changes(flag_id=None, pending_only=True)[source]¶

Get scheduled changes, optionally filtered by flag and status.

Parameters:

flag_id (UUID | None) – If provided, filter to changes for this flag only.
pending_only (bool) – If True, only return changes not yet executed.

Return type:

list[ScheduledFlagChange]

Returns:

List of scheduled changes matching the criteria.

async create_scheduled_change(change)[source]¶

Create a new scheduled change.

Parameters:: change (ScheduledFlagChange) – The scheduled change to create.
Return type:: ScheduledFlagChange
Returns:: The created scheduled change.

async update_scheduled_change(change)[source]¶

Update a scheduled change (e.g., mark as executed).

Parameters:: change (ScheduledFlagChange) – The scheduled change with updated values.
Return type:: ScheduledFlagChange
Returns:: The updated scheduled change.

async get_time_schedules(flag_id=None)[source]¶

Get time schedules for a flag or all flags.

Parameters:: flag_id (UUID | None) – If provided, filter to schedules for this flag only.
Return type:: list[TimeSchedule]
Returns:: List of time schedules matching the criteria.

async create_time_schedule(schedule)[source]¶

Create a new time schedule.

Parameters:: schedule (TimeSchedule) – The time schedule to create.
Return type:: TimeSchedule
Returns:: The created time schedule.

async delete_time_schedule(schedule_id)[source]¶

Delete a time schedule.

Parameters:: schedule_id (UUID) – The UUID of the time schedule to delete.
Return type:: bool
Returns:: True if the schedule was deleted, False if not found.

async get_rollout_phases(flag_id)[source]¶

Get rollout phases for a flag.

Parameters:: flag_id (UUID) – The UUID of the flag.
Return type:: list[RolloutPhase]
Returns:: List of rollout phases for the flag, ordered by phase number.

async create_rollout_phase(phase)[source]¶

Create a new rollout phase.

Parameters:: phase (RolloutPhase) – The rollout phase to create.
Return type:: RolloutPhase
Returns:: The created rollout phase.

Installation¶

uv (Recommended)

uv add litestar-flags[redis]

pip

pip install litestar-flags[redis]

Usage¶

from litestar_flags import FeatureFlagsConfig

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

# With authentication
config = FeatureFlagsConfig(
    backend="redis",
    redis_url="redis://:password@localhost:6379/0",
    redis_prefix="myapp:flags:",  # Custom key prefix
)

# Redis Sentinel
config = FeatureFlagsConfig(
    backend="redis",
    redis_url="redis+sentinel://sentinel1:26379,sentinel2:26379/mymaster/0",
)

Direct Usage¶

from litestar_flags.storage.redis import RedisStorageBackend

storage = await RedisStorageBackend.create(
    url="redis://localhost:6379/0",
    prefix="feature_flags:",
)

client = FeatureFlagClient(storage=storage)

# Close when done
await storage.close()

Key Structure¶

Data is stored with the following key patterns:

Flags: {prefix}flag:{key}
Overrides: {prefix}override:{flag_id}:{entity_type}:{entity_id}
Flag index: {prefix}flags (SET of all flag keys)

When to Use¶

High-performance distributed deployments
When you need fast reads across multiple instances
Microservices architecture
When database latency is a concern

Choosing a Backend¶

Criteria	Memory	Database	Redis
Persistence	No	Yes	Yes (configurable)
Multi-instance	No	Yes	Yes
Performance	Fastest	Good	Very Fast
Transactions	No	Yes	No
Setup complexity	None	Medium	Low

Custom Storage Backends¶

You can implement custom storage backends by implementing the StorageBackend protocol:

from litestar_flags.protocols import StorageBackend
from litestar_flags.models import FeatureFlag, FlagOverride
from collections.abc import Sequence
from uuid import UUID

class MyCustomBackend:
    """Custom storage backend implementation."""

    async def get_flag(self, key: str) -> FeatureFlag | None:
        # Implement flag retrieval
        ...

    async def get_flags(self, keys: Sequence[str]) -> dict[str, FeatureFlag]:
        # Implement bulk retrieval
        ...

    async def get_all_active_flags(self) -> list[FeatureFlag]:
        # Implement active flags retrieval
        ...

    async def get_override(
        self,
        flag_id: UUID,
        entity_type: str,
        entity_id: str,
    ) -> FlagOverride | None:
        # Implement override retrieval
        ...

    async def create_flag(self, flag: FeatureFlag) -> FeatureFlag:
        # Implement flag creation
        ...

    async def update_flag(self, flag: FeatureFlag) -> FeatureFlag:
        # Implement flag update
        ...

    async def delete_flag(self, key: str) -> bool:
        # Implement flag deletion
        ...

    async def health_check(self) -> bool:
        # Implement health check
        return True

    async def close(self) -> None:
        # Clean up resources
        ...

# Use with client
storage = MyCustomBackend()
client = FeatureFlagClient(storage=storage)