""" 进程级托管资源的轻量调用门面。 """ from __future__ import annotations import asyncio import threading from typing import Any, Literal, Optional, Protocol, overload MANAGED_RESOURCE_SYNC_KIND = "managed_resource.sync" MANAGED_RESOURCE_ASYNC_KIND = "managed_resource.async" class ManagedResourceRuntime(Protocol): """ Managed Resource 门面依赖的最小 Capability Runtime 合同。 """ @property def is_shutdown(self) -> bool: """ 返回 Runtime 是否已进入不可逆关闭态。 :return: 运行时是否已进入不可逆关闭态 """ def get_spec(self, capability_id: str) -> Any: """ 返回资源声明。 :param capability_id: 托管资源的能力标识;观测查询传 None 时返回全部能力记录 :return: 对应的能力声明;未登记时返回 None """ def get_running(self, capability_id: str) -> Any: """ 只查询已发布实例。 :param capability_id: 托管资源的能力标识;观测查询传 None 时返回全部能力记录 :return: 已经发布的资源实例;已登记但未运行时返回 None,未知标识由实现方报告领域错误 """ def snapshot(self, capability_id: str) -> Any: """ 返回资源状态快照。 :param capability_id: 托管资源的能力标识;观测查询传 None 时返回全部能力记录 :return: 指定能力的只读状态快照 """ def observations(self, capability_id: Optional[str] = None) -> tuple[Any, ...]: """ 返回资源转换观测。 :param capability_id: 托管资源的能力标识;观测查询传 None 时返回全部能力记录 :return: 按能力筛选的状态转换观测元组 """ def activate(self, capability_id: str, *, reason: str, retry: bool = False) -> Any: """ 通过同步 adapter 激活资源。 :param capability_id: 托管资源的能力标识;观测查询传 None 时返回全部能力记录 :param reason: 本次生命周期操作的触发原因,用于观测记录 :param retry: 是否允许重新尝试此前失败的激活,默认不重试 :return: 激活成功后发布的资源实例;已有可用实例时可直接复用 """ async def activate_async( self, capability_id: str, *, reason: str, retry: bool = False, ) -> Any: """ 通过异步 adapter 激活资源。 :param capability_id: 托管资源的能力标识;观测查询传 None 时返回全部能力记录 :param reason: 本次生命周期操作的触发原因,用于观测记录 :param retry: 是否允许重新尝试此前失败的激活,默认不重试 :return: 等待激活成功后发布的资源实例;已有可用实例时可直接复用 """ def stop(self, capability_id: str, *, reason: str) -> None: """ 通过同步 adapter 停止资源。 :param capability_id: 托管资源的能力标识;观测查询传 None 时返回全部能力记录 :param reason: 本次生命周期操作的触发原因,用于观测记录 """ async def stop_async(self, capability_id: str, *, reason: str) -> None: """ 通过异步 adapter 停止资源。 :param capability_id: 托管资源的能力标识;观测查询传 None 时返回全部能力记录 :param reason: 本次生命周期操作的触发原因,用于观测记录 """ async def shutdown_async(self, *, reason: str) -> bool: """ 关闭混合 Runtime,并返回全部资源是否收敛。 :param reason: 本次生命周期操作的触发原因,用于观测记录 :return: 所有资源均已实际停止时返回 True,否则返回 False 并保留未收敛资源 """ _runtime_lock = threading.RLock() _managed_resource_runtime: Optional[ManagedResourceRuntime] = None def configure_managed_resource_runtime(runtime: ManagedResourceRuntime) -> None: """ 由启动组合层注入唯一的托管资源运行时。 """ if runtime is None: raise ValueError("Managed Resource Runtime 不能为空") global _managed_resource_runtime with _runtime_lock: _managed_resource_runtime = runtime def reset_managed_resource_runtime() -> None: """ 释放已关闭 Runtime 的门面引用;未配置时保持幂等。 """ global _managed_resource_runtime with _runtime_lock: runtime = _managed_resource_runtime if runtime is not None and not runtime.is_shutdown: raise RuntimeError("Managed Resource Runtime 尚未关闭,不能释放引用") _managed_resource_runtime = None @overload def _runtime(*, required: Literal[True]) -> ManagedResourceRuntime: ... @overload def _runtime(*, required: Literal[False]) -> Optional[ManagedResourceRuntime]: ... def _runtime(*, required: bool) -> Optional[ManagedResourceRuntime]: """ 读取当前 Runtime;资源使用路径要求启动组合已经完成装配。 """ with _runtime_lock: runtime = _managed_resource_runtime if runtime is None and required: raise RuntimeError("Managed Resource Runtime 尚未初始化") return runtime def _resource_kind(runtime: ManagedResourceRuntime, capability_id: str) -> str: """ 返回声明的执行模式;未知资源继续沿用 Runtime 的领域错误。 """ spec = runtime.get_spec(capability_id) if spec is None: runtime.get_running(capability_id) raise RuntimeError(f"未知 Managed Resource:{capability_id}") return str(spec.kind) def acquire_managed_resource( capability_id: str, *, reason: str, retry: bool = True, ) -> Any: """ 同步激活一个声明为同步模式的托管资源。 """ runtime = _runtime(required=True) kind = _resource_kind(runtime, capability_id) if kind != MANAGED_RESOURCE_SYNC_KIND: raise RuntimeError(f"异步 Managed Resource 不能通过同步入口激活:{capability_id}") return runtime.activate(capability_id, reason=reason, retry=retry) async def acquire_managed_resource_async( capability_id: str, *, reason: str, retry: bool = True, ) -> Any: """ 异步激活资源;同步资源移交工作线程,避免阻塞事件循环。 """ runtime = _runtime(required=True) kind = _resource_kind(runtime, capability_id) if kind == MANAGED_RESOURCE_ASYNC_KIND: return await runtime.activate_async( capability_id, reason=reason, retry=retry, ) if kind == MANAGED_RESOURCE_SYNC_KIND: return await asyncio.to_thread( runtime.activate, capability_id, reason=reason, retry=retry, ) raise RuntimeError(f"未知 Managed Resource kind:{kind}") def get_running_managed_resource(capability_id: str) -> Any: """ 只查询已发布资源;Runtime 未配置时返回 None,不触发初始化。 """ runtime = _runtime(required=False) if runtime is None: return None return runtime.get_running(capability_id) def managed_resource_snapshot(capability_id: str) -> Any: """ 返回资源状态快照;Runtime 未配置时返回 None。 """ runtime = _runtime(required=False) if runtime is None: return None return runtime.snapshot(capability_id) def managed_resource_observations( capability_id: Optional[str] = None, ) -> tuple[Any, ...]: """ 返回资源转换观测;Runtime 未配置时返回空快照。 """ runtime = _runtime(required=False) if runtime is None: return () return runtime.observations(capability_id) def stop_managed_resource(capability_id: str, *, reason: str) -> None: """ 同步停止资源;Runtime 未配置时保持幂等且不反向初始化。 """ runtime = _runtime(required=False) if runtime is None: return kind = _resource_kind(runtime, capability_id) if kind != MANAGED_RESOURCE_SYNC_KIND: raise RuntimeError(f"异步 Managed Resource 不能通过同步入口停止:{capability_id}") runtime.stop(capability_id, reason=reason) async def stop_managed_resource_async(capability_id: str, *, reason: str) -> None: """ 异步停止资源;同步资源移交工作线程。 """ runtime = _runtime(required=False) if runtime is None: return kind = _resource_kind(runtime, capability_id) if kind == MANAGED_RESOURCE_ASYNC_KIND: await runtime.stop_async(capability_id, reason=reason) return if kind == MANAGED_RESOURCE_SYNC_KIND: await asyncio.to_thread(runtime.stop, capability_id, reason=reason) return raise RuntimeError(f"未知 Managed Resource kind:{kind}") async def shutdown_managed_resource_runtime(*, reason: str) -> bool: """ 关闭已配置 Runtime,未配置时按已收敛处理。 """ runtime = _runtime(required=False) if runtime is None: return True return await runtime.shutdown_async(reason=reason)