feat(cloud-api): compose control plane services

This commit is contained in:
2026-07-12 18:21:57 +08:00
parent 09cfe54dd9
commit bd261e4992
3 changed files with 102 additions and 4 deletions
+70 -3
View File
@@ -2,21 +2,54 @@ from __future__ import annotations
from collections.abc import Callable
from contextlib import asynccontextmanager
from dataclasses import dataclass
from typing import Any
from fastapi import FastAPI
from cloud.auth import create_auth_provider
from cloud.config import CloudConfig
from cloud.control_config import (
CloudControlConfig,
load_control_config,
validate_control_config,
)
from cloud.database import CloudDatabase
from cloud.internal_api.api import create_internal_router
from cloud.plugins import PluginRegistry
from cloud.pool import DevicePool
from cloud.scheduler import TaskScheduler
from cloud.sdk.api import create_cloud_router
DatabaseFactory = Callable[[CloudControlConfig], CloudDatabase]
class _RepositoryProxy:
def __init__(self) -> None:
self._target: Any | None = None
def bind(self, target: Any) -> None:
self._target = target
def unbind(self) -> None:
self._target = None
def __getattr__(self, name: str) -> Any:
if self._target is None:
raise RuntimeError("cloud repository is not available outside app lifespan")
return getattr(self._target, name)
@dataclass(frozen=True)
class CloudApplicationServices:
repository: _RepositoryProxy
pool: DevicePool
scheduler: TaskScheduler
plugin_registry: PluginRegistry
auth_provider: Any
def create_app(
*,
config: CloudControlConfig | None = None,
@@ -30,19 +63,53 @@ def create_app(
control_config.credentials,
allow_insecure_anonymous=control_config.allow_insecure_anonymous,
)
repository = _RepositoryProxy()
domain_config = CloudConfig(
lease_duration_seconds=control_config.lease_duration_seconds,
)
pool = DevicePool(repository, domain_config) # type: ignore[arg-type]
scheduler = TaskScheduler(pool, repository, domain_config) # type: ignore[arg-type]
plugin_registry = PluginRegistry(repository) # type: ignore[arg-type]
services = CloudApplicationServices(
repository=repository,
pool=pool,
scheduler=scheduler,
plugin_registry=plugin_registry,
auth_provider=auth_provider,
)
@asynccontextmanager
async def lifespan(app: FastAPI):
database = build_database(control_config)
repository.bind(database.repository)
app.state.cloud_config = control_config
app.state.database = database
app.state.auth_provider = auth_provider
app.state.cloud_services = services
try:
yield
finally:
database.close()
try:
database.close()
finally:
repository.unbind()
return FastAPI(title="Device Cloud API", lifespan=lifespan)
app = FastAPI(title="Device Cloud API", lifespan=lifespan)
app.include_router(
create_cloud_router(
pool=pool,
scheduler=scheduler,
plugin_registry=plugin_registry,
auth_provider=auth_provider,
)
)
app.include_router(
create_internal_router(
pool=pool,
auth_provider=auth_provider,
lease_duration_seconds=control_config.lease_duration_seconds,
)
)
return app
def _default_database_factory(config: CloudControlConfig) -> CloudDatabase:
+31
View File
@@ -12,12 +12,17 @@ def test_create_app_returns_independent_cloud_application() -> None:
assert app.title == "Device Cloud API"
assert callable(create_app)
paths = set(app.openapi()["paths"])
assert "/v1/tasks" in paths
assert "/internal/v1/hosts/{host_id}/heartbeat" in paths
def test_cloud_application_owns_database_lifecycle() -> None:
events: list[str] = []
class FakeDatabase:
repository = object()
def close(self) -> None:
events.append("closed")
@@ -29,11 +34,37 @@ def test_cloud_application_owns_database_lifecycle() -> None:
with TestClient(app):
assert app.state.database is fake_database
assert (
app.state.cloud_services.pool.store is app.state.cloud_services.repository
)
assert events == []
assert events == ["closed"]
def test_repository_proxy_is_available_only_during_lifespan() -> None:
class FakeRepository:
def health_check(self) -> None:
return None
class FakeDatabase:
repository = FakeRepository()
def close(self) -> None:
return None
app = create_app(
config=CloudControlConfig(database_url="sqlite:///:memory:"),
database_factory=lambda _config: FakeDatabase(), # type: ignore[arg-type,return-value]
)
with TestClient(app):
app.state.cloud_services.repository.health_check()
with pytest.raises(RuntimeError, match="outside app lifespan"):
app.state.cloud_services.repository.health_check()
def test_production_app_rejects_missing_credentials() -> None:
with pytest.raises(CloudConfigurationError, match="credential"):
create_app(