mirror of
https://github.com/hanzoai/python-sdk.git
synced 2026-10-23 20:23:00 +00:00
zeekay
6ef78a34c9
Flags were natively supported in Rust (the hanzo-flags crate — the evaluation core, pub-exported for Rust callers and FFI-exported for the rest), Go (in-process via CGO to that core), and TypeScript (@hanzo/flags over /v1/flags). Python was the gap. This closes it. hanzo-flags POSTs an evaluation context to cloud /v1/flags — the same PostHog-compatible endpoint every other client speaks — and answers is_enabled / variant / payload from the response. It mirrors @hanzo/flags exactly: a HanzoFlags client bound to one host, load(distinct_id, person_properties=, groups=), the same three-field result, an async twin, and a one-shot evaluate(). Two invariants match the family: fail-open (a transport error yields the last good or empty result with errors_while_computing set, never a raise on the hot path) and cache-by-context+TTL. Zero runtime dependencies — stdlib urllib only, so a flag check never drags httpx/pydantic into a service. Tests stand up a local /v1/flags stub and prove the request shape is PostHog-compatible, the accessors resolve booleans/variants/payloads, the TTL cache holds, and both a 5xx and an unreachable host fail open. 7/7 green.
85 lines
3.0 KiB
Python
85 lines
3.0 KiB
Python
# Copyright 2026 Hanzo AI, Inc. All rights reserved.
|
|
"""Value types for the Hanzo flags client — the same shapes @hanzo/flags carries.
|
|
|
|
A flag value is ``True`` (a boolean flag that is on), a non-empty ``str`` (the
|
|
active variant of a multivariate flag), or absent (off). The result also carries
|
|
per-flag JSON payloads and a soft error signal, matching the PostHog ``/decide``
|
|
response the cloud ``/v1/flags`` endpoint returns.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass, field
|
|
from typing import Any, Dict, Optional
|
|
|
|
# A flag evaluates to a bool (on/off) or a variant name.
|
|
FlagValue = Any # bool | str
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class Group:
|
|
"""One group the caller belongs to (org, team, …) for group-targeted flags."""
|
|
|
|
key: str
|
|
properties: Optional[Dict[str, Any]] = None
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class EvalContext:
|
|
"""Who is being evaluated. ``distinct_id`` is required; the rest refine targeting.
|
|
|
|
Mirrors @hanzo/flags EvalContext: distinctId / personProperties / groups.
|
|
"""
|
|
|
|
distinct_id: str
|
|
person_properties: Optional[Dict[str, Any]] = None
|
|
groups: Optional[Dict[str, Group]] = None
|
|
|
|
def wire(self) -> Dict[str, Any]:
|
|
"""The request body the /v1/flags endpoint expects (PostHog-shaped)."""
|
|
body: Dict[str, Any] = {"distinct_id": self.distinct_id}
|
|
if self.person_properties:
|
|
body["person_properties"] = self.person_properties
|
|
if self.groups:
|
|
body["groups"] = {
|
|
k: {"key": g.key, **({"properties": g.properties} if g.properties else {})}
|
|
for k, g in self.groups.items()
|
|
}
|
|
return body
|
|
|
|
|
|
@dataclass
|
|
class EvalResult:
|
|
"""A resolved set of flags — the same three fields @hanzo/flags returns.
|
|
|
|
``errors_while_computing`` is a SOFT signal: the client is fail-open, so a
|
|
transport error yields the last good (or empty) result with this set True,
|
|
never an exception on the hot path.
|
|
"""
|
|
|
|
feature_flags: Dict[str, FlagValue] = field(default_factory=dict)
|
|
feature_flag_payloads: Dict[str, Any] = field(default_factory=dict)
|
|
errors_while_computing: bool = False
|
|
|
|
@classmethod
|
|
def from_wire(cls, body: Dict[str, Any]) -> "EvalResult":
|
|
return cls(
|
|
feature_flags=body.get("featureFlags") or {},
|
|
feature_flag_payloads=body.get("featureFlagPayloads") or {},
|
|
errors_while_computing=bool(body.get("errorsWhileComputingFlags")),
|
|
)
|
|
|
|
def is_enabled(self, key: str) -> bool:
|
|
"""True when the flag is on: boolean ``True`` OR any non-empty variant."""
|
|
v = self.feature_flags.get(key)
|
|
return v is True or (isinstance(v, str) and v != "")
|
|
|
|
def variant(self, key: str) -> Optional[str]:
|
|
"""The active variant of a multivariate flag, or ``None``."""
|
|
v = self.feature_flags.get(key)
|
|
return v if isinstance(v, str) else None
|
|
|
|
def payload(self, key: str) -> Any:
|
|
"""The flag's JSON payload (per-variant or the ``true`` payload), or ``None``."""
|
|
return self.feature_flag_payloads.get(key)
|