curo1305
e186019066
- Add api/cloud/ package: connections.py, browse.py, schemas.py, __init__.py
- Add GET /api/cloud/connections/{connection_id}/items (T-12-01 IDOR, T-12-03 cred-free)
- Add PATCH /api/cloud/connections/{id} for display_name_override rename
- Add display_name_override ORM field to CloudConnection model
- Add CloudResourceAdapter service layer with str/UUID coercion
- Fix UUID type compatibility: test_cloud_items.py now uses UUID(as_uuid=True)
matching conftest — removes String(36) patch that caused type incompatibility
- Add 11 Phase 12 integration tests (IDOR, admin block, credential exclusion,
duplicate providers, rename, malformed UUID)
- Remove deleted api/cloud.py (replaced by api/cloud/ package)
87 lines
3.0 KiB
Python
87 lines
3.0 KiB
Python
"""
|
|
Whitelisted Pydantic response schemas for the cloud API package.
|
|
|
|
All schemas are explicit allowlists — credentials_enc, tokens, and passwords
|
|
are deliberately absent. T-12-03: credential exclusion by design.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
from datetime import datetime
|
|
from typing import List, Optional
|
|
|
|
from pydantic import BaseModel, field_validator
|
|
|
|
|
|
# ── Capability / item schemas ─────────────────────────────────────────────────
|
|
|
|
class CloudCapabilityOut(BaseModel):
|
|
"""Whitelisted capability descriptor. reason/message only when not supported."""
|
|
|
|
action: str
|
|
state: str # "supported" | "unsupported" | "temporarily_unavailable"
|
|
reason: Optional[str] = None
|
|
message: Optional[str] = None
|
|
|
|
|
|
class CloudItemOut(BaseModel):
|
|
"""Normalized cloud item metadata. No credentials or byte content."""
|
|
|
|
id: str # DocuVault stable UUID
|
|
provider_item_id: str
|
|
name: str
|
|
kind: str # "file" | "folder"
|
|
parent_ref: Optional[str] = None
|
|
content_type: Optional[str] = None
|
|
size: Optional[int] = None
|
|
modified_at: Optional[datetime] = None
|
|
etag: Optional[str] = None
|
|
capabilities: dict[str, CloudCapabilityOut] = {}
|
|
|
|
|
|
# ── Freshness / folder state schemas ─────────────────────────────────────────
|
|
|
|
class FolderFreshnessOut(BaseModel):
|
|
"""Freshness and error state for a browsed folder."""
|
|
|
|
refresh_state: str # "fresh" | "refreshing" | "warning"
|
|
last_refreshed_at: Optional[datetime] = None
|
|
error_code: Optional[str] = None
|
|
error_message: Optional[str] = None
|
|
|
|
|
|
# ── Browse response ───────────────────────────────────────────────────────────
|
|
|
|
class CloudBrowseResponse(BaseModel):
|
|
"""Owner-scoped connection-ID browse response.
|
|
|
|
T-12-01: items are always scoped to the resolved connection which is owned
|
|
by the requesting user. credentials_enc is never included.
|
|
"""
|
|
|
|
connection_id: str
|
|
provider: str
|
|
display_name: str
|
|
parent_ref: Optional[str]
|
|
items: List[CloudItemOut]
|
|
capabilities: dict[str, CloudCapabilityOut]
|
|
freshness: FolderFreshnessOut
|
|
|
|
|
|
# ── Connection schemas ────────────────────────────────────────────────────────
|
|
|
|
class ConnectionRenameRequest(BaseModel):
|
|
"""Validated PATCH body for renaming a connection display name.
|
|
|
|
Only display_name is accepted — mass assignment prevention.
|
|
"""
|
|
|
|
display_name: str
|
|
|
|
@field_validator("display_name")
|
|
@classmethod
|
|
def must_be_nonblank(cls, v: str) -> str:
|
|
stripped = v.strip()
|
|
if not stripped:
|
|
raise ValueError("display_name must not be blank")
|
|
return stripped
|