update: 数据资源文件上传
This commit is contained in:
@@ -55,15 +55,11 @@ def resource_payload(
|
|||||||
resource: DataResources,
|
resource: DataResources,
|
||||||
storage_object: StorageObjects,
|
storage_object: StorageObjects,
|
||||||
) -> dict[str, Any]:
|
) -> dict[str, Any]:
|
||||||
# ``object_key`` is stored as ``{ws_id}/{user_id}/.resources/{file_name}``
|
# ``object_key`` now follows ``{ws_id}/{user_id}/{target_path}/{file_name}``
|
||||||
# (see ``backend/services/storage.py:create_upload_record``). Two
|
# (target_path may be empty). Legacy objects still live under
|
||||||
# Jupyter-facing path views:
|
# ``.resources/{file_name}`` and must remain readable. Both shapes share
|
||||||
# - ``jupyter_accessible_path``: per-user Jupyter-relative path. The
|
# the same derivation: strip the workspace/user prefix and use the rest
|
||||||
# Jupyter root_dir is already ``workspace/{ws_id}/{user_id}/``, so
|
# as the Jupyter-relative path.
|
||||||
# we return the tail ``.resources/{file_name}`` for copy-paste use.
|
|
||||||
# - ``absolute_path``: full filesystem path inside the Jupyter
|
|
||||||
# container (the rclone mount root). Shape:
|
|
||||||
# ``{workspaces_root}/{ws_id}/{user_id}/.resources/{file_name}``.
|
|
||||||
workspace_prefix = f"{resource.workspace_id}/"
|
workspace_prefix = f"{resource.workspace_id}/"
|
||||||
user_prefix = f"{resource.owner_user_id}/"
|
user_prefix = f"{resource.owner_user_id}/"
|
||||||
jupyter_accessible_path = ""
|
jupyter_accessible_path = ""
|
||||||
@@ -73,7 +69,7 @@ def resource_payload(
|
|||||||
):
|
):
|
||||||
remainder = storage_object.object_key[len(workspace_prefix):]
|
remainder = storage_object.object_key[len(workspace_prefix):]
|
||||||
if remainder.startswith(user_prefix):
|
if remainder.startswith(user_prefix):
|
||||||
tail = remainder[len(user_prefix):] # ``.resources/{file_name}``
|
tail = remainder[len(user_prefix):]
|
||||||
jupyter_accessible_path = tail
|
jupyter_accessible_path = tail
|
||||||
absolute_path = (
|
absolute_path = (
|
||||||
workspaces_root()
|
workspaces_root()
|
||||||
@@ -141,11 +137,7 @@ async def resource_jupyter_relative_path(
|
|||||||
raise HTTPException(422, "script_path must be a clean Jupyter-relative POSIX path")
|
raise HTTPException(422, "script_path must be a clean Jupyter-relative POSIX path")
|
||||||
|
|
||||||
resource_path = resource_payload(resource, storage_object)["jupyter_accessible_path"]
|
resource_path = resource_payload(resource, storage_object)["jupyter_accessible_path"]
|
||||||
if not resource_path.startswith(".resources/"):
|
# Both legacy ``.resources/`` files and new flat/nested paths are valid.
|
||||||
raise HTTPException(
|
|
||||||
status.HTTP_409_CONFLICT,
|
|
||||||
"resource is not stored under .resources/",
|
|
||||||
)
|
|
||||||
relative = compute_jupyter_relative_path(script_path, resource_path)
|
relative = compute_jupyter_relative_path(script_path, resource_path)
|
||||||
return {
|
return {
|
||||||
"request_id": context.request_id,
|
"request_id": context.request_id,
|
||||||
@@ -176,6 +168,7 @@ async def create_resource_upload(
|
|||||||
expected_size_bytes=payload.expected_size_bytes,
|
expected_size_bytes=payload.expected_size_bytes,
|
||||||
expected_hash=payload.expected_hash,
|
expected_hash=payload.expected_hash,
|
||||||
idempotency_key=idempotency_key,
|
idempotency_key=idempotency_key,
|
||||||
|
target_path=payload.target_path,
|
||||||
),
|
),
|
||||||
session,
|
session,
|
||||||
request,
|
request,
|
||||||
|
|||||||
@@ -9,6 +9,18 @@ class CreateResourceUploadRequest(StrictModel):
|
|||||||
content_type: str = Field(min_length=1, max_length=255)
|
content_type: str = Field(min_length=1, max_length=255)
|
||||||
expected_size_bytes: int = Field(ge=0, le=100 * 1024 * 1024)
|
expected_size_bytes: int = Field(ge=0, le=100 * 1024 * 1024)
|
||||||
expected_hash: str | None = Field(default=None, min_length=64, max_length=64)
|
expected_hash: str | None = Field(default=None, min_length=64, max_length=64)
|
||||||
|
target_path: str = Field(default="", max_length=1024)
|
||||||
|
|
||||||
|
@field_validator("target_path")
|
||||||
|
@classmethod
|
||||||
|
def validate_target_path(cls, value: str) -> str:
|
||||||
|
# POSIX 相对路径,不能含 ..、绝对前缀、控制字符
|
||||||
|
if value and (value.startswith("/") or "\\" in value
|
||||||
|
or any(seg == ".." for seg in value.split("/"))
|
||||||
|
or any(ord(c) < 0x20 or ord(c) == 0x7F for c in value)):
|
||||||
|
raise ValueError("target_path must be a clean relative POSIX path")
|
||||||
|
# 去前导 /;允许尾部 /
|
||||||
|
return value.strip("/")
|
||||||
|
|
||||||
@field_validator("expected_hash")
|
@field_validator("expected_hash")
|
||||||
@classmethod
|
@classmethod
|
||||||
|
|||||||
@@ -82,6 +82,14 @@ def _safe_file_name(value: str) -> str:
|
|||||||
return name
|
return name
|
||||||
|
|
||||||
|
|
||||||
|
def _safe_path_segment(segment: str) -> str:
|
||||||
|
"""Single path segment: keep alnum / dash / underscore / dot; collapse the rest."""
|
||||||
|
cleaned = "".join(
|
||||||
|
c if c.isalnum() or c in "-_." else "_" for c in segment
|
||||||
|
).strip("._")
|
||||||
|
return cleaned or "untitled"
|
||||||
|
|
||||||
|
|
||||||
def _utcnow_naive() -> Any:
|
def _utcnow_naive() -> Any:
|
||||||
from datetime import datetime, UTC
|
from datetime import datetime, UTC
|
||||||
return datetime.now(UTC).replace(tzinfo=None)
|
return datetime.now(UTC).replace(tzinfo=None)
|
||||||
@@ -92,6 +100,41 @@ def _hash_bytes(value: str) -> bytes:
|
|||||||
return _h.sha256(value.encode("utf-8")).digest()
|
return _h.sha256(value.encode("utf-8")).digest()
|
||||||
|
|
||||||
|
|
||||||
|
async def _resolve_unique_object_key(
|
||||||
|
session: AsyncSession,
|
||||||
|
object_key: str,
|
||||||
|
) -> str:
|
||||||
|
"""Append a ULID suffix when an available object already occupies ``object_key``.
|
||||||
|
|
||||||
|
Keeps the original file name in metadata; only the on-disk key gets
|
||||||
|
the disambiguation suffix so repeated uploads of the same name never
|
||||||
|
overwrite existing bytes.
|
||||||
|
"""
|
||||||
|
key = object_key
|
||||||
|
for _ in range(5):
|
||||||
|
key_hash = _hash_bytes(key)
|
||||||
|
existing = await session.scalar(
|
||||||
|
select(StorageObjects).where(
|
||||||
|
StorageObjects.object_key_hash == key_hash,
|
||||||
|
StorageObjects.object_status == "available",
|
||||||
|
)
|
||||||
|
)
|
||||||
|
if existing is None:
|
||||||
|
return key
|
||||||
|
suffix = new_ulid().lower()
|
||||||
|
parts = key.rsplit("/", 1)
|
||||||
|
dir_part = parts[0] if len(parts) > 1 else ""
|
||||||
|
name = parts[-1]
|
||||||
|
# Preserve extension if present (ignore leading dot / hidden files).
|
||||||
|
if "." in name[1:]:
|
||||||
|
stem, ext = name.rsplit(".", 1)
|
||||||
|
new_name = f"{stem}-{suffix}.{ext}"
|
||||||
|
else:
|
||||||
|
new_name = f"{name}-{suffix}"
|
||||||
|
key = f"{dir_part}/{new_name}" if dir_part else new_name
|
||||||
|
return key
|
||||||
|
|
||||||
|
|
||||||
def _build_storage_object(
|
def _build_storage_object(
|
||||||
*,
|
*,
|
||||||
upload: UploadSessions,
|
upload: UploadSessions,
|
||||||
@@ -192,15 +235,25 @@ async def create_upload_record(
|
|||||||
payload.usage_type,
|
payload.usage_type,
|
||||||
workspace_artifact_bucket=workspace.artifact_bucket,
|
workspace_artifact_bucket=workspace.artifact_bucket,
|
||||||
)
|
)
|
||||||
# Use the sanitized file_name as the on-disk basename (no ULID). The
|
|
||||||
# extension is still derived for StorageObjects.file_extension.
|
|
||||||
safe_name = _safe_file_name(payload.file_name)
|
safe_name = _safe_file_name(payload.file_name)
|
||||||
file_extension = PurePosixPath(safe_name).suffix.lower()
|
# Defense-in-depth: schemas already validate target_path, but the
|
||||||
# Use the sanitized file_name as the on-disk basename (no ULID).
|
# helper is called directly from some callers so re-validate here.
|
||||||
# DataResources files live under a per-user ``.resources`` subdir
|
target_path = (payload.target_path or "").strip("/")
|
||||||
# so they don't collide with workspace-tree scripts/directories
|
if target_path and (target_path.startswith("/") or "\\" in target_path
|
||||||
# the user authors directly in Jupyter.
|
or any(seg == ".." for seg in target_path.split("/"))
|
||||||
object_key = f"{payload.workspace_id}/{payload.user_id}/.resources/{safe_name}"
|
or any(ord(c) < 0x20 or ord(c) == 0x7F for c in target_path)):
|
||||||
|
raise HTTPException(
|
||||||
|
status.HTTP_422_UNPROCESSABLE_ENTITY,
|
||||||
|
"target_path must be a clean relative POSIX path",
|
||||||
|
)
|
||||||
|
clean_target = "/".join(
|
||||||
|
_safe_path_segment(seg) for seg in target_path.split("/") if seg
|
||||||
|
)
|
||||||
|
if clean_target:
|
||||||
|
object_key = f"{payload.workspace_id}/{payload.user_id}/{clean_target}/{safe_name}"
|
||||||
|
else:
|
||||||
|
object_key = f"{payload.workspace_id}/{payload.user_id}/{safe_name}"
|
||||||
|
object_key = await _resolve_unique_object_key(session, object_key)
|
||||||
upload = UploadSessions(
|
upload = UploadSessions(
|
||||||
upload_id=new_ulid(),
|
upload_id=new_ulid(),
|
||||||
workspace_id=payload.workspace_id,
|
workspace_id=payload.workspace_id,
|
||||||
|
|||||||
Reference in New Issue
Block a user