Files
kite/backend/api/cloud/schemas.py
T
curo1305andClaude Sonnet 4.6 b6911fb4ed fix(12): resolve critical code review findings (CR-01 through CR-04)
- CR-01: fix browse URL from /folders/{id} to /items?parent_ref= (broken feature)
- CR-02: remove raw provider exception text from _TerminalProviderError messages
- CR-03: add user_id predicate to get_or_create_folder_state query (ownership boundary)
- CR-04: add max_length=255 to ConnectionRenameRequest.display_name

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-06-19 02:00:05 +02:00

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, 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(..., max_length=255)
@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