Skip to content
GitHubDiscord

API Reference

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.

Parameters
ParameterTypeDescription
`provider`FlagProviderProtocolThe flag provider to register.
`priority`intResolution priority (higher = queried first).
async def is_enabled(
    key: str,
    context: dict[str, Any] | None = None
) -> bool

Evaluate a boolean feature flag.

Parameters
ParameterTypeDescription
`key`strThe flag identifier.
`context`dict[str, Any] | NoneOptional evaluation context (user, tenant, etc.).
Returns
TypeDescription
boolTrue if the flag is enabled, False otherwise.
Raises
ExceptionDescription
FlagNotFoundErrorIf 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.

Parameters
ParameterTypeDescription
`key`strThe flag identifier.
`default`FlagValueFallback value when the flag cannot be resolved.
`context`dict[str, Any] | NoneOptional evaluation context.
Returns
TypeDescription
FlagValueThe resolved FlagValue, or ``default`` if not found.
async def evaluate(
    key: str,
    context: dict[str, Any] | None = None
) -> FlagEvaluation

Return the full evaluation result including metadata.

Parameters
ParameterTypeDescription
`key`strThe flag identifier.
`context`dict[str, Any] | NoneOptional evaluation context.
Returns
TypeDescription
FlagEvaluationA FlagEvaluation with value, type, reason, and metadata.
Raises
ExceptionDescription
FlagNotFoundErrorIf no provider can resolve the flag.
async def get_all_flags(context: dict[str, Any] | None = None) -> dict[str, FlagEvaluation]

Return evaluations for all known flags.

Parameters
ParameterTypeDescription
`context`dict[str, Any] | NoneOptional evaluation context applied to every flag.
Returns
TypeDescription
dict[str, FlagEvaluation]Mapping of flag key to its FlagEvaluation.

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 _sync suffix.
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.


Optional extension of FlagProviderProtocol that supports writes.
async def set_flag(
    name: str,
    value: bool
) -> None

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.

def set_flag_sync(
    name: str,
    value: bool
) -> None

Synchronously create or update a boolean flag value.

Only suitable when the backing store is fully in-memory. Prefer set_flag in async contexts.

async def set_variant(
    name: str,
    variant: str
) -> None

Asynchronously create or update the active variant for a flag.

This is the primary entry point.

def set_variant_sync(
    name: str,
    variant: str
) -> None

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.


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.


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.

Parameters
ParameterTypeDescription
`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.

Parameters
ParameterTypeDescription
`name`strThe flag name to look up.
Returns
TypeDescription
FlagThe Flag stored in the cache.
Raises
ExceptionDescription
FlagNotFoundErrorIf no flag with *name* exists in the cache.
async def add_flag(flag: Flag) -> None

Write flag to the cache with the configured TTL.

Parameters
ParameterTypeDescription
`flag`FlagThe Flag definition to persist.
async def remove_flag(name: str) -> None

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).

Parameters
ParameterTypeDescription
`name`strThe 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.

Returns
TypeDescription
list[Flag]Always an empty list.
async def evaluate(
    name: str,
    context: dict[str, Any] | None = None
) -> bool

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.

Parameters
ParameterTypeDescription
`name`strThe flag to evaluate.
`context`dict[str, Any] | NoneOptional context dict (not used by this provider; included for API consistency with other providers).
Returns
TypeDescription
bool``True`` if the flag exists and is enabled; ``False`` if the flag is disabled.
Raises
ExceptionDescription
FlagNotFoundErrorIf the flag does not exist in the cache.

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.


Feature flag provider that reads from environment variables.

Flag names are lowercased by stripping the prefix.

def __init__(prefix: str = ENV_PREFIX) -> None
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.

def invalidate() -> None

Discard the cached snapshot so the next read re-reads env vars.


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.


Payload fired when a feature flag definition is created or updated.

Attributes: flag_key: Unique key of the feature flag that changed.


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):
pass
def configure(
    cls,
    config: Any | None = None
) -> DynamicModule

Create a FeatureFlagsModule with explicit configuration.

Parameters
ParameterTypeDescription
`config`Any | NoneFeatureFlagsConfig or ``None`` for framework defaults (all flags disabled).
Returns
TypeDescription
DynamicModuleA 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.

Returns
TypeDescription
DynamicModuleA DynamicModule with all flags disabled.

Provider that registers feature-flag infrastructure in the container.

Registers:

  • FlagProviderProtocol (simple boolean contract) as a singleton backed by LocalProvider.
  • FlagManager as a singleton wrapping a LocalProvider seeded from initial_flags.
def __init__(config: FeatureFlagsConfig | None = None) -> None

Create the feature-flags provider.

Parameters
ParameterTypeDescription
`config`FeatureFlagsConfig | NoneOptional feature-flag configuration. When omitted, defaults are used (all flags disabled, cache TTL 60 s).
def from_config(
    cls,
    config: FeatureFlagsConfig,
    **context: Any
) -> Self

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.

async def boot(container: BootContainerProtocol) -> None

Wire the event bus into the flag manager when one is available.

async def health_check(timeout: float = 5.0) -> HealthCheckResult

Check health of feature flags.

Feature flags are in-memory and have no external dependencies, so this always returns HEALTHY.

Parameters
ParameterTypeDescription
`timeout`floatNot used - feature flags have no external dependencies.
Returns
TypeDescription
HealthCheckResultHealthCheckResult showing healthy status.
async def shutdown() -> None

No resources to release for the feature-flags module.

def get_simple_provider() -> LocalProvider | None

Return the registered simple provider after registration.

Returns
TypeDescription
LocalProvider | NoneThe ``LocalProvider`` instance, or ``None`` before registration.
def get_manager() -> FlagManager | None

Return the registered manager after registration.

Returns
TypeDescription
FlagManager | NoneThe ``FlagManager`` instance, or ``None`` before registration.

Full definition of a feature flag including its evaluation strategy.

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.

Parameters
ParameterTypeDescription
`flag_name`strName of the feature flag that changed.
`old_enabled`bool | NonePrior override state (``None`` if no override was set).
`new_enabled`boolNew override state after the change.
`actor`str | NoneOptional identifier of who triggered the change.

Evaluation context passed alongside a flag name.

Providers use the attributes here to resolve percentage rollouts, user lists, attribute rules, and variant assignments deterministically.

def get_attribute(
    key: str,
    default: Any = None
) -> Any

Return an attribute from user_attributes or custom, or default.

Checks user_attributes first, then custom, then returns default.

def as_dict() -> dict[str, Any]

Serialise to a plain dict for cache key computation.

def context_hash() -> str

Return a deterministic 12-char hex hash of the non-empty context fields.


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.


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.

Parameters
ParameterTypeDescription
`name`strFlag name.
`context`FlagContext | NoneOptional evaluation context.
`default`bool | NoneOverride the instance's ``default_enabled`` for this call.
Returns
TypeDescription
boolTrue 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.

Parameters
ParameterTypeDescription
`name`strFlag name.
`context`FlagContext | NoneOptional evaluation context.
Returns
TypeDescription
FlagEvaluationA 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.

Parameters
ParameterTypeDescription
`name`strFlag name.
`context`FlagContext | NoneOptional evaluation context.
`default`strFallback value when no variant string is available.
Returns
TypeDescription
strThe 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.

def get_override_state(name: str) -> bool | None

Return the current runtime override for name.

Returns
TypeDescription
bool | None``True`` if force-enabled, ``False`` if force-disabled, ``None`` if no override is active.
def enable(
    name: str,
    *,
    actor: str | None = None
) -> None

Force-enable name regardless of provider result.

def disable(
    name: str,
    *,
    actor: str | None = None
) -> None

Force-disable name regardless of provider result.

def set_override(
    name: str,
    enabled: bool,
    *,
    actor: str | None = None
) -> None

Convenience wrapper that calls enable or disable.

def clear_override(name: str) -> None

Remove any runtime override for name, restoring provider control.

def get_audit_log() -> list[FlagAuditEntry]

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.

Returns
TypeDescription
list[FlagAuditEntry]List of FlagAuditEntry instances representing every enable / disable / set_override call since this manager was created.
async def clear_cache() -> None

Flush all cached evaluations.

def clear_cache_for(name: str) -> None

Flush cached evaluations for a single flag name.

def add_listener_sync(fn: FlagChangeListener) -> None

Register a sync callback invoked when a flag override changes.

def remove_listener_sync(fn: FlagChangeListener) -> None

Deregister a previously registered sync listener.

def add_listener(fn: AsyncFlagChangeListener) -> None

Register an async callback invoked when a flag override changes.

def remove_listener(fn: AsyncFlagChangeListener) -> None

Deregister a previously registered async listener.

property provider() -> AbstractFlagProvider

The underlying flag provider instance.


Evaluation strategy for a feature flag.

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.

def remove_flag(name: str) -> bool

Remove a flag by name; returns True if it existed.

def set_enabled(
    name: str,
    enabled: bool
) -> bool

Toggle the master enabled switch; returns True if the flag exists.

async def set_flag(
    name: str,
    value: bool
) -> None

Store a boolean flag (async primary).

def set_flag_sync(
    name: str,
    value: bool
) -> None

Store a boolean flag (sync).

async def set_variant(
    name: str,
    variant: str
) -> None

Store a variant flag (async primary).

def set_variant_sync(
    name: str,
    variant: str
) -> None

Store a variant flag (sync).

def clear() -> None

Remove all stored flags and variants.


Configuration for FlagManager instances.

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)
def __init__() -> None
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.

def set_flag_sync(
    name: str,
    value: bool
) -> None

Define a simple boolean flag (sync version).

async def set_flag(
    name: str,
    value: bool
) -> None

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.

def clear_override(name: str) -> None

Remove the test override for name.

def reset() -> None

Clear all flags and overrides.


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.
Parameters
ParameterTypeDescription
`name`strFeature flag name to check.
`manager`FlagManager | NoneExplicit FlagManager; resolved via DI or a default when ``None``.
`fallback`Callable[Ellipsis, Any] | NoneOptional callable invoked when the flag is disabled.
`context`FlagContext | NoneOptional evaluation context passed to the manager.
Returns
TypeDescription
Callable[[F], F]A decorator for async functions.

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.

Parameters
ParameterTypeDescription
`name`strFeature flag name to check.
`manager`FlagManager | NoneExplicit manager; resolved via DI or a default when ``None``.
`fallback`Callable[Ellipsis, Any] | NoneOptional callable invoked when the flag is disabled.
`context`FlagContext | NoneOptional evaluation context.
Returns
TypeDescription
Callable[[F], F]A decorator for synchronous functions.

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.

Parameters
ParameterTypeDescription
`name`strFeature flag name to check.
`manager`FlagManager | NoneExplicit manager; resolved via DI or a default when ``None``.
`context`FlagContext | NoneOptional evaluation context.
Returns
TypeDescription
Callable[[F], F]A decorator for async functions.

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.

Parameters
ParameterTypeDescription
`name`strFeature flag name to check.
`manager`FlagManager | NoneExplicit manager; resolved via DI or a default when ``None``.
`context`FlagContext | NoneOptional evaluation context.
Returns
TypeDescription
Callable[[F], F]A decorator for synchronous functions.

Raised when a feature-guarded path is called with the flag disabled.
def __init__(flag_name: str) -> None

Base error for the feature flag subsystem.
def __init__(
    message: str = 'Feature flag error',
    **kwargs: Any
) -> None

Raised when a flag provider fails during evaluation.

Attributes: flag_key: The key of the flag that failed to evaluate.

def __init__(
    flag_key: str,
    message: str = 'Flag evaluation failed',
    **kwargs: Any
) -> None

Raised when a requested feature flag does not exist in any provider.

Attributes: flag_key: The key of the missing flag.

def __init__(
    flag_key: str,
    **kwargs: Any
) -> None