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.
61 lines
1.7 KiB
Python
61 lines
1.7 KiB
Python
# 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()
|