""" feature-control-sdk Thin organization wrapper around OpenFeature SDK. See docs/canon-mapping.md for how context projects from ITC-ORG/ACCESS/LAND facts. """ from __future__ import annotations from typing import Any, Dict, Optional try: from openfeature import OpenFeatureAPI from openfeature.evaluation_context import EvaluationContext from openfeature.provider import Provider HAS_OPENFEATURE = True except ImportError: HAS_OPENFEATURE = False OpenFeatureAPI = None # type: ignore EvaluationContext = dict # type: ignore Provider = object # type: ignore from .providers.local import LocalProvider from .registry import FeatureDefinition, FeatureRegistry from .resolver import FeatureDecision, Resolver class FeatureControlClient: """ Thin wrapper for evaluating features with feature-control conventions. Usage: client = FeatureControlClient() client.set_provider(LocalProvider({"my.feature": True})) value = client.get_boolean_value("my.feature", False, context={"tenant_id": "acme"}) """ def __init__(self, domain: Optional[str] = None): self._api = OpenFeatureAPI.get_instance() if HAS_OPENFEATURE else None self._domain = domain if HAS_OPENFEATURE: self._client = self._api.get_client(domain) if domain else self._api.get_client() else: self._client = None # local mode, will use resolver or direct self._resolver: Optional[Resolver] = None self._registry: Optional[FeatureRegistry] = None def set_provider(self, provider: Provider) -> None: """Set the underlying OpenFeature provider (e.g. LocalProvider for tests).""" if HAS_OPENFEATURE: if self._domain: self._api.set_provider(self._domain, provider) else: self._api.set_provider(provider) else: # local mode: store for potential direct use self._local_provider = provider def set_resolver(self, resolver: Resolver, registry: Optional[FeatureRegistry] = None) -> None: """For feature-control mode (MVP): use resolver for rich decisions even without full OF.""" self._resolver = resolver self._registry = registry or resolver.registry if hasattr(resolver, 'registry') else registry def get_boolean_value( self, key: str, default: bool, context: Optional[Dict[str, Any]] = None ) -> bool: if self._resolver is not None: decision = self._resolver.evaluate(key, default, context) return bool(decision.value) if HAS_OPENFEATURE and self._client: of_context = self._build_context(context) return self._client.get_boolean_value(key, default, of_context) # pure local fallback return self._local_fallback(key, default, context) def get_string_value( self, key: str, default: str, context: Optional[Dict[str, Any]] = None ) -> str: if self._resolver is not None: decision = self._resolver.evaluate(key, default, context) return str(decision.value) if HAS_OPENFEATURE and self._client: of_context = self._build_context(context) return self._client.get_string_value(key, default, of_context) return str(self._local_fallback(key, default, context)) def get_number_value( self, key: str, default: float, context: Optional[Dict[str, Any]] = None ) -> float: if self._resolver is not None: decision = self._resolver.evaluate(key, default, context) return float(decision.value) if HAS_OPENFEATURE and self._client: of_context = self._build_context(context) return self._client.get_number_value(key, default, of_context) return float(self._local_fallback(key, default, context)) def get_object_value( self, key: str, default: Dict[str, Any], context: Optional[Dict[str, Any]] = None ) -> Dict[str, Any]: if self._resolver is not None: decision = self._resolver.evaluate(key, default, context) return dict(decision.value) if isinstance(decision.value, dict) else default if HAS_OPENFEATURE and self._client: of_context = self._build_context(context) return self._client.get_object_value(key, default, of_context) val = self._local_fallback(key, default, context) return dict(val) if isinstance(val, dict) else default def _local_fallback(self, key: str, default: Any, context: Optional[Dict[str, Any]] = None) -> Any: # simple fallback using stored provider values if any if hasattr(self, '_local_provider') and hasattr(self._local_provider, '_values'): return self._local_provider._values.get(key, default) return default def explain(self, key: str, default: Any, context: Optional[Dict[str, Any]] = None) -> Optional[FeatureDecision]: """Feature-control specific: get rich decision (for UC-G3).""" if self._resolver is not None: return self._resolver.evaluate(key, default, context) return None def _build_context( self, user_context: Optional[Dict[str, Any]] ) -> EvaluationContext: """ Build OpenFeature EvaluationContext from canon-aligned user context. Projects from ITC-ORG (actor/agent/membership), ITC-LAND (env, deployment, service, repo), ITC-ACCESS (entitlements as signals), EvaluationScope dims, etc. See docs/canon-mapping.md, EvaluationScope in specs/UseCaseCatalog.md, and WP-0003. """ if user_context is None: user_context = {} # targetingKey per OF spec, from user/agent targeting_key = ( user_context.get("targeting_key") or user_context.get("user_id") or user_context.get("agent_id") or user_context.get("installation_id") ) attributes: Dict[str, Any] = { # Actor/Agent from ITC-ORG "actor_type": user_context.get("actor_type", "human"), "user_id": user_context.get("user_id"), "agent_id": user_context.get("agent_id"), "roles": user_context.get("roles"), # Landscape from ITC-LAND "installation_id": user_context.get("installation_id"), "environment": user_context.get("environment"), "deployment_id": user_context.get("deployment_id"), "service": user_context.get("service"), "repository": user_context.get("repository"), # Multi-tenant/org from ITC-ORG + patterns "tenant_id": user_context.get("tenant_id"), "domain_id": user_context.get("domain_id"), "organization_id": user_context.get("organization_id"), "group_ids": user_context.get("group_ids"), "plan": user_context.get("plan"), "region": user_context.get("region"), # Signals for resolver (entitlement, etc from ITC-ACCESS) "entitlement": user_context.get("entitlement"), "compute_class": user_context.get("compute_class"), "trust_level": user_context.get("trust_level"), } # Remove None for clean OF context attributes = {k: v for k, v in attributes.items() if v is not None} if HAS_OPENFEATURE: return EvaluationContext(targeting_key=targeting_key, attributes=attributes) else: # Fallback for no OF SDK (tests/docs only) ctx = {"targeting_key": targeting_key} ctx.update(attributes) return ctx # type: ignore[return-value] __all__ = ["FeatureControlClient", "LocalProvider", "FeatureDefinition", "FeatureRegistry", "FeatureDecision", "Resolver"]