# src/pages/builder/templates/agent-python/store.py """Agent store — InMemoryStore for dev, BlobBackedStore for production.""" from __future__ import annotations import asyncio import json import time from typing import Any, Optional STORE_ENVELOPE_VERSION = 1 # ─── Envelope helpers (match Node store.ts format) ─── def _build_envelope(value: Any, ttl: Optional[int] = None) -> dict: """Wrap a value in the cross-language envelope format.""" expires_at: Optional[int] = None if ttl is not None and ttl > 0: expires_at = int(time.time() * 1000) + int(ttl * 1000) return { "__eo_pages_store_v": STORE_ENVELOPE_VERSION, "value": value, "expiresAt": expires_at, "updatedAt": int(time.time() * 1000), } def _read_envelope(data: Any) -> Optional[dict]: """Extract envelope if data matches the versioned format. Returns None otherwise.""" if not isinstance(data, dict): return None if data.get("__eo_pages_store_v") != STORE_ENVELOPE_VERSION: return None expires_at = data.get("expiresAt") if not (expires_at is None or isinstance(expires_at, (int, float))): return None updated_at = data.get("updatedAt") if not isinstance(updated_at, (int, float)): return None return { "__eo_pages_store_v": STORE_ENVELOPE_VERSION, "value": data.get("value"), "expiresAt": expires_at, "updatedAt": updated_at, } def _is_expired(expires_at: Optional[int]) -> bool: return expires_at is not None and expires_at <= int(time.time() * 1000) # ─── InMemoryStore (dev / fallback) ─── class InMemoryStore: """Dict-based store with TTL and envelope semantics for local dev.""" def __init__(self) -> None: self._data: dict[str, Any] = {} self._expiry: dict[str, float] = {} async def get(self, key: str) -> Optional[Any]: self._evict(key) return self._data.get(key) async def set(self, key: str, value: Any, ttl: Optional[int] = None) -> None: self._data[key] = value if ttl is not None: self._expiry[key] = time.monotonic() + ttl elif key in self._expiry: del self._expiry[key] async def delete(self, key: str) -> None: self._data.pop(key, None) self._expiry.pop(key, None) async def list(self, *, prefix: Optional[str] = None) -> dict: now = time.monotonic() keys = [] for k in sorted(self._data.keys()): exp = self._expiry.get(k) if exp is not None and now > exp: continue if prefix is None or k.startswith(prefix): keys.append(k) return {"keys": keys, "complete": True} def _evict(self, key: str) -> None: exp = self._expiry.get(key) if exp is not None and time.monotonic() > exp: self._data.pop(key, None) self._expiry.pop(key, None) # ─── BlobBackedStore (production) ─── class BlobBackedStore: """Persistent store backed by pages-blob, with envelope format matching Node agent.""" def __init__(self, blob_store: Any) -> None: """ Args: blob_store: A pages_blob.Store instance from get_store(name). """ self._blob = blob_store async def get(self, key: str) -> Optional[Any]: raw = await self._blob.get(key, type="text") if raw is None: return None parsed = _try_parse_json(raw) envelope = _read_envelope(parsed) if envelope: if _is_expired(envelope["expiresAt"]): await self._blob.delete(key) return None return envelope["value"] # Backward compat: raw data without envelope return parsed async def set(self, key: str, value: Any, ttl: Optional[int] = None) -> None: envelope = _build_envelope(value, ttl) payload = json.dumps(envelope, ensure_ascii=False) await self._blob.set(key, payload) async def delete(self, key: str) -> None: await self._blob.delete(key) async def list(self, *, prefix: Optional[str] = None) -> dict: result = await self._blob.list(prefix=prefix) keys = [blob.key for blob in result.blobs] return {"keys": keys, "complete": True} def _try_parse_json(raw: str) -> Any: try: return json.loads(raw) except (json.JSONDecodeError, TypeError, ValueError): return raw # ─── SyncStoreWrapper (for executor threads) ─── class SyncStoreWrapper: """同步 wrapper —— 在 run_in_executor 等非 event-loop 线程里调 async store。 典型场景：CrewAI 的 crew.kickoff() 是同步调用，在 executor 线程里跑， 无法 await。通过 asyncio.run_coroutine_threadsafe 把协程提交回主 event loop。 用法（handler 内）:: async def handler(ctx): import asyncio def sync_work(): # 这里跑在 executor 线程，不能 await cached = ctx.store_sync.get("key") ctx.store_sync.set("key", value, ttl=3600) return cached result = await asyncio.get_running_loop().run_in_executor(None, sync_work) ⚠️ 不要在 async handler 顶层直接用 store_sync —— 会死锁。 async 上下文直接 ``await ctx.store.get(...)`` 即可。 """ def __init__(self, async_store: Any, loop: asyncio.AbstractEventLoop, timeout: float = 30.0) -> None: self._store = async_store self._loop = loop self._timeout = timeout def _run(self, coro: Any) -> Any: # 死锁保护：如果当前线程就是 event loop 线程， # run_coroutine_threadsafe + future.result() 会永远阻塞。 try: running = asyncio.get_running_loop() except RuntimeError: running = None if running is self._loop: raise RuntimeError( "SyncStoreWrapper._run() 被在 event loop 线程调用，这会导致死锁。" "请在 async 上下文里直接 await store 的异步方法。" ) future = asyncio.run_coroutine_threadsafe(coro, self._loop) return future.result(timeout=self._timeout) def get(self, key: str) -> Optional[Any]: return self._run(self._store.get(key)) def set(self, key: str, value: Any, ttl: Optional[int] = None) -> None: return self._run(self._store.set(key, value, ttl=ttl)) def delete(self, key: str) -> None: return self._run(self._store.delete(key)) def list(self, *, prefix: Optional[str] = None) -> dict: return self._run(self._store.list(prefix=prefix))