API Reference
Protocols
Section titled “Protocols”FlagManagerProtocol
Section titled “FlagManagerProtocol”Manages the lifecycle of feature flag providers and evaluates flags.
The manager chains multiple FlagProviderProtocol instances in order of priority.
Evaluation short-circuits to the first provider that can resolve the flag.
def add_provider( provider: FlagProviderProtocol, priority: int = 50 ) -> None
Register a flag provider at the given priority level.
Higher priority values are queried first. Providers with equal priority are queried in registration order.
| Parameter | Type | Description |
|---|---|---|
| `provider` | FlagProviderProtocol | The flag provider to register. |
| `priority` | int | Resolution priority (higher = queried first). |
Evaluate a boolean feature flag.
| Parameter | Type | Description |
|---|---|---|
| `key` | str | The flag identifier. |
| `context` | dict[str, Any] | None | Optional evaluation context (user, tenant, etc.). |
| Type | Description |
|---|---|
| bool | True if the flag is enabled, False otherwise. |
| Exception | Description |
|---|---|
| FlagNotFoundError | If no provider can resolve the flag. |
async def get_value( key: str, default: FlagValue, context: dict[str, Any] | None = None ) -> FlagValue
Evaluate a feature flag and return its resolved value.
| Parameter | Type | Description |
|---|---|---|
| `key` | str | The flag identifier. |
| `default` | FlagValue | Fallback value when the flag cannot be resolved. |
| `context` | dict[str, Any] | None | Optional evaluation context. |
| Type | Description |
|---|---|
| FlagValue | The resolved FlagValue, or ``default`` if not found. |
Return the full evaluation result including metadata.
| Parameter | Type | Description |
|---|---|---|
| `key` | str | The flag identifier. |
| `context` | dict[str, Any] | None | Optional evaluation context. |
| Type | Description |
|---|---|
| FlagEvaluation | A FlagEvaluation with value, type, reason, and metadata. |
| Exception | Description |
|---|---|
| FlagNotFoundError | If no provider can resolve the flag. |
Return evaluations for all known flags.
| Parameter | Type | Description |
|---|---|---|
| `context` | dict[str, Any] | None | Optional evaluation context applied to every flag. |
| Type | Description |
|---|---|
| dict[str, FlagEvaluation] | Mapping of flag key to its FlagEvaluation. |
FlagProviderProtocol
Section titled “FlagProviderProtocol”Protocol for reading feature flags.
- Supports synchronous and asynchronous access. Implementations may be backed by in-memory maps, redis, remote services, etc.
- Context parameter is provided for user/tenant/environment lookup.
- Variant resolution is part of the contract, allowing A/B testing or percentage-based flags.
- Write methods are deliberately excluded; see MutableFlagProviderProtocol for providers that support updates.
- Async methods are the primary entry points (no suffix). Sync variants
carry a
_syncsuffix.
async def get_flag( name: str, *, default: bool = False, context: dict[str, Any] | None = None ) -> bool
Asynchronously evaluate the boolean value of a flag.
This is the primary entry point for all applications.
context may contain arbitrary data used by the provider to make a
decision (e.g. user id, tenant id, request headers).
def get_flag_sync( name: str, *, default: bool = False, context: dict[str, Any] | None = None ) -> bool
Synchronously evaluate the boolean value of a flag.
Only suitable when the backing store is fully in-memory and no I/O is required. Prefer get_flag in async contexts.
async def get_variant( name: str, *, default: str = '', context: dict[str, Any] | None = None ) -> str
Asynchronously return a string variant for an A/B test or multivalue flag.
This is the primary entry point. Not all providers will support
variants; those may simply return default.
def get_variant_sync( name: str, *, default: str = '', context: dict[str, Any] | None = None ) -> str
Synchronously return a string variant for an A/B test or multivalue flag.
Only suitable when the backing store is fully in-memory. Prefer get_variant in async contexts.
MutableFlagProviderProtocol
Section titled “MutableFlagProviderProtocol”Optional extension of FlagProviderProtocol that supports writes.
Asynchronously create or update a boolean flag value.
This is the primary entry point. Required for providers backed by remote stores (Redis, database, etc.). Implementations with an in-memory backing store may perform the operation synchronously.
Synchronously create or update a boolean flag value.
Only suitable when the backing store is fully in-memory. Prefer set_flag in async contexts.
Asynchronously create or update the active variant for a flag.
This is the primary entry point.
Synchronously create or update the active variant for a flag.
Only suitable when the backing store is fully in-memory. Prefer set_variant in async contexts.
Classes
Section titled “Classes”AbstractFlagProvider
Section titled “AbstractFlagProvider”Rich abstract flag provider with full per-type evaluation logic.
Subclasses must implement get_flag_definition and get_all_flags. All evaluation helpers and the evaluate coroutine are provided here.
This class is not a FlagProvider
protocol implementation — get_flag_definition returns a
Flag model, not a bare boolean.
Use LocalProvider
when you only need the simple boolean protocol.
async def get_flag_definition(name: str) -> Flag | None
Return the full Flag definition or None if not found.
async def get_all_flags() -> dict[str, Flag]
Return all known flag definitions keyed by flag name.
async def evaluate( name: str, context: FlagContext | None = None ) -> FlagEvaluation
Evaluate flag name for context and return a FlagEvaluation.
Returns a flag_not_found evaluation (enabled=False) when the
flag does not exist, so callers can safely treat unknown flags the same
as disabled ones.
def evaluate_sync( name: str, context: FlagContext | None = None ) -> FlagEvaluation
Synchronous evaluation — only valid when the store is in-memory.
Subclasses that back their data with a sync structure should override
this if they need to avoid the event loop. The default raises
NotImplementedError to prevent accidental blocking I/O.
async def get_flag( name: str, *, default: bool = False, context: dict[str, Any] | None = None ) -> bool
Asynchronous boolean access (primary) — delegates to evaluate.
def get_flag_sync( name: str, *, default: bool = False, context: dict[str, Any] | None = None ) -> bool
Synchronous boolean access — delegates to evaluate_sync.
async def get_variant( name: str, *, default: str = '', context: dict[str, Any] | None = None ) -> str
Asynchronous variant access (primary) — delegates to evaluate.
def get_variant_sync( name: str, *, default: str = '', context: dict[str, Any] | None = None ) -> str
Synchronous variant access — delegates to evaluate_sync.
CacheBackendFlagProvider
Section titled “CacheBackendFlagProvider”Feature-flag provider backed by a CacheBackendProtocol.
Flags are stored as JSON-encoded objects in the cache under keys of the
form {prefix}{name}. Each entry is given an independent TTL so that
flags expire and are treated as absent after the TTL elapses.
| Parameter | Type | Description |
|---|---|---|
| `cache` | The cache backend to use for flag storage and retrieval. | |
| `ttl` | Time-to-live in seconds for each cached flag entry. Defaults to 3 600 seconds (1 hour). | |
| `prefix` | Key prefix applied to every flag name to avoid collisions with other cache users. Defaults to ``"lexigram:features:flag:"``. |
def __init__( cache: CacheBackendProtocol, ttl: int = 3600, prefix: str = 'lexigram:features:flag:' ) -> None
async def get_flag(name: str) -> Flag
Return the Flag definition stored under name.
| Parameter | Type | Description |
|---|---|---|
| `name` | str | The flag name to look up. |
| Exception | Description |
|---|---|
| FlagNotFoundError | If no flag with *name* exists in the cache. |
async def add_flag(flag: Flag) -> None
Write flag to the cache with the configured TTL.
Delete the flag with name from the cache.
A no-op if the flag does not exist (the cache backend returns False
from delete, which is silently ignored here).
| Parameter | Type | Description |
|---|---|---|
| `name` | str | The flag name to remove. |
async def list_flags() -> list[Flag]
Return all known flags.
Note This method always returns an empty list because the standard CacheBackendProtocol protocol does not expose a key-scan or key-enumeration operation (e.g. Redis
SCAN). If you need to enumerate flags, maintain a separate index in the cache (a set under a dedicated key) or use a provider that natively supports enumeration such as LocalProvider.
| Type | Description |
|---|---|
| list[Flag] | Always an empty list. |
Evaluate whether the flag name is enabled.
Fetches the flag from the cache and returns its enabled field.
This is a simple boolean check; for richer evaluation (percentage
rollout, user-list, variant, etc.) use a full
AbstractFlagProvider.
| Parameter | Type | Description |
|---|---|---|
| `name` | str | The flag to evaluate. |
| `context` | dict[str, Any] | None | Optional context dict (not used by this provider; included for API consistency with other providers). |
| Type | Description |
|---|---|
| bool | ``True`` if the flag exists and is enabled; ``False`` if the flag is disabled. |
| Exception | Description |
|---|---|
| FlagNotFoundError | If the flag does not exist in the cache. |
ChainedProvider
Section titled “ChainedProvider”Layered provider that queries multiple providers in order.
The first provider that returns a non-None flag definition wins. When merging all flags, earlier providers take precedence over later ones.
def __init__(providers: list[AbstractFlagProvider]) -> None
async def get_flag_definition(name: str) -> Flag | None
Return the first match across the provider chain, or None.
async def get_all_flags() -> dict[str, Flag]
Merge flags from all providers; earlier providers take precedence.
EnvProvider
Section titled “EnvProvider”Feature flag provider that reads from environment variables.
Flag names are lowercased by stripping the prefix.
async def get_flag_definition(name: str) -> Flag | None
Return the parsed Flag for name, or None.
async def get_all_flags() -> dict[str, Flag]
Return all flags parsed from the current environment.
def evaluate_sync( name: str, context: FlagContext | None = None ) -> FlagEvaluation
Synchronous evaluation using the last-loaded env snapshot.
Discard the cached snapshot so the next read re-reads env vars.
FeatureFlagEvaluatedHook
Section titled “FeatureFlagEvaluatedHook”Payload fired when a feature flag is evaluated for a given context.
Attributes: flag_key: Unique key of the feature flag that was evaluated. enabled: Resolved boolean result of the evaluation.
FeatureFlagUpdatedHook
Section titled “FeatureFlagUpdatedHook”Payload fired when a feature flag definition is created or updated.
Attributes: flag_key: Unique key of the feature flag that changed.
FeatureFlagsModule
Section titled “FeatureFlagsModule”In-memory and extensible feature flag evaluation.
Call configure to seed initial flag state from a FeatureFlagsConfig.
Usage
from lexigram.features.config import FeatureFlagsConfig
@module( imports=[ FeatureFlagsModule.configure( FeatureFlagsConfig(initial_flags={"beta_feature": True}) ) ])class AppModule(Module): passdef configure( cls, config: Any | None = None ) -> DynamicModule
Create a FeatureFlagsModule with explicit configuration.
| Parameter | Type | Description |
|---|---|---|
| `config` | Any | None | FeatureFlagsConfig or ``None`` for framework defaults (all flags disabled). |
| Type | Description |
|---|---|
| DynamicModule | A DynamicModule descriptor. |
def stub( cls, config: Any = None ) -> DynamicModule
Return an in-memory FeatureFlagsModule for testing.
Registers the feature flag provider with no flags enabled by default. Suitable for unit and integration tests.
| Type | Description |
|---|---|
| DynamicModule | A DynamicModule with all flags disabled. |
FeatureFlagsProvider
Section titled “FeatureFlagsProvider”Provider that registers feature-flag infrastructure in the container.
Registers:
FlagProviderProtocol(simple boolean contract) as a singleton backed by LocalProvider.FlagManageras a singleton wrapping a LocalProvider seeded from initial_flags.
Create the feature-flags provider.
| Parameter | Type | Description |
|---|---|---|
| `config` | FeatureFlagsConfig | None | Optional feature-flag configuration. When omitted, defaults are used (all flags disabled, cache TTL 60 s). |
Create provider from config object.
async def register(container: ContainerRegistrarProtocol) -> None
Register flag infrastructure in the container.
Registers FlagProviderProtocol (simple boolean API) and FlagManager
(rich evaluation API) as singletons, seeded from the provider config.
Wire the event bus into the flag manager when one is available.
Check health of feature flags.
Feature flags are in-memory and have no external dependencies, so this always returns HEALTHY.
| Parameter | Type | Description |
|---|---|---|
| `timeout` | float | Not used - feature flags have no external dependencies. |
| Type | Description |
|---|---|
| HealthCheckResult | HealthCheckResult showing healthy status. |
No resources to release for the feature-flags module.
def get_simple_provider() -> LocalProvider | None
Return the registered simple provider after registration.
| Type | Description |
|---|---|
| LocalProvider | None | The ``LocalProvider`` instance, or ``None`` before registration. |
def get_manager() -> FlagManager | None
Return the registered manager after registration.
| Type | Description |
|---|---|
| FlagManager | None | The ``FlagManager`` instance, or ``None`` before registration. |
Full definition of a feature flag including its evaluation strategy.
FlagChangeEvent
Section titled “FlagChangeEvent”Published when a feature-flag runtime override is applied or cleared.
Attributes:
flag_name: Name of the feature flag that changed.
old_enabled: State before the change (None if no prior override).
new_enabled: State after the change.
actor: Optional identifier of the entity that triggered the change.
def __init__( flag_name: str, old_enabled: bool | None, new_enabled: bool, actor: str | None = None ) -> None
Create a FlagChangeEvent.
| Parameter | Type | Description |
|---|---|---|
| `flag_name` | str | Name of the feature flag that changed. |
| `old_enabled` | bool | None | Prior override state (``None`` if no override was set). |
| `new_enabled` | bool | New override state after the change. |
| `actor` | str | None | Optional identifier of who triggered the change. |
FlagContext
Section titled “FlagContext”Evaluation context passed alongside a flag name.
Providers use the attributes here to resolve percentage rollouts, user lists, attribute rules, and variant assignments deterministically.
Return an attribute from user_attributes or custom, or default.
Checks user_attributes first, then custom, then returns default.
Serialise to a plain dict for cache key computation.
Return a deterministic 12-char hex hash of the non-empty context fields.
FlagEvaluation
Section titled “FlagEvaluation”Result of evaluating a feature flag for a given context.
enabled is the boolean result; value holds the full evaluation
value — a bool for boolean/percentage/list/attribute/time flags, or a
variant name string for VARIANT flags.
FlagManager
Section titled “FlagManager”Central feature flag manager with caching, overrides, and variant support.
Wraps an AbstractFlagProvider and handles cache invalidation, runtime overrides, and change notification.
Example
provider = LocalProvider({ "new_checkout": Flag("new_checkout", enabled=True),})manager = FlagManager(provider, cache_ttl=60)
ctx = FlagContext(user_id="user-123")if await manager.is_enabled("new_checkout", ctx): ...def __init__( provider: AbstractFlagProvider | None = None, *, cache_ttl: int = 300, default_enabled: bool = False, event_bus: EventBusProtocol | None = None ) -> None
async def is_enabled( name: str, context: FlagContext | None = None, *, default: bool | None = None ) -> bool
Return whether name is enabled for context.
Runtime overrides take precedence over the provider result.
| Parameter | Type | Description |
|---|---|---|
| `name` | str | Flag name. |
| `context` | FlagContext | None | Optional evaluation context. |
| `default` | bool | None | Override the instance's ``default_enabled`` for this call. |
| Type | Description |
|---|---|
| bool | True if the flag is enabled. |
async def evaluate( name: str, context: FlagContext | None = None ) -> FlagEvaluation
Evaluate name for context, using the TTL cache when available.
| Parameter | Type | Description |
|---|---|---|
| `name` | str | Flag name. |
| `context` | FlagContext | None | Optional evaluation context. |
| Type | Description |
|---|---|
| FlagEvaluation | A FlagEvaluation result. |
async def get_variant( name: str, context: FlagContext | None = None, *, default: str = '' ) -> str
Return the variant string for a VARIANT-type flag.
Returns default if the flag is not found, disabled, or is not a VARIANT flag.
| Parameter | Type | Description |
|---|---|---|
| `name` | str | Flag name. |
| `context` | FlagContext | None | Optional evaluation context. |
| `default` | str | Fallback value when no variant string is available. |
| Type | Description |
|---|---|
| str | The variant string or *default*. |
def add_provider( provider: AbstractFlagProvider, priority: int = 50 ) -> None
Register a flag provider at the given priority level.
async def get_value( key: str, default: object, context: FlagContext | None = None ) -> object
Evaluate a feature flag and return its resolved value.
async def get_all_flags(context: FlagContext | None = None) -> dict[str, Any]
Return evaluations for all known flags.
Return the current runtime override for name.
| Type | Description |
|---|---|
| bool | None | ``True`` if force-enabled, ``False`` if force-disabled, ``None`` if no override is active. |
Force-enable name regardless of provider result.
Force-disable name regardless of provider result.
Convenience wrapper that calls enable or disable.
Remove any runtime override for name, restoring provider control.
Return a copy of all recorded flag change audit entries.
Each entry captures the flag name, actor, old value, new value, and the UTC timestamp of the change. Entries are ordered oldest-first.
| Type | Description |
|---|---|
| list[FlagAuditEntry] | List of FlagAuditEntry instances representing every enable / disable / set_override call since this manager was created. |
Flush all cached evaluations.
Flush cached evaluations for a single flag name.
Register a sync callback invoked when a flag override changes.
Deregister a previously registered sync listener.
Register an async callback invoked when a flag override changes.
Deregister a previously registered async listener.
property provider() -> AbstractFlagProvider
The underlying flag provider instance.
FlagType
Section titled “FlagType”Evaluation strategy for a feature flag.
LocalProvider
Section titled “LocalProvider”In-memory provider backed by a ``dict[str, Flag]``.
Dynamic update methods (add_flag, remove_flag, set_enabled) mutate the store safely within a single process.
def __init__(flags: dict[str, Flag] | None = None) -> None
async def get_flag_definition(name: str) -> Flag | None
Return the Flag definition for name, or None.
async def get_all_flags() -> dict[str, Flag]
Return a snapshot copy of all defined flags.
def evaluate_sync( name: str, context: FlagContext | None = None ) -> FlagEvaluation
Synchronous evaluation — safe because the backing store is in-memory.
def add_flag(flag: Flag) -> None
Add or replace a flag definition.
Remove a flag by name; returns True if it existed.
Toggle the master enabled switch; returns True if the flag exists.
Store a boolean flag (async primary).
Store a boolean flag (sync).
Store a variant flag (async primary).
Store a variant flag (sync).
Remove all stored flags and variants.
ManagerConfig
Section titled “ManagerConfig”Configuration for FlagManager instances.
MemoryProvider
Section titled “MemoryProvider”Lightweight in-memory provider designed for use in tests.
Unlike LocalProvider this provider works with plain booleans and supports per-flag overrides that short-circuit normal evaluation.
Note Choosing between in-memory providers:
* Use MemoryProvider (this class) in **unit tests** — the ``_overrides`` dict lets you force specific flag outcomes without touching the flag configuration.
* Use <a href='/packages/infra/lexigram-features/api/#localprovider' style='color:inherit; text-decoration:underline; text-decoration-color:rgba(128,128,128,0.3); text-underline-offset:2px;'>LocalProvider</a> for production single-node deployments or integration tests where you only need ``set_flag`` / ``get_flag`` semantics and do not need per-test overrides.Typical usage in tests
provider = MemoryProvider()provider.set_flag("new_billing", enabled=True)manager = FlagManager(provider)async def evaluate( name: str, context: FlagContext | None = None ) -> FlagEvaluation
Evaluate, applying any explicit test override first.
def evaluate_sync( name: str, context: FlagContext | None = None ) -> FlagEvaluation
Synchronous evaluation, applying overrides first.
Define a simple boolean flag (sync version).
Define a simple boolean flag.
def override( name: str, *, enabled: bool, value: Any = None, reason: str = 'test_override' ) -> None
Register a hard override for name that bypasses normal evaluation.
Remove the test override for name.
Clear all flags and overrides.
Functions
Section titled “Functions”feature_flag
Section titled “feature_flag”
def feature_flag( name: str, *, manager: FlagManager | None = None, fallback: Callable[Ellipsis, Any] | None = None, context: FlagContext | None = None ) -> Callable[[F], F]
Decorate an async function so it only runs when name is enabled.
When the flag is disabled:
- If fallback is provided it is called with the same arguments and its return value is returned.
- Otherwise FeatureFlagDisabledError is raised.
| Parameter | Type | Description |
|---|---|---|
| `name` | str | Feature flag name to check. |
| `manager` | FlagManager | None | Explicit FlagManager; resolved via DI or a default when ``None``. |
| `fallback` | Callable[Ellipsis, Any] | None | Optional callable invoked when the flag is disabled. |
| `context` | FlagContext | None | Optional evaluation context passed to the manager. |
feature_flag_sync
Section titled “feature_flag_sync”
def feature_flag_sync( name: str, *, manager: FlagManager | None = None, fallback: Callable[Ellipsis, Any] | None = None, context: FlagContext | None = None ) -> Callable[[F], F]
Decorate a synchronous function so it only runs when name is enabled.
Evaluation uses evaluate_sync on the underlying provider.
| Parameter | Type | Description |
|---|---|---|
| `name` | str | Feature flag name to check. |
| `manager` | FlagManager | None | Explicit manager; resolved via DI or a default when ``None``. |
| `fallback` | Callable[Ellipsis, Any] | None | Optional callable invoked when the flag is disabled. |
| `context` | FlagContext | None | Optional evaluation context. |
require_flag
Section titled “require_flag”
def require_flag( name: str, *, manager: FlagManager | None = None, context: FlagContext | None = None ) -> Callable[[F], F]
Decorate an async function to raise when name is disabled.
Semantically identical to feature_flag with no fallback, but the intent is clearer when used as an access guard.
| Parameter | Type | Description |
|---|---|---|
| `name` | str | Feature flag name to check. |
| `manager` | FlagManager | None | Explicit manager; resolved via DI or a default when ``None``. |
| `context` | FlagContext | None | Optional evaluation context. |
require_flag_sync
Section titled “require_flag_sync”
def require_flag_sync( name: str, *, manager: FlagManager | None = None, context: FlagContext | None = None ) -> Callable[[F], F]
Synchronous variant of require_flag.
Raises FeatureFlagDisabledError when name is disabled.
| Parameter | Type | Description |
|---|---|---|
| `name` | str | Feature flag name to check. |
| `manager` | FlagManager | None | Explicit manager; resolved via DI or a default when ``None``. |
| `context` | FlagContext | None | Optional evaluation context. |
Exceptions
Section titled “Exceptions”FeatureFlagDisabledError
Section titled “FeatureFlagDisabledError”Raised when a feature-guarded path is called with the flag disabled.
FeatureFlagError
Section titled “FeatureFlagError”Base error for the feature flag subsystem.
FlagEvaluationError
Section titled “FlagEvaluationError”Raised when a flag provider fails during evaluation.
Attributes: flag_key: The key of the flag that failed to evaluate.
FlagNotFoundError
Section titled “FlagNotFoundError”Raised when a requested feature flag does not exist in any provider.
Attributes: flag_key: The key of the missing flag.