update: debug create file by jupyter
This commit is contained in:
@@ -4,7 +4,7 @@ from dataclasses import dataclass
|
|||||||
from typing import Any
|
from typing import Any
|
||||||
|
|
||||||
import httpx
|
import httpx
|
||||||
|
from loguru import logger
|
||||||
|
|
||||||
@dataclass(frozen=True)
|
@dataclass(frozen=True)
|
||||||
class RuntimeClientError(Exception):
|
class RuntimeClientError(Exception):
|
||||||
@@ -77,5 +77,133 @@ class RuntimeClient:
|
|||||||
{"action": "start", "workspace_id": workspace_id},
|
{"action": "start", "workspace_id": workspace_id},
|
||||||
)
|
)
|
||||||
|
|
||||||
|
async def _ensure_workspace(
|
||||||
|
self,
|
||||||
|
workspace_id: str,
|
||||||
|
) -> dict[str, Any]:
|
||||||
|
"""Return a running workspace descriptor, starting it if needed.
|
||||||
|
|
||||||
|
Mirrors the lazy-start pattern used by
|
||||||
|
:func:`backend.jupyter.verify_jupyter_access`: try ``get``
|
||||||
|
first, fall through to ``start`` if the workspace is not yet
|
||||||
|
running. Bumps ``last_used_at`` via the runtime registry on the
|
||||||
|
way in, so the idle reaper is satisfied for the duration of the
|
||||||
|
subsequent Jupyter call.
|
||||||
|
"""
|
||||||
|
ws = await self.get_workspace(workspace_id)
|
||||||
|
if not ws or ws.get("status") != "running":
|
||||||
|
ws = await self.start_workspace(workspace_id)
|
||||||
|
if not ws.get("port") or not ws.get("token"):
|
||||||
|
raise RuntimeClientError(
|
||||||
|
500,
|
||||||
|
{
|
||||||
|
"code": "JUPYTER_DESCRIPTOR_INVALID",
|
||||||
|
"message": "Runtime returned no port/token for workspace",
|
||||||
|
"retryable": False,
|
||||||
|
"details": {"workspace_id": workspace_id},
|
||||||
|
},
|
||||||
|
)
|
||||||
|
return ws
|
||||||
|
|
||||||
|
async def _jupyter_contents_post(
|
||||||
|
self,
|
||||||
|
workspace_id: str,
|
||||||
|
ws: dict[str, Any],
|
||||||
|
body: dict[str, Any],
|
||||||
|
) -> dict[str, Any]:
|
||||||
|
"""POST to the workspace Jupyter's ``/api/contents/`` directly.
|
||||||
|
|
||||||
|
Builds the URL from the per-workspace ``base_url`` + ``port``
|
||||||
|
and authenticates with the runtime-issued token. We reuse the
|
||||||
|
existing ``httpx.AsyncClient`` (its ``base_url`` is overridden
|
||||||
|
by the absolute URL we hand it).
|
||||||
|
"""
|
||||||
|
url = (
|
||||||
|
f"{ws['base_url']}:{ws['port']}"
|
||||||
|
f"/jupyter/{workspace_id}/api/contents/"
|
||||||
|
)
|
||||||
|
logger.debug(url)
|
||||||
|
logger.debug(body)
|
||||||
|
headers = {"Authorization": f"token {ws['token']}"}
|
||||||
|
try:
|
||||||
|
response = await self.client.post(
|
||||||
|
url, json=body, headers=headers
|
||||||
|
)
|
||||||
|
except httpx.RequestError as exc:
|
||||||
|
raise _RUNTIME_TRANSPORT_ERROR from exc
|
||||||
|
if response.is_error:
|
||||||
|
try:
|
||||||
|
detail = response.json()
|
||||||
|
except ValueError:
|
||||||
|
detail = response.text
|
||||||
|
raise RuntimeClientError(response.status_code, detail)
|
||||||
|
return response.json()
|
||||||
|
|
||||||
|
async def create_notebook(
|
||||||
|
self,
|
||||||
|
workspace_id: str,
|
||||||
|
*,
|
||||||
|
name: str,
|
||||||
|
cells: list[dict[str, Any]] | None = None,
|
||||||
|
) -> dict[str, Any]:
|
||||||
|
"""Create a notebook in the workspace root.
|
||||||
|
|
||||||
|
Auto-starts the workspace's Jupyter if it is not yet running,
|
||||||
|
then posts to the Jupyter contents API
|
||||||
|
(``/jupyter/{ws_id}/api/contents/``) directly using the
|
||||||
|
runtime-issued token. Returns the Jupyter contents descriptor
|
||||||
|
(``path``, ``type``, ``created``, ``writable``...).
|
||||||
|
|
||||||
|
``name`` must be a safe basename (no path separators); the
|
||||||
|
Jupyter side will reject anything that escapes the workspace
|
||||||
|
root.
|
||||||
|
"""
|
||||||
|
ws = await self._ensure_workspace(workspace_id)
|
||||||
|
logger.debug(ws)
|
||||||
|
body = {
|
||||||
|
"type": "notebook",
|
||||||
|
"name": name,
|
||||||
|
"content": {
|
||||||
|
"cells": cells if cells is not None else [],
|
||||||
|
"metadata": {},
|
||||||
|
"nbformat": 4,
|
||||||
|
"nbformat_minor": 5,
|
||||||
|
},
|
||||||
|
}
|
||||||
|
|
||||||
|
return await self._jupyter_contents_post(workspace_id, ws, body)
|
||||||
|
|
||||||
|
async def upload_file(
|
||||||
|
self,
|
||||||
|
workspace_id: str,
|
||||||
|
*,
|
||||||
|
name: str,
|
||||||
|
content: str,
|
||||||
|
content_type: str = "text/plain",
|
||||||
|
) -> dict[str, Any]:
|
||||||
|
"""Upload a text file to the workspace root.
|
||||||
|
|
||||||
|
``content_type`` selects the Jupyter ``format`` field: anything
|
||||||
|
under ``text/`` or ``application/json`` is sent as ``text``.
|
||||||
|
Binary uploads are not yet supported and raise ``ValueError``
|
||||||
|
— the Jupyter contents API base64 pathway is not wired up here.
|
||||||
|
"""
|
||||||
|
if not (
|
||||||
|
content_type.startswith("text/")
|
||||||
|
or content_type == "application/json"
|
||||||
|
):
|
||||||
|
raise ValueError(
|
||||||
|
f"content_type '{content_type}' is not supported; "
|
||||||
|
"only text/* and application/json are accepted"
|
||||||
|
)
|
||||||
|
ws = await self._ensure_workspace(workspace_id)
|
||||||
|
body = {
|
||||||
|
"type": "file",
|
||||||
|
"name": name,
|
||||||
|
"content": content,
|
||||||
|
"format": "text",
|
||||||
|
}
|
||||||
|
return await self._jupyter_contents_post(workspace_id, ws, body)
|
||||||
|
|
||||||
|
|
||||||
__all__ = ["RuntimeClient", "RuntimeClientError"]
|
__all__ = ["RuntimeClient", "RuntimeClientError"]
|
||||||
|
|||||||
@@ -7,7 +7,7 @@ import mimetypes
|
|||||||
from datetime import UTC, datetime
|
from datetime import UTC, datetime
|
||||||
from pathlib import PurePosixPath
|
from pathlib import PurePosixPath
|
||||||
from typing import Any
|
from typing import Any
|
||||||
|
from loguru import logger
|
||||||
from fastapi import (
|
from fastapi import (
|
||||||
APIRouter,
|
APIRouter,
|
||||||
BackgroundTasks,
|
BackgroundTasks,
|
||||||
@@ -32,6 +32,7 @@ from backend.dependencies import (
|
|||||||
database_session,
|
database_session,
|
||||||
request_context,
|
request_context,
|
||||||
)
|
)
|
||||||
|
from backend.runtime_client import RuntimeClientError
|
||||||
from backend.schemas import (
|
from backend.schemas import (
|
||||||
CreateScriptRequest,
|
CreateScriptRequest,
|
||||||
CreateWorkspaceDirectoryRequest,
|
CreateWorkspaceDirectoryRequest,
|
||||||
@@ -279,7 +280,7 @@ async def create_script_record(
|
|||||||
) if parent_path is None else normalize_user_path(parent_path)
|
) if parent_path is None else normalize_user_path(parent_path)
|
||||||
child_path = f"{folder}/{name}" if folder else name
|
child_path = f"{folder}/{name}" if folder else name
|
||||||
relative_path = user_relative_path(context, child_path)
|
relative_path = user_relative_path(context, child_path)
|
||||||
|
logger.debug(relative_path)
|
||||||
existing_script = await session.scalar(
|
existing_script = await session.scalar(
|
||||||
select(Scripts)
|
select(Scripts)
|
||||||
.join(
|
.join(
|
||||||
@@ -290,6 +291,7 @@ async def create_script_record(
|
|||||||
Scripts.workspace_id == context.workspace.workspace_id,
|
Scripts.workspace_id == context.workspace.workspace_id,
|
||||||
StorageObjects.relative_path == relative_path,
|
StorageObjects.relative_path == relative_path,
|
||||||
Scripts.status == "active",
|
Scripts.status == "active",
|
||||||
|
Scripts.is_deleted == 0,
|
||||||
)
|
)
|
||||||
)
|
)
|
||||||
if existing_script is not None:
|
if existing_script is not None:
|
||||||
@@ -298,19 +300,51 @@ async def create_script_record(
|
|||||||
"a file with the same path already exists",
|
"a file with the same path already exists",
|
||||||
)
|
)
|
||||||
|
|
||||||
storage_data = await request.app.state.storage_client.create_server_object(
|
# Push the file directly to the workspace's live Jupyter instance.
|
||||||
workspace_id=context.workspace.workspace_id,
|
# Auto-starts the workspace if no Jupyter is running yet. Once
|
||||||
user_id=context.user.user_id,
|
# Jupyter has the file in its local mount, the rclone VFS will
|
||||||
usage_type="working_copy",
|
# eventually replicate it back to object storage.
|
||||||
file_name=name,
|
runtime_client = request.app.state.runtime_client
|
||||||
content_type=mimetypes.guess_type(name)[0]
|
workspace_id = context.workspace.workspace_id
|
||||||
or "application/octet-stream",
|
content_hash = hashlib.sha256(content).hexdigest()
|
||||||
content=content,
|
size_bytes = len(content)
|
||||||
visibility=visibility,
|
try:
|
||||||
is_immutable=False,
|
if script_type == "notebook":
|
||||||
idempotency_key=f"script:{context.workspace.workspace_id}:{relative_path}",
|
notebook = json.loads(content.decode("utf-8"))
|
||||||
relative_path=relative_path,
|
logger.debug(notebook)
|
||||||
)
|
jupyter_resp = await runtime_client.create_notebook(
|
||||||
|
workspace_id,
|
||||||
|
name=name,
|
||||||
|
cells=notebook.get("cells"),
|
||||||
|
)
|
||||||
|
else:
|
||||||
|
jupyter_resp = await runtime_client.upload_file(
|
||||||
|
workspace_id,
|
||||||
|
name=name,
|
||||||
|
content=content.decode("utf-8"),
|
||||||
|
content_type=(
|
||||||
|
mimetypes.guess_type(name)[0] or "text/plain"
|
||||||
|
),
|
||||||
|
)
|
||||||
|
logger.debug(jupyter_resp)
|
||||||
|
except RuntimeClientError as exc:
|
||||||
|
raise HTTPException(
|
||||||
|
status_code=exc.status_code,
|
||||||
|
detail=exc.detail,
|
||||||
|
) from exc
|
||||||
|
|
||||||
|
# Synthesise a storage_data-shaped dict from the Jupyter response
|
||||||
|
# so the existing script_payload + response shape keep working.
|
||||||
|
# The storage_object_id is a fresh ULID — there is no real
|
||||||
|
# StorageObject row for this file; downstream list/get operations
|
||||||
|
# that JOIN StorageObjects will skip jupyter-only scripts.
|
||||||
|
storage_data = {
|
||||||
|
"storage_object_id": new_ulid(),
|
||||||
|
"relative_path": relative_path,
|
||||||
|
"object_key": f"{workspace_id}/{relative_path}",
|
||||||
|
"content_hash": content_hash,
|
||||||
|
"size_bytes": size_bytes,
|
||||||
|
}
|
||||||
object_id = storage_data["storage_object_id"]
|
object_id = storage_data["storage_object_id"]
|
||||||
script = await session.scalar(
|
script = await session.scalar(
|
||||||
select(Scripts).where(Scripts.current_object_id == object_id)
|
select(Scripts).where(Scripts.current_object_id == object_id)
|
||||||
|
|||||||
Reference in New Issue
Block a user