187 lines
4.4 KiB
Python
187 lines
4.4 KiB
Python
from __future__ import annotations
|
|
|
|
from typing import Any, Literal
|
|
|
|
from pydantic import BaseModel, Field
|
|
|
|
from app.storage.common import FileConflictResolution
|
|
|
|
|
|
class FileSpaceResponse(BaseModel):
|
|
id: str
|
|
label: str
|
|
owner_type: Literal["user", "group"]
|
|
owner_id: str
|
|
description: str | None = None
|
|
|
|
|
|
class FileSpacesResponse(BaseModel):
|
|
spaces: list[FileSpaceResponse]
|
|
|
|
|
|
class FileShareResponse(BaseModel):
|
|
id: str
|
|
target_type: str
|
|
target_id: str
|
|
permission: str
|
|
created_at: str
|
|
revoked_at: str | None = None
|
|
|
|
|
|
class FileAssetResponse(BaseModel):
|
|
id: str
|
|
tenant_id: str
|
|
owner_type: str
|
|
owner_id: str
|
|
display_path: str
|
|
filename: str
|
|
description: str | None = None
|
|
size_bytes: int
|
|
content_type: str | None = None
|
|
checksum_sha256: str
|
|
version_id: str
|
|
created_at: str
|
|
updated_at: str
|
|
deleted_at: str | None = None
|
|
audit_relevant: bool = False
|
|
metadata: dict[str, Any] | None = None
|
|
shares: list[FileShareResponse] = Field(default_factory=list)
|
|
|
|
|
|
class FileFolderResponse(BaseModel):
|
|
id: str
|
|
tenant_id: str
|
|
owner_type: str
|
|
owner_id: str
|
|
path: str
|
|
created_at: str
|
|
updated_at: str
|
|
deleted_at: str | None = None
|
|
|
|
|
|
class FileFoldersResponse(BaseModel):
|
|
folders: list[FileFolderResponse]
|
|
|
|
|
|
class FileFolderCreateRequest(BaseModel):
|
|
owner_type: Literal["user", "group"]
|
|
owner_id: str
|
|
path: str
|
|
|
|
|
|
class FileFolderDeleteRequest(BaseModel):
|
|
owner_type: Literal["user", "group"]
|
|
owner_id: str
|
|
path: str
|
|
recursive: bool = True
|
|
|
|
|
|
class FileFolderDeleteResponse(BaseModel):
|
|
deleted_folders: int
|
|
deleted_files: int
|
|
|
|
|
|
class FileListResponse(BaseModel):
|
|
files: list[FileAssetResponse]
|
|
|
|
|
|
class FileUploadResponse(BaseModel):
|
|
files: list[FileAssetResponse]
|
|
|
|
|
|
class BulkDeleteRequest(BaseModel):
|
|
file_ids: list[str]
|
|
|
|
|
|
class BulkDeleteResponse(BaseModel):
|
|
deleted_count: int
|
|
|
|
|
|
class ConflictResolutionRequest(BaseModel):
|
|
target_path: str
|
|
action: Literal["overwrite", "rename", "skip"]
|
|
new_path: str | None = None
|
|
|
|
|
|
def _conflict_resolutions(items: list[ConflictResolutionRequest] | None) -> list[FileConflictResolution]:
|
|
return [FileConflictResolution(target_path=item.target_path, action=item.action, new_path=item.new_path) for item in items or []]
|
|
|
|
|
|
class FileShareRequest(BaseModel):
|
|
target_type: Literal["user", "group", "campaign", "tenant"]
|
|
target_id: str
|
|
permission: Literal["read", "write", "manage"] = "read"
|
|
|
|
|
|
class RenameRequest(BaseModel):
|
|
file_ids: list[str] = Field(default_factory=list)
|
|
folder_paths: list[str] = Field(default_factory=list)
|
|
owner_type: Literal["user", "group"] | None = None
|
|
owner_id: str | None = None
|
|
mode: Literal["direct", "prefix", "suffix", "replace"]
|
|
new_name: str | None = None
|
|
find: str | None = None
|
|
replacement: str = ""
|
|
prefix: str = ""
|
|
suffix: str = ""
|
|
recursive: bool = False
|
|
dry_run: bool = True
|
|
|
|
|
|
class RenamePreviewItem(BaseModel):
|
|
kind: Literal["file", "folder"]
|
|
id: str
|
|
file_id: str | None = None
|
|
folder_path: str | None = None
|
|
old_path: str
|
|
new_path: str
|
|
|
|
|
|
class RenameResponse(BaseModel):
|
|
dry_run: bool
|
|
items: list[RenamePreviewItem]
|
|
|
|
|
|
class TransferRequest(BaseModel):
|
|
operation: Literal["move", "copy"]
|
|
file_ids: list[str] = Field(default_factory=list)
|
|
folder_paths: list[str] = Field(default_factory=list)
|
|
source_owner_type: Literal["user", "group"]
|
|
source_owner_id: str
|
|
target_owner_type: Literal["user", "group"]
|
|
target_owner_id: str
|
|
target_folder: str = ""
|
|
conflict_strategy: Literal["reject", "overwrite", "rename"] = "reject"
|
|
conflict_resolutions: list[ConflictResolutionRequest] = Field(default_factory=list)
|
|
|
|
|
|
class TransferResponse(BaseModel):
|
|
operation: str
|
|
files: int
|
|
folders: int
|
|
|
|
|
|
class ArchiveRequest(BaseModel):
|
|
file_ids: list[str]
|
|
filename: str = "files.zip"
|
|
|
|
|
|
class PatternResolveRequest(BaseModel):
|
|
patterns: list[str]
|
|
owner_type: Literal["user", "group"] | None = None
|
|
owner_id: str | None = None
|
|
campaign_id: str | None = None
|
|
path_prefix: str | None = None
|
|
include_unmatched: bool = True
|
|
case_sensitive: bool = False
|
|
|
|
|
|
class PatternMatchResponse(BaseModel):
|
|
pattern: str
|
|
matches: list[FileAssetResponse]
|
|
|
|
|
|
class PatternResolveResponse(BaseModel):
|
|
patterns: list[PatternMatchResponse]
|
|
unmatched: list[FileAssetResponse] = Field(default_factory=list)
|