hanzoai/python-sdk
2
0
Fork 0
mirror of https://github.com/hanzoai/python-sdk.git synced 2026-10-23 20:23:00 +00:00
Code Issues Packages Projects Releases Wiki Activity

hanzo-flags: the Python client for the native flags engine

Browse Source 
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.
This commit is contained in:
zeekay
2026-07-17 13:00:33 -07:00
parent 44edb35d63
commit 81dce282a5
10 changed files with 576 additions and 26 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/hanzo-flags/README.md
+80
View File
@@ -0,0 +1,80 @@
# hanzo-flags
Feature flags + A/B testing for Python, on the Hanzo native flags engine.
The Python client for cloud `/v1/flags` — the same PostHog-compatible evaluation
endpoint the Rust core, Go binding, and `@hanzo/flags` (TypeScript) all speak. One
flag definition, evaluated identically in every language.
**Zero runtime dependencies** — stdlib `urllib` only. A flag check must never drag
a dependency tree into a service.
## Install
```bash
pip install hanzo-flags # or: uv add hanzo-flags
```
## Use
```python
from hanzo_flags import HanzoFlags
flags = HanzoFlags("https://api.hanzo.ai", token="hk-...")
flags.load("user-123", person_properties={"plan": "pro"})
if flags.is_enabled("checkout-exp"):
...
variant = flags.variant("pricing-test") # "control" | "b" | None
payload = flags.payload("pricing-test") # the flag's JSON payload, or None
```
Group targeting (orgs/teams):
```python
from hanzo_flags import Group
flags.load(
"user-123",
groups={"0": Group(key="acme", properties={"tier": "gold"})},
)
```
Async (asyncio services):
```python
from hanzo_flags import AsyncHanzoFlags
flags = AsyncHanzoFlags("https://api.hanzo.ai", token="hk-...")
await flags.load("user-123")
if flags.is_enabled("checkout-exp"):
...
```
One-shot:
```python
from hanzo_flags import evaluate
res = evaluate("https://api.hanzo.ai", "user-123", token="hk-...")
res.is_enabled("checkout-exp")
```
## Guarantees
- **Fail-open.** A transport or decode error returns the last good (or empty)
result with `errors_while_computing` set — `load()` never raises on the hot path.
- **Cached by context + TTL** (default 15s). Re-evaluating the same context inside
the TTL is free, so a hot path may call `load()` freely.
## The family
| Language | Package | Mode |
|------------|-----------------------|----------------------------------------|
| Rust | `hanzo-flags` crate | native (the evaluation core) |
| Go | `cloud/clients/flags` | in-process via FFI to the Rust core |
| TypeScript | `@hanzo/flags` | HTTP to `/v1/flags` (browser + node) |
| Python | `hanzo-flags` | HTTP to `/v1/flags` (this package) |
All four resolve the same flag to the same value: the definitions live once in the
cloud flags cockpit (`/v1/flags/defs`), and every client evaluates against them.
pkg/hanzo-flags/hanzo_flags/__init__.py
+29
View File
@@ -0,0 +1,29 @@
# Copyright 2026 Hanzo AI, Inc. All rights reserved.
"""hanzo-flags — feature flags + A/B for the Hanzo native flags engine.
The Python client for cloud ``/v1/flags`` (the PostHog-compatible evaluation
endpoint the Rust core, Go, and @hanzo/flags all speak). Zero runtime deps.
from hanzo_flags import HanzoFlags
flags = HanzoFlags("https://api.hanzo.ai", token=tok)
flags.load("user-123", person_properties={"plan": "pro"})
if flags.is_enabled("checkout-exp"):
...
variant = flags.variant("pricing-test") # "control" | "b" | None
"""
from .async_client import AsyncHanzoFlags
from .client import HanzoFlags, evaluate
from .models import EvalContext, EvalResult, Group
__all__ = [
"HanzoFlags",
"AsyncHanzoFlags",
"evaluate",
"EvalContext",
"EvalResult",
"Group",
]
__version__ = "0.1.0"
pkg/hanzo-flags/hanzo_flags/async_client.py
+60
View File
@@ -0,0 +1,60 @@
# Copyright 2026 Hanzo AI, Inc. All rights reserved.
"""Async facade for :class:`hanzo_flags.HanzoFlags`.
Zero-dependency: the sync client's blocking urllib call runs in the default
executor, so an asyncio service gets ``await flags.load(...)`` without pulling an
async HTTP library in. The caching, fail-open, and accessors are the sync
client's — this only moves the one blocking call off the event loop.
"""
from __future__ import annotations
import asyncio
from typing import Any, Dict, Optional
from .client import HanzoFlags
from .models import EvalResult, Group
class AsyncHanzoFlags:
"""The async twin of :class:`HanzoFlags`; same constructor, awaitable ``load``."""
def __init__(
self,
host: str,
*,
token: Optional[str] = None,
project: Optional[str] = None,
ttl_ms: int = 15_000,
timeout_s: float = 3.0,
) -> None:
self._c = HanzoFlags(
host, token=token, project=project, ttl_ms=ttl_ms, timeout_s=timeout_s
)
async def load(
self,
distinct_id: str,
*,
person_properties: Optional[Dict[str, Any]] = None,
groups: Optional[Dict[str, Group]] = None,
) -> EvalResult:
loop = asyncio.get_running_loop()
return await loop.run_in_executor(
None,
lambda: self._c.load(
distinct_id, person_properties=person_properties, groups=groups
),
)
def is_enabled(self, key: str) -> bool:
return self._c.is_enabled(key)
def variant(self, key: str) -> Optional[str]:
return self._c.variant(key)
def payload(self, key: str) -> Any:
return self._c.payload(key)
def all(self) -> EvalResult:
return self._c.all()
pkg/hanzo-flags/hanzo_flags/client.py
+143
View File
@@ -0,0 +1,143 @@
# Copyright 2026 Hanzo AI, Inc. All rights reserved.
"""Zero-dependency Python client for the Hanzo native flags engine.
It POSTs an evaluation context to ``<host>/v1/flags`` (the same PostHog-compatible
endpoint @hanzo/flags and the Go/Rust cores speak) and answers ``is_enabled`` /
``variant`` / ``payload`` from the response.
Two invariants, identical to @hanzo/flags:
* **Fail-open.** A flag client must never take a request down. Any transport or
decode error returns the last good result (or an empty one), with
``errors_while_computing`` set — never a raised exception on ``load``.
* **Cached by context + TTL.** Re-evaluating the same context inside the TTL
returns the cached result, so a hot path can call ``load`` freely.
Stdlib only (``urllib``) — a flag check should not drag httpx/pydantic into a
service and risk a version conflict.
"""
from __future__ import annotations
import json
import time
import urllib.error
import urllib.request
from typing import Any, Dict, Optional
from .models import EvalContext, EvalResult, Group
_DEFAULT_TTL_MS = 15_000
_DEFAULT_TIMEOUT_S = 3.0
class HanzoFlags:
"""A flags client bound to one cloud host.
Args:
host: the cloud base URL, e.g. ``https://api.hanzo.ai``. Required.
token: a bearer token; when absent the request relies on ambient
(cookie/gateway) auth, exactly like the browser client.
project: an optional ``X-Project-Id`` scope.
ttl_ms: cache lifetime for a given context (default 15s).
timeout_s: per-request timeout (default 3s).
"""
def __init__(
self,
host: str,
*,
token: Optional[str] = None,
project: Optional[str] = None,
ttl_ms: int = _DEFAULT_TTL_MS,
timeout_s: float = _DEFAULT_TIMEOUT_S,
) -> None:
if not host:
raise ValueError("hanzo-flags: host is required")
self._host = host.rstrip("/")
self._token = token
self._project = project
self._ttl_ms = ttl_ms
self._timeout_s = timeout_s
self._result = EvalResult()
self._ctx_key = ""
self._loaded_at = 0.0
def load(
self,
distinct_id: str,
*,
person_properties: Optional[Dict[str, Any]] = None,
groups: Optional[Dict[str, Group]] = None,
) -> EvalResult:
"""Evaluate flags for ``distinct_id`` and cache the result.
Returns the same :class:`EvalResult` the convenience accessors read; call
:meth:`is_enabled` / :meth:`variant` / :meth:`payload` after, or read the
returned result directly.
"""
ctx = EvalContext(distinct_id, person_properties=person_properties, groups=groups)
key = json.dumps(ctx.wire(), sort_keys=True)
if self._ctx_key == key and (time.monotonic() - self._loaded_at) * 1000 < self._ttl_ms:
return self._result
try:
body = self._post(ctx.wire())
self._result = EvalResult.from_wire(body)
except Exception:
# Fail-open: keep the last good result, flag the soft error.
self._result.errors_while_computing = True
finally:
self._ctx_key = key
self._loaded_at = time.monotonic()
return self._result
def _post(self, wire: Dict[str, Any]) -> Dict[str, Any]:
headers = {"Content-Type": "application/json"}
if self._token:
headers["Authorization"] = f"Bearer {self._token}"
if self._project:
headers["X-Project-Id"] = self._project
req = urllib.request.Request(
f"{self._host}/v1/flags",
data=json.dumps(wire).encode("utf-8"),
headers=headers,
method="POST",
)
with urllib.request.urlopen(req, timeout=self._timeout_s) as resp:
if resp.status >= 400:
raise urllib.error.HTTPError(req.full_url, resp.status, "flags", resp.headers, None)
return json.loads(resp.read().decode("utf-8"))
# ---- accessors read the last loaded result (call load() first) ----
def is_enabled(self, key: str) -> bool:
"""True when ``key`` is on (boolean True or any non-empty variant)."""
return self._result.is_enabled(key)
def variant(self, key: str) -> Optional[str]:
"""The active variant of a multivariate flag, or ``None``."""
return self._result.variant(key)
def payload(self, key: str) -> Any:
"""The flag's JSON payload, or ``None``."""
return self._result.payload(key)
def all(self) -> EvalResult:
"""The last loaded result."""
return self._result
def evaluate(
host: str,
distinct_id: str,
*,
token: Optional[str] = None,
project: Optional[str] = None,
person_properties: Optional[Dict[str, Any]] = None,
groups: Optional[Dict[str, Group]] = None,
timeout_s: float = _DEFAULT_TIMEOUT_S,
) -> EvalResult:
"""One-shot evaluation — the module-level convenience, like @hanzo/flags evaluateFlags."""
client = HanzoFlags(host, token=token, project=project, timeout_s=timeout_s)
return client.load(distinct_id, person_properties=person_properties, groups=groups)
pkg/hanzo-flags/hanzo_flags/models.py
+84
View File
@@ -0,0 +1,84 @@
# 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)
pkg/hanzo-flags/hanzo_flags/py.typed
View File
pkg/hanzo-flags/pyproject.toml
+53
View File
@@ -0,0 +1,53 @@
[project]
name = "hanzo-flags"
version = "0.1.0"
description = "Hanzo feature flags + A/B client — the Python side of the native flags engine (cloud /v1/flags), zero runtime deps, PostHog-compatible"
readme = "README.md"
license = { text = "MIT" }
requires-python = ">=3.9"
authors = [{ name = "Hanzo AI", email = "[email protected]" }]
keywords = ["feature-flags", "ab-testing", "posthog", "hanzo", "experiments"]
classifiers = [
"Development Status :: 4 - Beta",
"Intended Audience :: Developers",
"License :: OSI Approved :: MIT License",
"Operating System :: OS Independent",
"Programming Language :: Python :: 3.9",
"Programming Language :: Python :: 3.10",
"Programming Language :: Python :: 3.11",
"Programming Language :: Python :: 3.12",
"Programming Language :: Python :: 3.13",
"Topic :: Software Development :: Libraries :: Python Modules",
"Typing :: Typed",
]
# Zero runtime dependencies — a flag check must never drag a dependency tree into
# a service. The client is stdlib urllib only.
dependencies = []
[project.optional-dependencies]
dev = ["pytest>=7.0.0", "ruff>=0.1.0", "mypy>=1.0.0"]
[project.urls]
Homepage = "https://hanzo.ai"
Documentation = "https://docs.hanzo.ai"
Repository = "https://github.com/hanzoai/python-sdk"
[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"
[tool.hatch.build.targets.wheel]
packages = ["hanzo_flags"]
[tool.ruff]
target-version = "py39"
line-length = 100
[tool.ruff.lint]
select = ["E", "W", "F", "I", "B", "C4", "UP"]
ignore = ["E501"]
[tool.mypy]
python_version = "3.9"
strict = true
pkg/hanzo-flags/tests/test_client.py
+125
View File
@@ -0,0 +1,125 @@
# Copyright 2026 Hanzo AI, Inc. All rights reserved.
"""hanzo-flags client tests — no network: a local stub HTTP server stands in for
cloud /v1/flags, so the request shape, response decode, caching, fail-open, and
accessors are all exercised deterministically.
"""
from __future__ import annotations
import json
import threading
from http.server import BaseHTTPRequestHandler, HTTPServer
import pytest
from hanzo_flags import EvalResult, Group, HanzoFlags, evaluate
# ---- a tiny stub of the /v1/flags endpoint ---------------------------------
_RESPONSE = {
"featureFlags": {"checkout-exp": True, "pricing-test": "variant-b", "off-flag": False},
"featureFlagPayloads": {"pricing-test": {"price": 9}},
"errorsWhileComputingFlags": False,
}
class _Handler(BaseHTTPRequestHandler):
last_body = None
last_headers = None
status = 200
def do_POST(self): # noqa: N802
n = int(self.headers.get("Content-Length", 0))
_Handler.last_body = json.loads(self.rfile.read(n) or b"{}")
_Handler.last_headers = dict(self.headers)
self.send_response(_Handler.status)
self.send_header("Content-Type", "application/json")
self.end_headers()
self.wfile.write(json.dumps(_RESPONSE).encode())
def log_message(self, *_): # silence
pass
@pytest.fixture()
def server():
srv = HTTPServer(("127.0.0.1", 0), _Handler)
t = threading.Thread(target=srv.serve_forever, daemon=True)
t.start()
_Handler.status = 200
yield f"http://127.0.0.1:{srv.server_address[1]}"
srv.shutdown()
# ---- the deliverable: login/usage accessors resolve correctly --------------
def test_is_enabled_variant_payload(server):
flags = HanzoFlags(server, token="tok")
flags.load("user-123", person_properties={"plan": "pro"})
assert flags.is_enabled("checkout-exp") is True
# A variant flag is "enabled" AND exposes its variant + payload.
assert flags.is_enabled("pricing-test") is True
assert flags.variant("pricing-test") == "variant-b"
assert flags.payload("pricing-test") == {"price": 9}
# A false flag is off; an unknown flag is off and yields no variant/payload.
assert flags.is_enabled("off-flag") is False
assert flags.is_enabled("nope") is False
assert flags.variant("nope") is None
assert flags.payload("nope") is None
def test_request_shape_is_posthog_compatible(server):
flags = HanzoFlags(server, token="tok", project="proj-1")
flags.load(
"u1",
person_properties={"plan": "pro"},
groups={"0": Group(key="acme", properties={"tier": "gold"})},
)
body = _Handler.last_body
assert body["distinct_id"] == "u1"
assert body["person_properties"] == {"plan": "pro"}
assert body["groups"] == {"0": {"key": "acme", "properties": {"tier": "gold"}}}
# Auth + project scoping ride as headers.
assert _Handler.last_headers.get("Authorization") == "Bearer tok"
assert _Handler.last_headers.get("X-Project-Id") == "proj-1"
def test_caches_within_ttl(server):
flags = HanzoFlags(server, ttl_ms=60_000)
flags.load("u1")
_Handler.last_body = None # a second load with the same ctx must NOT hit the server
flags.load("u1")
assert _Handler.last_body is None
# A different context DOES re-evaluate.
flags.load("u2")
assert _Handler.last_body is not None
def test_fail_open_on_server_error(server):
_Handler.status = 500
flags = HanzoFlags(server)
res = flags.load("u1") # must NOT raise
assert isinstance(res, EvalResult)
assert res.errors_while_computing is True
assert flags.is_enabled("anything") is False # empty, fail-open
def test_fail_open_on_unreachable_host():
# Nothing is listening — load must fail open, never raise.
flags = HanzoFlags("http://127.0.0.1:9", timeout_s=0.2)
res = flags.load("u1")
assert res.errors_while_computing is True
assert flags.is_enabled("x") is False
def test_module_evaluate_oneshot(server):
res = evaluate(server, "u1", token="tok")
assert res.is_enabled("checkout-exp") is True
assert res.variant("pricing-test") == "variant-b"
def test_host_required():
with pytest.raises(ValueError):
HanzoFlags("")
pyproject.toml
+1
View File
@@ -71,6 +71,7 @@ members = [
"pkg/hanzo-memory",
"pkg/hanzo-network",
"pkg/hanzo-dev-py",
"pkg/hanzo-flags",
"pkg/hanzo-s3",
"pkg/hanzo-train",
"pkg/hanzo-tools-ui",
uv.lock
Generated
+1 -26
View File
@@ -28,7 +28,6 @@ members = [
"hanzo-tools-shell",
"hanzo-tools-todo",
"hanzo-tools-ui",
"hanzo-train",
"hanzoai",
]
@@ -1545,7 +1544,7 @@ provides-extras = ["dev"]
[[package]]
name = "hanzo-tools"
version = "0.3.2"
version = "0.3.3"
source = { editable = "pkg/hanzo-tools" }
dependencies = [
{ name = "aiofiles" },
@@ -2211,30 +2210,6 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/b3/ee/a6d265860d51a52a8ac931579f722a15173ca480bbcce233fd78505b491b/hanzo_tools_vector-0.2.0-py3-none-any.whl", hash = "sha256:2c5dca66e657bf1098d9ac4f8aff625addf4cb623387a1d2cb234a935ab4e7fc", size = 32480, upload-time = "2025-12-26T18:29:58.591Z" },
]
[[package]]
name = "hanzo-train"
version = "0.1.0"
source = { editable = "pkg/hanzo-train" }
dependencies = [
{ name = "httpx" },
]
[package.optional-dependencies]
dev = [
{ name = "mypy" },
{ name = "pytest" },
{ name = "ruff" },
]
[package.metadata]
requires-dist = [
{ name = "httpx", specifier = ">=0.25.0" },
{ name = "mypy", marker = "extra == 'dev'", specifier = ">=1.10.0" },
{ name = "pytest", marker = "extra == 'dev'", specifier = ">=8.0.0" },
{ name = "ruff", marker = "extra == 'dev'", specifier = ">=0.5.0" },
]
provides-extras = ["dev"]
[[package]]
name = "hanzo-zap"
version = "0.7.0"