diff --git a/schedule/src/schedule/infrastructure/__init__.py b/schedule/src/schedule/infrastructure/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/schedule/src/schedule/infrastructure/storage/__init__.py b/schedule/src/schedule/infrastructure/storage/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/schedule/src/schedule/infrastructure/storage/client.py b/schedule/src/schedule/infrastructure/storage/client.py new file mode 100644 index 0000000..72a97a9 --- /dev/null +++ b/schedule/src/schedule/infrastructure/storage/client.py @@ -0,0 +1,82 @@ +"""Schedule-side HTTP client for the backend's storage API. + +The schedule worker uploads run logs / run results by calling +``POST {backend}/internal/v1/objects`` (the backend's +``create_server_object_payload`` route, which is in the same process +as the public API). The response is the StorageObjects row payload. + +The schedule does NOT have its own DB session for storage metadata, +so it must go through the backend to create the StorageObjects row +(``logs_object_id`` / ``result_object_id`` are FKs into that table). +""" + +from __future__ import annotations + +import base64 +from typing import Any + +import httpx +from loguru import logger + + +class SchedulerStorageClient: + def __init__(self, http_client: httpx.AsyncClient) -> None: + self._http = http_client + + async def create_object( + self, + *, + workspace_id: str, + user_id: str, + usage_type: str, + file_name: str, + content_type: str, + content: bytes, + idempotency_key: str, + ) -> dict[str, Any]: + """Upload a run_log / run_result via the backend's storage API. + + The backend returns ``{"data": , "meta": {...}}``; + we return the inner ``data`` dict (which includes + ``storage_object_id`` and ``storage_uri``). + """ + logger.debug( + "storage create_object: workspace={} usage_type={} file={} size={}B", + workspace_id[-12:], + usage_type, + file_name, + len(content), + ) + response = await self._http.post( + "/internal/v1/objects", + json={ + "workspace_id": workspace_id, + "user_id": user_id, + "usage_type": usage_type, + "file_name": file_name, + "content_type": content_type, + "content_base64": base64.b64encode(content).decode("ascii"), + "visibility": "workspace", + "is_immutable": True, + "idempotency_key": idempotency_key, + "relative_path": None, + }, + ) + if response.is_error: + logger.warning( + "storage create_object HTTP error: status={} url={}", + response.status_code, + response.request.url, + ) + response.raise_for_status() + body = response.json() + logger.info( + "storage create_object done: workspace={} usage_type={} storage_object_id={}", + workspace_id[-12:], + usage_type, + body["data"].get("storage_object_id"), + ) + return body["data"] + + +__all__ = ["SchedulerStorageClient"] diff --git a/schedule/src/schedule/main.py b/schedule/src/schedule/main.py index 237816e..3c978f3 100644 --- a/schedule/src/schedule/main.py +++ b/schedule/src/schedule/main.py @@ -14,7 +14,7 @@ from schedule.service import ( build_object_store, build_storage_http_client, ) -from schedule.storage_client import SchedulerStorageClient +from schedule.infrastructure.storage.client import SchedulerStorageClient @asynccontextmanager