Skip to content
GitHub

Feature Flags

lexigram-features provides protocol-backed feature flag evaluation. Application code depends on FlagManagerProtocol; the underlying store — in-memory, environment variables, a shared cache, or your own adapter — is chosen in configuration. Evaluations are cached in-process, targeting is driven by an explicit context object, and runtime overrides let operators force outcomes without redeploying.

For the full configuration reference, decorators, and audit log, see the lexigram-features package docs.


Services depend on FlagManagerProtocol. The manager wraps one or more providers, applies caching and overrides, and returns either a plain boolean or a full FlagEvaluation.

from typing import Any, Protocol, runtime_checkable
from lexigram.contracts.feature_flags.models import FlagEvaluation, FlagValue
@runtime_checkable
class FlagManagerProtocol(Protocol):
async def is_enabled(self, key: str, context: dict[str, Any] | None = None) -> bool: ...
async def get_value(self, key: str, default: FlagValue, context: dict[str, Any] | None = None) -> FlagValue: ...
async def evaluate(self, key: str, context: dict[str, Any] | None = None) -> FlagEvaluation: ...
async def get_all_flags(self, context: dict[str, Any] | None = None) -> dict[str, FlagEvaluation]: ...

Targeting data is carried in a FlagContext value object — user id, session id, and arbitrary attributes used by percentage, user-list, and attribute-rule flags.


Register the module and configure the features: section. With no config block the subsystem boots with safe defaults (300-second cache, no seed flags, default-disabled fallback).

from lexigram import Application
from lexigram.features import FeatureFlagsModule
app = Application(name="my-app")
app.add_module(FeatureFlagsModule.configure())
application.yaml
features:
enabled: true
cache_ttl: 300 # seconds; 0 disables the evaluation cache
default_enabled: false # fallback when a flag is not defined
flag_env_prefix: "LEX_FLAG_"
initial_flags:
new_checkout: true
legacy_search: false

Each entry in initial_flags becomes a boolean Flag seeded into the default LocalProvider. For rich flag types (percentage rollouts, user lists, variants) construct Flag objects in code:

from lexigram.features import Flag, FlagType, LocalProvider
provider = LocalProvider({
"new_search": Flag(name="new_search", type=FlagType.PERCENTAGE, percentage=25),
"beta_users": Flag(name="beta_users", type=FlagType.USER_LIST,
user_list=["user-1", "user-7", "user-42"]),
})

Inject FlagManagerProtocol (or the concrete FlagManager) and call is_enabled at the decision point:

from lexigram.contracts.feature_flags import FlagManagerProtocol
from lexigram.features import FlagContext
class CheckoutService:
def __init__(self, flags: FlagManagerProtocol) -> None:
self._flags = flags
async def checkout(self, user_id: str, cart: Cart) -> Receipt:
ctx = FlagContext(user_id=user_id, user_attributes={"tier": cart.tier})
if await self._flags.is_enabled("new_checkout", ctx):
return await self._new_flow(cart)
return await self._legacy_flow(cart)

Unknown flags return default_enabled (configurable per-call via default=). Provider errors are logged and degrade to the configured default — flag lookups never raise into the request path.


A Flag carries its evaluation strategy. The shipped strategies are BOOLEAN, PERCENTAGE, USER_LIST, USER_ATTRIBUTE, TIME_BASED, and VARIANT. Percentage rollouts hash the user_id so the same user always lands on the same side of the rollout.

For A/B tests, define a VARIANT flag with weights summing to 100 and call get_variant:

from lexigram.features import Flag, FlagType, FlagContext, FlagManager, LocalProvider
provider = LocalProvider({
"homepage_layout": Flag(
name="homepage_layout",
type=FlagType.VARIANT,
variants={"control": 50, "compact": 25, "rich": 25},
default_variant="control",
),
})
manager = FlagManager(provider, cache_ttl=60)
variant = await manager.get_variant("homepage_layout", FlagContext(user_id="user-7"))
# -> "control" | "compact" | "rich", stable for this user

FlagManager caches every evaluation in-process for cache_ttl seconds (default 300). The cache key is the flag name plus a deterministic hash of the non-empty FlagContext fields, so the same user-and-flag combination is a single entry. Set cache_ttl: 0 to disable. Force a refresh with await manager.clear_cache() or manager.clear_cache_for(name) after pushing a flag change.

Runtime overrides take precedence over both cache and provider:

manager.enable("new_checkout", actor="ops@example.com") # force-on
manager.disable("legacy_search", actor="ops@example.com") # force-off
manager.clear_override("new_checkout") # restore provider control

Every override is recorded; inspect manager.get_audit_log() for the history.


lexigram-features ships these backends — all implement AbstractFlagProvider and plug into FlagManager interchangeably:

BackendPurpose
LocalProviderIn-memory definitions, the default
EnvProviderReads flags from environment variables (LEX_FLAG_* by default)
ChainedProviderQueries several providers in order; first hit wins
CacheBackendFlagProviderStores flag definitions in any CacheBackendProtocol (Redis, memory)
MemoryProviderTest double with explicit per-flag overrides

A common production layout chains env-vars over a shared cache over local defaults:

from lexigram.features import (
ChainedProvider, EnvProvider, LocalProvider, CacheBackendFlagProvider, FlagManager,
)
provider = ChainedProvider([
EnvProvider(prefix="LEX_FLAG_"), # ops overrides
CacheBackendFlagProvider(cache, ttl=3600), # shared store across instances
LocalProvider(seed_flags), # baked-in defaults
])
manager = FlagManager(provider, cache_ttl=60)

EnvProvider understands true/false, percentage:25, and users:alice,bob value formats — useful for one-off operator toggles without redeploying config.


FeatureFlagsModule.stub() boots the subsystem with an empty in-memory provider, suitable for unit and integration tests:

from lexigram import Application
from lexigram.features import FeatureFlagsModule, FlagManager, MemoryProvider
async def test_checkout_uses_new_flow() -> None:
async with Application.boot(modules=[FeatureFlagsModule.stub()]) as app:
manager = await app.container.resolve(FlagManager)
manager.enable("new_checkout")
assert await manager.is_enabled("new_checkout") is True

For lower-level tests, use MemoryProvider directly — it supports hard per-flag overrides that bypass evaluation entirely:

provider = MemoryProvider()
provider.override("new_checkout", enabled=True, reason="test_override")
manager = FlagManager(provider)