|
|
@@ -8,6 +8,8 @@ from ._protocols import ChatClient, ServiceClient
|
|
|
from .types import DayInfo, ServiceAuthError
|
|
|
|
|
|
_log = logging.getLogger(__name__)
|
|
|
+_instance: 'ServiceAPI | None' = None
|
|
|
+
|
|
|
|
|
|
class ServiceAPI:
|
|
|
"""Unified async client for all booking platform services.
|
|
|
@@ -24,25 +26,44 @@ class ServiceAPI:
|
|
|
'booking': _BookingClient,
|
|
|
}
|
|
|
|
|
|
+ @classmethod
|
|
|
+ def get(cls) -> 'ServiceAPI':
|
|
|
+ """Return the process-wide singleton, creating it on first call."""
|
|
|
+ global _instance
|
|
|
+ if _instance is None:
|
|
|
+ _instance = cls()
|
|
|
+ return _instance
|
|
|
+
|
|
|
+ @classmethod
|
|
|
+ def set(cls, instance: 'ServiceAPI') -> None:
|
|
|
+ """Replace the singleton — use in tests to install a stub."""
|
|
|
+ global _instance
|
|
|
+ _instance = instance
|
|
|
+
|
|
|
def __init__(self, clients: dict[str, ServiceClient] | None = None) -> None:
|
|
|
if clients is not None:
|
|
|
self._clients: dict[str, ServiceClient] = clients
|
|
|
+ self._disabled: dict[str, str] = {}
|
|
|
return
|
|
|
- from room_cal._utils import get_config
|
|
|
self._clients = {}
|
|
|
self._disabled: dict[str, str] = {}
|
|
|
- svc_cfg = get_config().services
|
|
|
for key, cls in self._CLIENT_CLASSES.items():
|
|
|
- if (cfg := svc_cfg.get(key)) and cfg.disabled:
|
|
|
- _log.info('service %s disabled by config', key)
|
|
|
- self._disabled[key] = 'disabled in config'
|
|
|
- continue
|
|
|
try:
|
|
|
self._clients[key] = cls.from_env()
|
|
|
except Exception as e:
|
|
|
- _log.warning('service %s disabled: %s', key, e)
|
|
|
+ _log.warning('service %s failed to load: %s', key, e)
|
|
|
self._disabled[key] = str(e)
|
|
|
|
|
|
+ def keys(self) -> list[str]:
|
|
|
+ """Return the keys of all successfully loaded services."""
|
|
|
+ return list(self._clients.keys())
|
|
|
+
|
|
|
+ def service_name(self, key: str) -> str:
|
|
|
+ """Return the human-readable display name for a service key."""
|
|
|
+ cls = self._CLIENT_CLASSES.get(key)
|
|
|
+ return getattr(cls, 'service_name', key.capitalize()) if cls else key.capitalize()
|
|
|
+
|
|
|
+
|
|
|
async def refresh_credentials(self, key: str) -> None:
|
|
|
"""Trigger credential refresh for the named service. Raises ServiceAuthError on failure."""
|
|
|
cls = self._CLIENT_CLASSES.get(key)
|