Skip to content

forge.featureflags

Feature flags with boolean, percentage rollout, and user-segment evaluation.

forge.featureflags

Feature Flags module — boolean, percentage rollout, and user-segment evaluation.

Provides typed flag definitions, evaluation context, consistent-hash based percentage rollout, in-memory and Redis-backed storage, and CLI management.

Classes

EvaluationContext

Bases: BaseModel

Contextual information used during flag evaluation.

Source code in src/forge/featureflags/models.py 
17
18
19
20
21
22
class EvaluationContext(BaseModel):
    """Contextual information used during flag evaluation."""

    user_id: str
    region: str = ""
    properties: dict[str, Any] = Field(default_factory=dict)

EvaluationReason

Bases: StrEnum

Reason why a flag evaluated to a particular value.

Source code in src/forge/featureflags/models.py 
52
53
54
55
56
57
58
59
60
class EvaluationReason(enum.StrEnum):
    """Reason why a flag evaluated to a particular value."""

    DEFAULT = "default"
    OVERRIDE = "override"
    RULE_MATCH = "rule_match"
    PERCENTAGE_ROLLOUT = "percentage_rollout"
    SEGMENT_MATCH = "segment_match"
    NO_MATCH = "no_match"

EvaluationResult

Bases: BaseModel

Result of a single flag evaluation.

Source code in src/forge/featureflags/models.py 
63
64
65
66
67
68
69
class EvaluationResult(BaseModel):
    """Result of a single flag evaluation."""

    flag_name: str
    value: Any
    reason: EvaluationReason = EvaluationReason.DEFAULT
    matched_rule_index: int | None = None

FeatureFlagError

Bases: ForgeError

Base exception for all feature flag-related errors.

Source code in src/forge/featureflags/exceptions.py 
6
7
class FeatureFlagError(ForgeError):
    """Base exception for all feature flag-related errors."""

FeatureFlagsModule

Bases: ForgeModule

Manages feature flag definitions, storage, and evaluation.

Supports boolean, percentage rollout, and user-segment based flag evaluation with in-memory and optional Redis backends.

Source code in src/forge/featureflags/module.py 
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
class FeatureFlagsModule(ForgeModule):
    """
    Manages feature flag definitions, storage, and evaluation.

    Supports boolean, percentage rollout, and user-segment based flag
    evaluation with in-memory and optional Redis backends.
    """

    name = "featureflags"
    dependencies: ClassVar[list[str]] = ["config"]

    def __init__(self) -> None:
        super().__init__()
        self._store: FlagStore | None = None
        self._evaluator: FlagEvaluator | None = None
        self._runtime: Runtime | None = None

    @property
    def store(self) -> FlagStore:
        if self._store is None:
            raise FeatureFlagError("Feature flag store is not initialized.")
        return self._store

    @property
    def evaluator(self) -> FlagEvaluator:
        if self._evaluator is None:
            raise FeatureFlagError("Feature flag evaluator is not initialized.")
        return self._evaluator

    async def setup(self, runtime: Runtime) -> None:
        """Initialize the feature flags module."""
        self._runtime = runtime
        set_featureflags_module(self)

        from forge.config.module import ConfigModule

        config_module = cast("ConfigModule", runtime.get(ConfigModule))
        config = getattr(config_module.config, "featureflags", None)

        backend_type = "memory"
        redis_url = None
        redis_key_prefix = "forge:featureflags:"
        redis_max_connections = 10
        flags_config: list[dict[str, Any]] | None = None

        if config is not None:
            backend_type = getattr(config, "backend", "memory")
            flags_config = getattr(config, "flags", None)
            redis_config = getattr(config, "redis", None)
            if redis_config is not None:
                redis_url = getattr(redis_config, "url", None)
                redis_key_prefix = getattr(redis_config, "key_prefix", "forge:featureflags:")
                redis_max_connections = getattr(redis_config, "max_connections", 10)

        if backend_type == "redis":
            from forge.featureflags.store import RedisFlagStore

            url = redis_url or "redis://localhost:6379/0"
            redis_store = RedisFlagStore(
                redis_url=url,
                key_prefix=redis_key_prefix,
                max_connections=redis_max_connections,
            )
            await redis_store.connect()
            self._store = redis_store
        else:
            self._store = MemoryFlagStore()

        self._evaluator = FlagEvaluator(self._store)

        # Pre-load flags from config if provided
        if flags_config:
            for flag_data in flags_config:
                flag = FlagDefinition.model_validate(flag_data)
                await self._store.set_flag(flag)

    async def teardown(self) -> None:
        """Teardown the feature flags module."""
        set_featureflags_module(None)
        if self._store:
            try:
                await self._store.close()
            except Exception as exc:
                import logging

                logging.getLogger(__name__).warning("Error closing flag store: %s", exc)
        self._store = None
        self._evaluator = None
        self._runtime = None

    async def evaluate(
        self,
        flag_name: str,
        context: EvaluationContext,
    ) -> EvaluationResult:
        """Evaluate a single flag for the given context."""
        return await self.evaluator.evaluate(flag_name, context)

    async def evaluate_bulk(
        self,
        flag_names: list[str],
        context: EvaluationContext,
    ) -> dict[str, EvaluationResult]:
        """Evaluate multiple flags for the given context."""
        return await self.evaluator.evaluate_bulk(flag_names, context)

    async def evaluate_all(
        self,
        context: EvaluationContext,
    ) -> dict[str, EvaluationResult]:
        """Evaluate all flags for the given context."""
        return await self.evaluator.evaluate_all(context)

    async def get_flag(self, name: str) -> FlagDefinition | None:
        """Get a flag definition by name."""
        return await self.store.get_flag(name)

    async def set_flag(self, flag: FlagDefinition) -> None:
        """Store or update a flag definition."""
        await self.store.set_flag(flag)

    async def delete_flag(self, name: str) -> bool:
        """Delete a flag definition by name."""
        return await self.store.delete_flag(name)

    async def list_flags(self) -> list[FlagDefinition]:
        """List all stored flag definitions."""
        return await self.store.list_flags()

    def _health_check_redis(self, store: Any) -> HealthResult:
        """Check health of Redis-backed flag store."""
        if not store.is_connected:
            return HealthResult.error("Redis flag store not connected")
        from forge.core.redis_health import check_redis_health

        return check_redis_health(store.url, label="Redis flag store")

    def health_check(self) -> HealthResult:
        """Check the health status of the feature flags backend."""
        if self._store is None:
            return HealthResult.error("Feature flag store not initialized")

        if isinstance(self._store, MemoryFlagStore):
            return HealthResult(HealthResult.OK, "Memory flag store active")

        from forge.featureflags.store import RedisFlagStore

        if isinstance(self._store, RedisFlagStore):
            return self._health_check_redis(self._store)

        return HealthResult.ok()
Methods:
delete_flag async 
delete_flag(name: str) -> bool

Delete a flag definition by name.

Source code in src/forge/featureflags/module.py 
137
138
139
async def delete_flag(self, name: str) -> bool:
    """Delete a flag definition by name."""
    return await self.store.delete_flag(name)
evaluate async 
evaluate(flag_name: str, context: EvaluationContext) -> EvaluationResult

Evaluate a single flag for the given context.

Source code in src/forge/featureflags/module.py 
106
107
108
109
110
111
112
async def evaluate(
    self,
    flag_name: str,
    context: EvaluationContext,
) -> EvaluationResult:
    """Evaluate a single flag for the given context."""
    return await self.evaluator.evaluate(flag_name, context)
evaluate_all async 
evaluate_all(context: EvaluationContext) -> dict[str, EvaluationResult]

Evaluate all flags for the given context.

Source code in src/forge/featureflags/module.py 
122
123
124
125
126
127
async def evaluate_all(
    self,
    context: EvaluationContext,
) -> dict[str, EvaluationResult]:
    """Evaluate all flags for the given context."""
    return await self.evaluator.evaluate_all(context)
evaluate_bulk async 
evaluate_bulk(flag_names: list[str], context: EvaluationContext) -> dict[str, EvaluationResult]

Evaluate multiple flags for the given context.

Source code in src/forge/featureflags/module.py 
114
115
116
117
118
119
120
async def evaluate_bulk(
    self,
    flag_names: list[str],
    context: EvaluationContext,
) -> dict[str, EvaluationResult]:
    """Evaluate multiple flags for the given context."""
    return await self.evaluator.evaluate_bulk(flag_names, context)
get_flag async 
get_flag(name: str) -> FlagDefinition | None

Get a flag definition by name.

Source code in src/forge/featureflags/module.py 
129
130
131
async def get_flag(self, name: str) -> FlagDefinition | None:
    """Get a flag definition by name."""
    return await self.store.get_flag(name)
health_check 
health_check() -> HealthResult

Check the health status of the feature flags backend.

Source code in src/forge/featureflags/module.py 
153
154
155
156
157
158
159
160
161
162
163
164
165
166
def health_check(self) -> HealthResult:
    """Check the health status of the feature flags backend."""
    if self._store is None:
        return HealthResult.error("Feature flag store not initialized")

    if isinstance(self._store, MemoryFlagStore):
        return HealthResult(HealthResult.OK, "Memory flag store active")

    from forge.featureflags.store import RedisFlagStore

    if isinstance(self._store, RedisFlagStore):
        return self._health_check_redis(self._store)

    return HealthResult.ok()
list_flags async 
list_flags() -> list[FlagDefinition]

List all stored flag definitions.

Source code in src/forge/featureflags/module.py 
141
142
143
async def list_flags(self) -> list[FlagDefinition]:
    """List all stored flag definitions."""
    return await self.store.list_flags()
set_flag async 
set_flag(flag: FlagDefinition) -> None

Store or update a flag definition.

Source code in src/forge/featureflags/module.py 
133
134
135
async def set_flag(self, flag: FlagDefinition) -> None:
    """Store or update a flag definition."""
    await self.store.set_flag(flag)
setup async 
setup(runtime: ForgeRuntime) -> None

Initialize the feature flags module.

Source code in src/forge/featureflags/module.py 
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
async def setup(self, runtime: Runtime) -> None:
    """Initialize the feature flags module."""
    self._runtime = runtime
    set_featureflags_module(self)

    from forge.config.module import ConfigModule

    config_module = cast("ConfigModule", runtime.get(ConfigModule))
    config = getattr(config_module.config, "featureflags", None)

    backend_type = "memory"
    redis_url = None
    redis_key_prefix = "forge:featureflags:"
    redis_max_connections = 10
    flags_config: list[dict[str, Any]] | None = None

    if config is not None:
        backend_type = getattr(config, "backend", "memory")
        flags_config = getattr(config, "flags", None)
        redis_config = getattr(config, "redis", None)
        if redis_config is not None:
            redis_url = getattr(redis_config, "url", None)
            redis_key_prefix = getattr(redis_config, "key_prefix", "forge:featureflags:")
            redis_max_connections = getattr(redis_config, "max_connections", 10)

    if backend_type == "redis":
        from forge.featureflags.store import RedisFlagStore

        url = redis_url or "redis://localhost:6379/0"
        redis_store = RedisFlagStore(
            redis_url=url,
            key_prefix=redis_key_prefix,
            max_connections=redis_max_connections,
        )
        await redis_store.connect()
        self._store = redis_store
    else:
        self._store = MemoryFlagStore()

    self._evaluator = FlagEvaluator(self._store)

    # Pre-load flags from config if provided
    if flags_config:
        for flag_data in flags_config:
            flag = FlagDefinition.model_validate(flag_data)
            await self._store.set_flag(flag)
teardown async 
teardown() -> None

Teardown the feature flags module.

Source code in src/forge/featureflags/module.py 
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
async def teardown(self) -> None:
    """Teardown the feature flags module."""
    set_featureflags_module(None)
    if self._store:
        try:
            await self._store.close()
        except Exception as exc:
            import logging

            logging.getLogger(__name__).warning("Error closing flag store: %s", exc)
    self._store = None
    self._evaluator = None
    self._runtime = None

FlagDefinition

Bases: BaseModel

Definition of a single feature flag.

Source code in src/forge/featureflags/models.py 
41
42
43
44
45
46
47
48
49
class FlagDefinition(BaseModel):
    """Definition of a single feature flag."""

    name: str
    type: FlagType = FlagType.BOOLEAN
    default_value: Any = False
    description: str = ""
    rules: list[FlagRule] = Field(default_factory=list)
    overrides: dict[str, Any] = Field(default_factory=dict)

FlagEvaluationError

Bases: FeatureFlagError

Raised when flag evaluation encounters an error.

Source code in src/forge/featureflags/exceptions.py 
18
19
class FlagEvaluationError(FeatureFlagError):
    """Raised when flag evaluation encounters an error."""

FlagEvaluator

Evaluates feature flags against an evaluation context.

Source code in src/forge/featureflags/evaluator.py 
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
class FlagEvaluator:
    """Evaluates feature flags against an evaluation context."""

    def __init__(self, store: FlagStore) -> None:
        self._store = store

    @property
    def store(self) -> FlagStore:
        return self._store

    async def evaluate(
        self,
        flag_name: str,
        context: EvaluationContext,
    ) -> EvaluationResult:
        """Evaluate a single feature flag for the given context."""
        flag = await self._store.get_flag(flag_name)
        if flag is None:
            raise FlagNotFoundError(f"Flag '{flag_name}' not found.")

        return self._evaluate_flag(flag, context)

    async def evaluate_bulk(
        self,
        flag_names: list[str],
        context: EvaluationContext,
    ) -> dict[str, EvaluationResult]:
        """Evaluate multiple feature flags for the given context."""
        results: dict[str, EvaluationResult] = {}
        for name in flag_names:
            with contextlib.suppress(FlagNotFoundError):
                results[name] = await self.evaluate(name, context)
        return results

    async def evaluate_all(
        self,
        context: EvaluationContext,
    ) -> dict[str, EvaluationResult]:
        """Evaluate all stored feature flags for the given context."""
        flags = await self._store.list_flags()
        results: dict[str, EvaluationResult] = {}
        for flag in flags:
            try:
                results[flag.name] = await self.evaluate(flag.name, context)
            except FlagNotFoundError:
                continue
            except Exception as exc:
                logger.warning("Failed to evaluate flag '%s': %s", flag.name, exc)
        return results

    def _evaluate_flag(
        self,
        flag: FlagDefinition,
        context: EvaluationContext,
    ) -> EvaluationResult:
        """Evaluate a flag definition against context."""
        # 1. Check overrides (highest priority)
        override_reason = self._check_overrides(flag, context)
        if override_reason is not None:
            return override_reason

        # 2. Check rules (e.g., percentage rollout, segment match)
        for idx, rule in enumerate(flag.rules):
            matched = self._evaluate_rule(flag, rule, context)
            if matched:
                return EvaluationResult(
                    flag_name=flag.name,
                    value=rule.value,
                    reason=self._reason_for_rule(flag.type, rule),
                    matched_rule_index=idx,
                )

        # 3. Fall back to default
        return EvaluationResult(
            flag_name=flag.name,
            value=flag.default_value,
            reason=EvaluationReason.DEFAULT,
        )

    def _check_overrides(
        self,
        flag: FlagDefinition,
        context: EvaluationContext,
    ) -> EvaluationResult | None:
        """Check if the user has a direct override."""
        if not flag.overrides:
            return None

        # Check user_id override
        if context.user_id in flag.overrides:
            return EvaluationResult(
                flag_name=flag.name,
                value=flag.overrides[context.user_id],
                reason=EvaluationReason.OVERRIDE,
            )

        # Check region override
        if context.region and context.region in flag.overrides:
            return EvaluationResult(
                flag_name=flag.name,
                value=flag.overrides[context.region],
                reason=EvaluationReason.OVERRIDE,
            )

        return None

    def _evaluate_rule(
        self,
        flag: FlagDefinition,
        rule: FlagRule,
        context: EvaluationContext,
    ) -> bool:
        """Evaluate a single rule against the context."""
        if flag.type == FlagType.PERCENTAGE:
            percentage = rule.percentage
            if percentage is not None:
                hash_input = f"{flag.name}:{context.user_id}"
                hash_val = _consistent_hash(hash_input)
                return hash_val < percentage

        if flag.type == FlagType.SEGMENT:
            segments = list(rule.segments or [])
            return _check_segments_match(context, segments)

        # Boolean or fallback: check segment conditions if present
        segments = list(rule.segments or [])
        if segments:
            return _check_segments_match(context, segments)

        # No segments — match all for boolean rules
        return True

    def _reason_for_rule(
        self,
        flag_type: FlagType,
        rule: FlagRule,
    ) -> EvaluationReason:
        """Determine the evaluation reason based on flag type and rule."""
        if flag_type == FlagType.PERCENTAGE:
            return EvaluationReason.PERCENTAGE_ROLLOUT
        if flag_type == FlagType.SEGMENT:
            return EvaluationReason.SEGMENT_MATCH
        return EvaluationReason.RULE_MATCH
Methods:
evaluate async 
evaluate(flag_name: str, context: EvaluationContext) -> EvaluationResult

Evaluate a single feature flag for the given context.

Source code in src/forge/featureflags/evaluator.py 
85
86
87
88
89
90
91
92
93
94
95
async def evaluate(
    self,
    flag_name: str,
    context: EvaluationContext,
) -> EvaluationResult:
    """Evaluate a single feature flag for the given context."""
    flag = await self._store.get_flag(flag_name)
    if flag is None:
        raise FlagNotFoundError(f"Flag '{flag_name}' not found.")

    return self._evaluate_flag(flag, context)
evaluate_all async 
evaluate_all(context: EvaluationContext) -> dict[str, EvaluationResult]

Evaluate all stored feature flags for the given context.

Source code in src/forge/featureflags/evaluator.py 
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
async def evaluate_all(
    self,
    context: EvaluationContext,
) -> dict[str, EvaluationResult]:
    """Evaluate all stored feature flags for the given context."""
    flags = await self._store.list_flags()
    results: dict[str, EvaluationResult] = {}
    for flag in flags:
        try:
            results[flag.name] = await self.evaluate(flag.name, context)
        except FlagNotFoundError:
            continue
        except Exception as exc:
            logger.warning("Failed to evaluate flag '%s': %s", flag.name, exc)
    return results
evaluate_bulk async 
evaluate_bulk(flag_names: list[str], context: EvaluationContext) -> dict[str, EvaluationResult]

Evaluate multiple feature flags for the given context.

Source code in src/forge/featureflags/evaluator.py 
 97
 98
 99
100
101
102
103
104
105
106
107
async def evaluate_bulk(
    self,
    flag_names: list[str],
    context: EvaluationContext,
) -> dict[str, EvaluationResult]:
    """Evaluate multiple feature flags for the given context."""
    results: dict[str, EvaluationResult] = {}
    for name in flag_names:
        with contextlib.suppress(FlagNotFoundError):
            results[name] = await self.evaluate(name, context)
    return results

FlagNotFoundError

Bases: FeatureFlagError

Raised when a requested flag is not found in the store.

Source code in src/forge/featureflags/exceptions.py 
10
11
class FlagNotFoundError(FeatureFlagError):
    """Raised when a requested flag is not found in the store."""

FlagRule

Bases: BaseModel

An override rule for a feature flag.

Source code in src/forge/featureflags/models.py 
33
34
35
36
37
38
class FlagRule(BaseModel):
    """An override rule for a feature flag."""

    value: Any
    segments: list[SegmentRule] = Field(default_factory=list)
    percentage: int | None = Field(default=None, ge=0, le=100)

FlagStoreError

Bases: FeatureFlagError

Raised when a flag store operation fails.

Source code in src/forge/featureflags/exceptions.py 
14
15
class FlagStoreError(FeatureFlagError):
    """Raised when a flag store operation fails."""

FlagType

Bases: StrEnum

Supported feature flag evaluation types.

Source code in src/forge/featureflags/models.py 
 9
10
11
12
13
14
class FlagType(enum.StrEnum):
    """Supported feature flag evaluation types."""

    BOOLEAN = "boolean"
    PERCENTAGE = "percentage"
    SEGMENT = "segment"

InvalidFlagDefinitionError

Bases: FeatureFlagError

Raised when a flag definition is invalid or malformed.

Source code in src/forge/featureflags/exceptions.py 
22
23
class InvalidFlagDefinitionError(FeatureFlagError):
    """Raised when a flag definition is invalid or malformed."""

MemoryFlagStore

In-memory flag store using a dict.

Source code in src/forge/featureflags/store.py 
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
class MemoryFlagStore:
    """In-memory flag store using a dict."""

    def __init__(self) -> None:
        self._flags: dict[str, FlagDefinition] = {}

    async def get_flag(self, name: str) -> FlagDefinition | None:
        return self._flags.get(name)

    async def set_flag(self, flag: FlagDefinition) -> None:
        self._flags[flag.name] = flag

    async def delete_flag(self, name: str) -> bool:
        if name in self._flags:
            del self._flags[name]
            return True
        return False

    async def list_flags(self) -> list[FlagDefinition]:
        return list(self._flags.values())

    async def close(self) -> None:
        self._flags.clear()

RedisFlagStore

Redis-backed flag store using redis.asyncio.

Source code in src/forge/featureflags/store.py 
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
class RedisFlagStore:
    """Redis-backed flag store using redis.asyncio."""

    def __init__(
        self,
        redis_url: str,
        key_prefix: str = "forge:featureflags:",
        max_connections: int = 10,
    ) -> None:
        self._url = redis_url
        self._prefix = key_prefix
        self._max_connections = max_connections
        self._client: Any = None
        self._pool: Any = None

    @property
    def url(self) -> str:
        return self._url

    @property
    def is_connected(self) -> bool:
        return self._client is not None

    async def connect(self) -> None:
        import redis.asyncio as aioredis

        try:
            self._pool = aioredis.ConnectionPool.from_url(
                self._url,
                max_connections=self._max_connections,
            )
            self._client = aioredis.Redis(connection_pool=self._pool)
            await self._client.ping()
        except Exception as exc:
            logger.warning("Failed to connect to Redis at %s: %s", self._url, exc)
            self._client = None
            self._pool = None
            raise FlagStoreError(f"Failed to connect to Redis: {exc}") from exc

    async def close(self) -> None:
        if self._client:
            try:
                await self._client.aclose()
            except Exception as exc:
                logger.warning("Error closing Redis client: %s", exc)
        if self._pool:
            try:
                await self._pool.disconnect()
            except Exception as exc:
                logger.warning("Error disconnecting Redis pool: %s", exc)
        self._client = None
        self._pool = None

    def _flag_key(self, name: str) -> str:
        return f"{self._prefix}{name}"

    def _flags_set_key(self) -> str:
        return f"{self._prefix}__flags__"

    def _serialize(self, flag: FlagDefinition) -> str:
        return flag.model_dump_json()

    def _deserialize(self, raw: str) -> FlagDefinition:
        data = json.loads(raw)
        return FlagDefinition.model_validate(data)

    async def get_flag(self, name: str) -> FlagDefinition | None:
        if not self._client:
            raise FlagStoreError("Redis store is not connected.")
        try:
            raw = await self._client.get(self._flag_key(name))
            if raw is None:
                return None
            return self._deserialize(raw.decode("utf-8"))
        except Exception as exc:
            logger.warning("Redis GET failed for flag %s: %s", name, exc)
            raise FlagStoreError(f"Redis GET failed: {exc}") from exc

    async def set_flag(self, flag: FlagDefinition) -> None:
        if not self._client:
            raise FlagStoreError("Redis store is not connected.")
        try:
            serialized = self._serialize(flag)
            await self._client.set(self._flag_key(flag.name), serialized)
            await self._client.sadd(self._flags_set_key(), flag.name)
        except Exception as exc:
            logger.warning("Redis SET failed for flag %s: %s", flag.name, exc)
            raise FlagStoreError(f"Redis SET failed: {exc}") from exc

    async def delete_flag(self, name: str) -> bool:
        if not self._client:
            raise FlagStoreError("Redis store is not connected.")
        try:
            deleted = await self._client.delete(self._flag_key(name))
            await self._client.srem(self._flags_set_key(), name)
            return bool(deleted and deleted > 0)
        except Exception as exc:
            logger.warning("Redis DELETE failed for flag %s: %s", name, exc)
            raise FlagStoreError(f"Redis DELETE failed: {exc}") from exc

    async def list_flags(self) -> list[FlagDefinition]:
        if not self._client:
            raise FlagStoreError("Redis store is not connected.")
        try:
            flag_names = await self._client.smembers(self._flags_set_key())
            if not flag_names:
                return []
            flags: list[FlagDefinition] = []
            for name_bytes in flag_names:
                name = name_bytes.decode("utf-8")
                raw = await self._client.get(self._flag_key(name))
                if raw is not None:
                    flags.append(self._deserialize(raw.decode("utf-8")))
            return flags
        except Exception as exc:
            logger.warning("Redis LIST failed: %s", exc)
            raise FlagStoreError(f"Redis LIST failed: {exc}") from exc

SegmentRule

Bases: BaseModel

A rule defining a user segment for flag targeting.

Source code in src/forge/featureflags/models.py 
25
26
27
28
29
30
class SegmentRule(BaseModel):
    """A rule defining a user segment for flag targeting."""

    attribute: str
    operator: str = "eq"
    values: list[str] = Field(default_factory=list)