feat(core): define managed tabular file contract
Module Package Release / publish-packages (push) Successful in 12s

This commit is contained in:
2026-08-21 19:29:48 +02:00
parent f75ad48d78
commit 142ccbc587
7 changed files with 216 additions and 7 deletions
+119
View File
@@ -2,6 +2,7 @@ from __future__ import annotations
from collections.abc import Mapping
from dataclasses import dataclass, field
from datetime import datetime
from typing import Protocol, runtime_checkable
from govoplan_core.core.access import ResourceAccessExplanationProvider
@@ -10,6 +11,48 @@ from govoplan_core.core.access import ResourceAccessExplanationProvider
CAPABILITY_FILES_ACCESS = "files.access"
CAPABILITY_FILES_ARTIFACT_STORE = "files.artifact_store"
CAPABILITY_FILES_POSTBOX_REFERENCES = "files.postbox_references"
CAPABILITY_FILES_TABULAR_CONTENT = "files.tabular_content"
class ManagedTabularFileError(ValueError):
"""Stable base error for exact-version managed tabular file access."""
class ManagedTabularFileNotFoundError(ManagedTabularFileError):
pass
class ManagedTabularFileAccessError(ManagedTabularFileError):
pass
class ManagedTabularFileUnavailableError(ManagedTabularFileError):
pass
class ManagedTabularFileValidationError(ManagedTabularFileError):
pass
@dataclass(frozen=True, slots=True)
class ManagedTabularFile:
"""Authorized metadata for one immutable managed file version."""
file_asset_id: str
file_version_id: str
filename: str
display_path: str
content_type: str | None
size_bytes: int
sha256: str
updated_at: datetime | None = None
current_version: bool = True
@dataclass(frozen=True, slots=True)
class ManagedTabularFileContent:
file: ManagedTabularFile
payload: bytes
@dataclass(frozen=True, slots=True)
@@ -91,6 +134,39 @@ class PostboxFileReferenceProvider(Protocol):
) -> tuple[PostboxFileReferenceRef, ...]: ...
@runtime_checkable
class ManagedTabularFileProvider(Protocol):
"""List and open authorized CSV/XLSX content without exposing Files internals."""
def list_tabular_files(
self,
session: object,
principal: object,
*,
query: str = "",
limit: int = 100,
) -> tuple[ManagedTabularFile, ...]: ...
def get_tabular_file(
self,
session: object,
principal: object,
*,
file_asset_id: str,
file_version_id: str | None = None,
) -> ManagedTabularFile | None: ...
def read_tabular_file(
self,
session: object,
principal: object,
*,
file_asset_id: str,
file_version_id: str,
max_bytes: int,
) -> ManagedTabularFileContent: ...
def postbox_file_reference_provider(
registry: object | None,
) -> PostboxFileReferenceProvider | None:
@@ -107,3 +183,46 @@ def postbox_file_reference_provider(
"PostboxFileReferenceProvider"
)
return provider
def managed_tabular_file_provider(
registry: object | None,
) -> ManagedTabularFileProvider | None:
if (
registry is None
or not hasattr(registry, "has_capability")
or not registry.has_capability(CAPABILITY_FILES_TABULAR_CONTENT)
):
return None
provider = registry.require_capability(CAPABILITY_FILES_TABULAR_CONTENT)
if not isinstance(provider, ManagedTabularFileProvider):
raise TypeError(
"files.tabular_content provider does not implement "
"ManagedTabularFileProvider"
)
return provider
__all__ = [
"CAPABILITY_FILES_ACCESS",
"CAPABILITY_FILES_ARTIFACT_STORE",
"CAPABILITY_FILES_POSTBOX_REFERENCES",
"CAPABILITY_FILES_TABULAR_CONTENT",
"FileAccessProvider",
"ManagedArtifactRef",
"ManagedArtifactStore",
"ManagedArtifactWriteRequest",
"ManagedTabularFile",
"ManagedTabularFileAccessError",
"ManagedTabularFileContent",
"ManagedTabularFileError",
"ManagedTabularFileNotFoundError",
"ManagedTabularFileProvider",
"ManagedTabularFileUnavailableError",
"ManagedTabularFileValidationError",
"PostboxFileReferenceProvider",
"PostboxFileReferenceRef",
"PostboxFileReferenceRequest",
"managed_tabular_file_provider",
"postbox_file_reference_provider",
]