refactor(schedule): extract infrastructure/storage/ layer

Stage 2 of the layered refactor. Move the storage HTTP client one
package deeper so that infrastructure code lives under a dedicated
namespace.

- Add schedule/src/schedule/infrastructure/__init__.py
- Add schedule/src/schedule/infrastructure/storage/__init__.py
- Add schedule/src/schedule/infrastructure/storage/client.py
  (verbatim copy of old schedule/src/schedule/storage_client.py,
  byte-identical via diff — 2682 bytes)
- main.py line 17: import path rewrite to the new module
  (only consumer — service.py and worker.py take storage_client as
  an `Any` constructor param and never imported the class)
- old schedule/src/schedule/storage_client.py left on disk; stage 6
  deletes it once all layers are extracted.

Validation:
- uv run --package schedule pytest schedule/tests -q: 18 passed
- uv run python -m compileall schedule/src: zero errors
- grep 'from schedule.storage_client' (old path): 0 matches
- service.py and worker.py byte-identical to HEAD
- main.py / pyproject.toml / tests/ unchanged apart from the 1 import line

Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
tao.chen
2026-09-02 10:10:41 +08:00
committed by tao.chen
co-authored by Claude
parent 53ad6718ed
commit 8675868bdc
4 changed files with 83 additions and 1 deletions
@@ -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": <StorageObjectPayload>, "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"]
+1 -1
View File
@@ -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