m4
4fc74e7da7
Docker / build (push) Has been cancelled
Build / build (push) Has been cancelled
Lint / ruff (push) Has been cancelled
Test / pytest (ubuntu-latest, 3.11) (push) Has been cancelled
Test / pytest (ubuntu-latest, 3.12) (push) Has been cancelled
Test / pytest (windows-latest, 3.11) (push) Has been cancelled
Test / pytest (windows-latest, 3.12) (push) Has been cancelled
117 lines
3.8 KiB
Python
117 lines
3.8 KiB
Python
"""Optional runtime services supplied by an application embedding EvoScientist.
|
|
|
|
The CLI package must not import a concrete web gateway. Applications such as
|
|
Ai4Sci-Web can register their database, storage, metering, and media services
|
|
at process startup through this module.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from collections.abc import Awaitable, Callable
|
|
from dataclasses import dataclass, replace
|
|
from datetime import date
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
AsyncProvider = Callable[[], Awaitable[Any]]
|
|
AsyncFileHandler = Callable[[Path], Awaitable[Any]]
|
|
AsyncUsageRecorder = Callable[[str, str], Awaitable[Any]]
|
|
ModelResolver = Callable[[str | None, str | None], Any | None]
|
|
|
|
|
|
class RuntimeIntegrationUnavailable(RuntimeError):
|
|
"""Raised when an optional host-provided service is not configured."""
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class RuntimeIntegrations:
|
|
app_connection_provider: AsyncProvider | None = None
|
|
session_connection_provider: AsyncProvider | None = None
|
|
session_dsn_provider: Callable[[], str | None] | None = None
|
|
current_date_provider: Callable[[], date] | None = None
|
|
user_storage_root_provider: Callable[[str], Path] | None = None
|
|
knowledge_file_handler: AsyncFileHandler | None = None
|
|
usage_recorder: AsyncUsageRecorder | None = None
|
|
image_backend_factory: Callable[[], Any] | None = None
|
|
model_resolver: ModelResolver | None = None
|
|
|
|
|
|
_integrations = RuntimeIntegrations()
|
|
|
|
|
|
def configure_runtime_integrations(**services: Any) -> RuntimeIntegrations:
|
|
"""Register host-provided services and return the resulting configuration."""
|
|
global _integrations
|
|
_integrations = replace(_integrations, **services)
|
|
return _integrations
|
|
|
|
|
|
def reset_runtime_integrations() -> None:
|
|
"""Clear all host-provided services, primarily for tests."""
|
|
global _integrations
|
|
_integrations = RuntimeIntegrations()
|
|
|
|
|
|
def has_session_connection_provider() -> bool:
|
|
return _integrations.session_connection_provider is not None
|
|
|
|
|
|
def get_session_dsn() -> str | None:
|
|
provider = _integrations.session_dsn_provider
|
|
return provider() if provider is not None else None
|
|
|
|
|
|
async def get_session_connection() -> Any:
|
|
provider = _integrations.session_connection_provider
|
|
if provider is None:
|
|
raise RuntimeIntegrationUnavailable(
|
|
"No session connection provider is configured"
|
|
)
|
|
return await provider()
|
|
|
|
|
|
async def get_app_connection() -> Any:
|
|
provider = _integrations.app_connection_provider
|
|
if provider is None:
|
|
raise RuntimeIntegrationUnavailable(
|
|
"No application connection provider is configured"
|
|
)
|
|
return await provider()
|
|
|
|
|
|
def current_date() -> date:
|
|
provider = _integrations.current_date_provider
|
|
return provider() if provider is not None else date.today()
|
|
|
|
|
|
def resolve_user_storage_root(user_id: str) -> Path | None:
|
|
provider = _integrations.user_storage_root_provider
|
|
return provider(user_id) if provider is not None else None
|
|
|
|
|
|
def resolve_runtime_model(model: str | None, provider: str | None = None) -> Any | None:
|
|
"""Resolve a host-managed model configuration when one is registered."""
|
|
resolver = _integrations.model_resolver
|
|
return resolver(model, provider) if resolver is not None else None
|
|
|
|
|
|
async def handle_knowledge_file(path: Path) -> None:
|
|
handler = _integrations.knowledge_file_handler
|
|
if handler is not None:
|
|
await handler(path)
|
|
|
|
|
|
async def record_service_usage(service: str, action: str) -> None:
|
|
recorder = _integrations.usage_recorder
|
|
if recorder is not None:
|
|
await recorder(service, action)
|
|
|
|
|
|
def get_image_backend() -> Any:
|
|
factory = _integrations.image_backend_factory
|
|
if factory is None:
|
|
raise RuntimeIntegrationUnavailable(
|
|
"Image generation is unavailable in this runtime. Configure an image backend first."
|
|
)
|
|
return factory()
|