Files
ObsiGate/backend/schemas.py
T
bruno f049e208b6
CI / lint (push) Successful in 1m10s
CI / security (push) Successful in 43s
CI / test (push) Successful in 2m34s
CI / build (push) Successful in 43s
CI / e2e (push) Successful in 10m59s
feat(ai): commandes @/ & skills, analyse d'images et capacites des modeles (#81)
2026-09-12 11:38:46 -04:00

518 lines
16 KiB
Python

"""Additional Pydantic response models for the ObsiGate REST API.
These models enrich the auto-generated OpenAPI 3.1 schema (#72). They are
intentionally permissive (``extra="allow"``) so that adding a new field to an
existing endpoint never breaks response validation — the model documents the
stable, public shape while still accepting internal additions.
Models whose endpoints are already typed in :mod:`backend.main` (e.g.
``FileContentResponse``) live there; this module only holds the ones used to
fill the gaps.
"""
from typing import Any
from pydantic import BaseModel, ConfigDict, Field
# ---------------------------------------------------------------------------
# Generic helpers
# ---------------------------------------------------------------------------
class StatusResponse(BaseModel):
"""Generic ``{"status": "..."}`` acknowledgement."""
model_config = ConfigDict(extra="allow")
status: str = Field(default="ok", description="Operation status")
class ErrorResponse(BaseModel):
"""Standard FastAPI error envelope."""
detail: str | list[dict[str, Any]] | None = Field(
default=None,
description="Human-readable error message, or a list of validation errors",
examples=["Vault 'Notes' not found"],
)
# ---------------------------------------------------------------------------
# Recent files & bookmarks
# ---------------------------------------------------------------------------
class RecentFileItem(BaseModel):
"""A recently opened or recently modified file."""
path: str = Field(description="Relative path within the vault")
title: str = Field(description="File title")
vault: str = Field(description="Vault name")
mtime: float | int | str | None = Field(default=None, description="Modification timestamp")
mtime_human: str | None = Field(default=None, description="Human-readable modification time")
mtime_iso: str | None = Field(default=None, description="ISO-8601 modification time")
size_bytes: int | None = Field(default=None, description="File size in bytes")
tags: list[str] = Field(default_factory=list, description="Up to 5 leading tags")
preview: str | None = Field(default=None, description="Short content preview")
bookmarked: bool | None = Field(default=None, description="Whether the file is bookmarked")
class RecentResponse(BaseModel):
"""Response for ``GET /api/recent``."""
files: list[RecentFileItem]
total: int = Field(description="Total number of files in the selected mode")
limit: int = Field(description="Applied limit")
mode: str = Field(description="'opened' or 'modified'")
model_config = ConfigDict(
json_schema_extra={
"example": {
"files": [
{
"path": "notes/Accueil.md",
"title": "Accueil",
"vault": "TestVault",
"mtime": 1750000000.0,
"mtime_human": "il y a 2 h",
"size_bytes": 1024,
"tags": ["#accueil"],
"preview": "# Bienvenue…",
"bookmarked": False,
}
],
"total": 1,
"limit": 20,
"mode": "opened",
}
}
)
class BookmarkFileItem(BaseModel):
"""A bookmarked file."""
path: str = Field(description="Relative path within the vault")
title: str = Field(description="File title")
vault: str = Field(description="Vault name")
mtime: float | int | str | None = Field(default=None, description="Bookmark timestamp")
mtime_human: str | None = Field(default=None, description="Human-readable bookmark time")
size_bytes: int | None = Field(default=None, description="File size in bytes")
tags: list[str] = Field(default_factory=list, description="Up to 5 leading tags")
bookmarked: bool | None = Field(default=True, description="Always true for this endpoint")
class BookmarksResponse(BaseModel):
"""Response for ``GET /api/bookmarks``."""
files: list[BookmarkFileItem]
total: int = Field(description="Number of bookmarked files")
class BookmarkToggleResponse(BaseModel):
"""Response for ``POST /api/bookmarks/toggle``."""
bookmarked: bool = Field(description="New bookmark state")
class SavedSearch(BaseModel):
"""A persisted search definition."""
model_config = ConfigDict(extra="allow")
id: str = Field(description="Unique identifier (millisecond timestamp)")
query: str = Field(description="Search query text")
vault: str = Field(default="all", description="Vault filter")
case_sensitive: bool = Field(default=False)
whole_word: bool = Field(default=False)
regex: bool = Field(default=False)
include_paths: str = Field(default="")
exclude_paths: str = Field(default="")
created_at: float = Field(description="Creation timestamp (epoch seconds)")
# ---------------------------------------------------------------------------
# Backups & diffs
# ---------------------------------------------------------------------------
class BackupsResponse(BaseModel):
"""Response for ``GET /api/file/{vault}/backups``."""
vault: str
path: str
backups: list[dict[str, Any]] = Field(description="Backups, newest first")
class BacklinksResponse(BaseModel):
"""Response for ``GET /api/file/{vault}/backlinks``."""
vault: str
path: str
backlinks: list[dict[str, Any]] = Field(description="Files linking to the target")
total: int
class BackupsListResponse(BaseModel):
"""Response for ``GET /api/backups``."""
backups: list[dict[str, Any]] = Field(description="All backups across vaults")
total: int
total_size_bytes: int
class BackupsDeletedResponse(BaseModel):
"""Response for backup deletion / purge endpoints."""
deleted: int = Field(description="Number of backup files deleted")
class BackupContentResponse(BaseModel):
"""Response for ``GET /api/backups/content``."""
content: str = Field(description="Backup file content (truncated to 100 KB)")
name: str = Field(description="Backup file name")
size: int = Field(description="Returned content size in characters")
class BackupsCompressResponse(BaseModel):
"""Response for ``POST /api/backups/compress``."""
compressed: int = Field(description="Number of backups processed")
saved_bytes: int = Field(description="Bytes saved by gzip compression")
dry_run: bool
class BackupsAutoResponse(BaseModel):
"""Response for ``POST /api/backups/auto``."""
backed_up: int = Field(description="Number of files backed up")
since_hours: int | float = Field(description="Look-back window in hours")
# ---------------------------------------------------------------------------
# PDF
# ---------------------------------------------------------------------------
class PdfInfoResponse(BaseModel):
"""Response for ``GET /api/file/{vault}/pdf/info``."""
vault: str
path: str
pages: int = Field(description="Page count")
title: str = Field(description="PDF title (metadata or filename)")
author: str = Field(default="", description="PDF author")
size_bytes: int = Field(description="File size in bytes")
model_config = ConfigDict(
json_schema_extra={
"example": {
"vault": "TestVault",
"path": "docs/rapport.pdf",
"pages": 12,
"title": "Rapport annuel",
"author": "ObsiGate",
"size_bytes": 524288,
}
}
)
# ---------------------------------------------------------------------------
# Search & replace
# ---------------------------------------------------------------------------
class ReplaceMatch(BaseModel):
"""A single file affected by a find/replace operation."""
model_config = ConfigDict(extra="allow")
vault: str
path: str
title: str | None = None
match_count: int | None = Field(default=None, description="Occurrences found (dry run)")
replacements: int | None = Field(default=None, description="Occurrences replaced")
preview: list[str] = Field(default_factory=list, description="Context snippets (dry run)")
class ReplaceResponse(BaseModel):
"""Response for ``POST /api/search/replace``."""
model_config = ConfigDict(extra="allow")
matches: list[ReplaceMatch] = Field(default_factory=list, description="Dry-run matches")
replaced: list[ReplaceMatch] = Field(default_factory=list, description="Applied replacements")
total_matches: int | None = Field(default=None, description="Total matches (dry run)")
total_replacements: int = Field(description="Total replacements performed or previewed")
dry_run: bool | None = Field(default=None, description="True when no file was written")
# ---------------------------------------------------------------------------
# Vaults, attachments & settings
# ---------------------------------------------------------------------------
class VaultStatsResponse(BaseModel):
"""Response for ``POST /api/vaults/add`` and ``GET /api/index/reload/{vault}``."""
model_config = ConfigDict(extra="allow")
status: str = Field(default="ok")
vault: str
stats: dict[str, Any] = Field(default_factory=dict, description="Index statistics for the vault")
class VaultActionResponse(BaseModel):
"""Response for ``DELETE /api/vaults/{vault}``."""
status: str = Field(default="ok")
vault: str
class VaultStatusEntry(BaseModel):
"""Per-vault status entry."""
file_count: int
tag_count: int
path: str = ""
watching: bool = False
class VaultsStatusResponse(BaseModel):
"""Response for ``GET /api/vaults/status``."""
vaults: dict[str, VaultStatusEntry]
watcher_active: bool
sse_clients: int
class AttachmentRescanResponse(BaseModel):
"""Response for ``POST /api/attachments/rescan/{vault}``."""
status: str = Field(default="ok")
vault: str
attachment_count: int
class AttachmentStatsResponse(BaseModel):
"""Response for ``GET /api/attachments/stats``."""
vaults: dict[str, Any] = Field(description="Vault name → attachment statistics")
class VaultSettingsResponse(BaseModel):
"""Response for the per-vault display settings endpoints."""
model_config = ConfigDict(extra="allow")
hideHiddenFiles: bool = Field(default=False, description="Hide dotfiles in the tree")
class AllVaultSettingsResponse(BaseModel):
"""Response for ``GET /api/vaults/settings/all``."""
model_config = ConfigDict(extra="allow")
class VaultFileEntry(BaseModel):
"""A file entry returned by the vault home listing."""
model_config = ConfigDict(extra="allow")
name: str
path: str
vault: str
size: int = 0
modified: float = 0
modified_iso: str | None = None
extension: str = ""
rel_dir: str | None = None
class VaultFilesResponse(BaseModel):
"""Response for ``GET /api/vault/{vault}/files``."""
vault: str
directory: str = ""
recursive: bool = True
count: int
files: list[VaultFileEntry]
# ---------------------------------------------------------------------------
# Configuration, AI keys & diagnostics
# ---------------------------------------------------------------------------
class AppConfigResponse(BaseModel):
"""Application configuration (``GET/POST /api/config``)."""
model_config = ConfigDict(extra="allow")
class AIKeysResponse(BaseModel):
"""Masked AI provider keys (``GET /api/config/ai-keys``)."""
model_config = ConfigDict(extra="allow")
class AIKeyDeleteResponse(BaseModel):
"""Response for ``DELETE /api/config/ai-keys/{provider}``."""
status: str = Field(default="deleted")
key: str = Field(description="Deleted environment variable name")
class AITestResponse(BaseModel):
"""Response for ``POST /api/config/ai-keys/test`` (provider → status)."""
model_config = ConfigDict(extra="allow")
class AIModelsResponse(BaseModel):
"""Response for ``GET /api/config/ai-models``."""
model_config = ConfigDict(extra="allow")
models: list[str] = Field(default_factory=list, description="Available model identifiers")
source: str = Field(default="fallback", description="'live', 'fallback' or 'validation'")
count: int | None = Field(default=None, description="Number of live models")
error: str | None = Field(default=None, description="Provider/network error, if any")
note: str | None = Field(default=None, description="Explanatory note when using fallback")
capabilities: dict[str, dict[str, bool]] = Field(
default_factory=dict,
description="Per-model capability flags (chat, embeddings, vision, …)",
)
class DiagnosticsResponse(BaseModel):
"""Response for ``GET /api/diagnostics``."""
model_config = ConfigDict(extra="allow")
index: dict[str, Any] = Field(default_factory=dict)
inverted_index: dict[str, Any] = Field(default_factory=dict)
config: dict[str, Any] = Field(default_factory=dict)
class DashboardVaultStat(BaseModel):
"""Per-vault dashboard statistics."""
name: str
file_count: int
tag_count: int
total_size_bytes: int
class DashboardResponse(BaseModel):
"""Response for ``GET /api/dashboard``."""
vaults: list[DashboardVaultStat]
total_files: int
total_tags: int
total_size_bytes: int
# ---------------------------------------------------------------------------
# Webhooks, sharing & conflicts
# ---------------------------------------------------------------------------
class WebhookModel(BaseModel):
"""A configured webhook."""
model_config = ConfigDict(extra="allow")
id: str = Field(description="Webhook UUID")
name: str
url: str = Field(description="Target HTTP(S) URL")
events: list[str] = Field(default_factory=list, description="Subscribed event types")
secret: str | None = Field(default=None, description="HMAC-SHA256 signing secret")
enabled: bool = True
created_at: str | None = None
last_fired_at: str | None = None
class ShareModel(BaseModel):
"""A public document share."""
model_config = ConfigDict(extra="allow")
id: str
token: str = Field(description="Opaque share token")
vault: str
path: str
url: str | None = Field(default=None, description="Relative public URL (/s/{token})")
created_by: str | None = None
created_at: str | None = None
expires_at: str | None = None
access_count: int = 0
last_accessed: str | None = None
class ConflictEntry(BaseModel):
"""A Syncthing sync-conflict file."""
model_config = ConfigDict(extra="allow")
vault: str
path: str
original_path: str | None = None
class ConflictsResponse(BaseModel):
"""Response for ``GET /api/conflicts``."""
conflicts: list[ConflictEntry]
total: int
class ConflictResolveResponse(BaseModel):
"""Response for ``POST /api/conflicts/resolve``."""
status: str = Field(default="resolved")
action: str | None = Field(default=None, description="'keep_local' or 'keep_conflict'")
# ---------------------------------------------------------------------------
# AI status & BooksLM context
# ---------------------------------------------------------------------------
class AIProviderStatus(BaseModel):
"""Availability of a single AI provider."""
available: bool
model: str | None = Field(default=None, description="Default model when the provider is configured")
class AIAutocompleteStatus(BaseModel):
"""Status of the local Ollama autocomplete backend."""
available: bool = False
server_ok: bool = False
model_loaded: bool = False
model: str = ""
error: str | None = None
class AIStatusResponse(BaseModel):
"""Response for ``GET /api/ai/status``."""
configured: bool = Field(description="True when at least one provider has an API key")
default_provider: str
providers: dict[str, AIProviderStatus]
autocomplete: AIAutocompleteStatus
class BooksLMContextFile(BaseModel):
"""A single document included in the BooksLM context."""
path: str
title: str
content: str
type: str = Field(default="markdown", description="'markdown' (and future types)")
class BooksLMContextResponse(BaseModel):
"""Response for ``POST /api/ai/bookslm/context``."""
files: list[BooksLMContextFile]
total_chars: int
file_count: int
directory_tree: str = ""
max_total_chars: int = Field(default=200000, description="Configured context character limit")
max_files: int = Field(default=200, description="Configured file-count limit")
scope: str = Field(default="directory", description="Context scope: 'directory', 'documents' or 'general'")