"""Pydantic schemas for REST and services."""
from datetime import datetime
from enum import StrEnum
from typing import Generic, TypeVar
from uuid import UUID
from pydantic import BaseModel, ConfigDict, Field, field_validator
T = TypeVar("T")
class SecretStatus(StrEnum):
actual = "actual"
outdated = "outdated"
class Channel(StrEnum):
ui = "ui"
rest = "rest"
mcp = "mcp"
class Scope(StrEnum):
read = "read"
reveal = "reveal"
write = "write"
admin = "admin"
mcp = "mcp"
class SecretFieldIn(BaseModel):
name: str = Field(min_length=1, max_length=120)
value: str = Field(max_length=65536)
encrypted: bool = False
masked: bool = False
position: int = 0
class SecretFieldOut(BaseModel):
name: str
value: str | None = None
encrypted: bool
masked: bool
position: int
class SecretCreate(BaseModel):
title: str = Field(min_length=1, max_length=255)
purpose: str | None = Field(default=None, max_length=255)
category: str | None = Field(default=None, max_length=120)
source: str | None = Field(default=None, max_length=255)
notes: str | None = Field(default=None, max_length=140)
tags: list[str] = Field(default_factory=list)
status: SecretStatus = SecretStatus.actual
archived: bool = False
allow_ui: bool = True
allow_rest_api: bool = True
allow_mcp: bool = False
fields: list[SecretFieldIn] = Field(default_factory=list)
@field_validator("tags")
@classmethod
def normalize_tags(cls, tags: list[str]) -> list[str]:
result = []
for tag in tags:
value = tag.strip().lower()
if value and value not in result:
result.append(value[:80])
return result
class SecretUpdate(BaseModel):
title: str | None = Field(default=None, min_length=1, max_length=255)
purpose: str | None = Field(default=None, max_length=255)
category: str | None = Field(default=None, max_length=120)
source: str | None = Field(default=None, max_length=255)
notes: str | None = Field(default=None, max_length=140)
tags: list[str] | None = None
status: SecretStatus | None = None
archived: bool | None = None
allow_ui: bool | None = None
allow_rest_api: bool | None = None
allow_mcp: bool | None = None
fields: list[SecretFieldIn] | None = None
class SecretRead(BaseModel):
id: UUID
title: str
purpose: str | None
category: str | None
source: str | None
notes: str | None
tags: list[str]
status: SecretStatus
archived: bool
allow_ui: bool
allow_rest_api: bool
allow_mcp: bool
created_at: datetime
updated_at: datetime
fields: list[SecretFieldOut]
class SecretReveal(SecretRead):
version_id: UUID
version_number: int
class SecretVersionRead(BaseModel):
id: UUID
version_number: int
created_at: datetime
fields: list[SecretFieldOut]
class Page(BaseModel, Generic[T]):
items: list[T]
total: int
offset: int
limit: int
class ApiTokenCreate(BaseModel):
name: str = Field(min_length=1, max_length=120)
scopes: list[Scope]
class ApiTokenRead(BaseModel):
id: UUID
public_id: str
name: str
scopes: list[str]
created_at: datetime
revoked_at: datetime | None
last_used_at: datetime | None
class ApiTokenCreated(ApiTokenRead):
token: str
class AuditEventRead(BaseModel):
model_config = ConfigDict(from_attributes=True)
id: UUID
user_id: UUID | None
actor_user_id: UUID | None
api_token_id: UUID | None
secret_id: UUID | None
channel: str
action: str
ip_address: str | None
user_agent: str | None
audit_metadata: dict = Field(serialization_alias="metadata")
created_at: datetime
class UserRead(BaseModel):
id: UUID
email: str
display_name: str | None
locale: str | None
role: str
status: str
avatar_url: str | None = None
auth_profile_url: str | None = None
class UserUpdate(BaseModel):
display_name: str | None = Field(default=None, max_length=120)
locale: str | None = Field(default=None, max_length=10)
class ExportResponse(BaseModel):
format: str
version: int
exported_at: datetime
secrets: list[dict]
class ImportPayload(BaseModel):
format: str
version: int
exported_at: datetime
secrets: list[SecretCreate]
class StatsRead(BaseModel):
total_secrets: int
active_secrets: int
mcp_enabled_secrets: int