chendelian
117d31298e
Add a shared SDK scaffold for entitlement checks and a reusable Gitea release workflow template to standardize skill packaging and publishing across projects.
64 lines
2.2 KiB
Python
64 lines
2.2 KiB
Python
import os
|
|
from typing import Any
|
|
|
|
import requests
|
|
|
|
from .errors import EntitlementServiceError
|
|
from .models import EntitlementResult
|
|
|
|
|
|
class EntitlementClient:
|
|
def __init__(
|
|
self,
|
|
base_url: str | None = None,
|
|
api_key: str | None = None,
|
|
timeout_seconds: int | None = None,
|
|
) -> None:
|
|
self.base_url = (base_url or os.getenv("JIANGCHANG_AUTH_BASE_URL", "")).rstrip("/")
|
|
self.api_key = api_key or os.getenv("JIANGCHANG_AUTH_API_KEY", "")
|
|
self.timeout_seconds = timeout_seconds or int(
|
|
os.getenv("JIANGCHANG_AUTH_TIMEOUT_SECONDS", "5")
|
|
)
|
|
if not self.base_url:
|
|
raise EntitlementServiceError("missing JIANGCHANG_AUTH_BASE_URL")
|
|
|
|
def check_entitlement(
|
|
self,
|
|
user_id: str,
|
|
skill_slug: str,
|
|
trace_id: str = "",
|
|
context: dict[str, Any] | None = None,
|
|
) -> EntitlementResult:
|
|
url = f"{self.base_url}/api/entitlements/check"
|
|
payload = {
|
|
"user_id": user_id,
|
|
"skill_slug": skill_slug,
|
|
"trace_id": trace_id,
|
|
"context": context or {},
|
|
}
|
|
headers = {"Content-Type": "application/json"}
|
|
if self.api_key:
|
|
headers["Authorization"] = f"Bearer {self.api_key}"
|
|
try:
|
|
res = requests.post(
|
|
url,
|
|
json=payload,
|
|
headers=headers,
|
|
timeout=self.timeout_seconds,
|
|
)
|
|
except requests.RequestException as exc:
|
|
raise EntitlementServiceError(f"entitlement request failed: {exc}") from exc
|
|
|
|
if res.status_code != 200:
|
|
raise EntitlementServiceError(f"entitlement http status {res.status_code}")
|
|
try:
|
|
body = res.json()
|
|
except ValueError as exc:
|
|
raise EntitlementServiceError("entitlement response is not json") from exc
|
|
|
|
data = body.get("data") or {}
|
|
allow = bool(data.get("allow", False))
|
|
reason = str(data.get("reason") or body.get("msg") or "")
|
|
expire_at = str(data.get("expire_at") or "")
|
|
return EntitlementResult(allow=allow, reason=reason, expire_at=expire_at, raw=body)
|