File size: 5,121 Bytes
c72325f | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 | """Ownership-safe persistence helpers for source-document object metadata."""
from __future__ import annotations
from dataclasses import dataclass
from typing import Optional
from sqlalchemy import select
from backend.app.storage.base import document_pdf_object_key, validate_object_key
from persistence.ownership import require_owned_document
class DocumentObjectMetadataError(ValueError):
"""Raised when document object metadata is invalid or internally inconsistent."""
@dataclass(frozen=True)
class StoredDocumentObjectMetadata:
document_id: str
user_id: str
object_key: str
content_type: str
byte_size: int
def _validate_metadata(
*,
document_id: str,
object_key: str,
content_type: str,
byte_size: int,
) -> tuple[str, str, int]:
validated_key = validate_object_key(object_key)
expected_key = document_pdf_object_key(document_id)
if validated_key != expected_key:
raise DocumentObjectMetadataError(
"Document object key does not match the document's generated storage key."
)
normalized_content_type = content_type.strip() if isinstance(content_type, str) else ""
if not normalized_content_type or len(normalized_content_type) > 255:
raise DocumentObjectMetadataError("Document content type is invalid.")
if isinstance(byte_size, bool) or not isinstance(byte_size, int) or byte_size <= 0:
raise DocumentObjectMetadataError("Document byte size must be a positive integer.")
return validated_key, normalized_content_type, byte_size
def document_object_metadata_from_row(document: object) -> Optional[StoredDocumentObjectMetadata]:
"""Validate and project object metadata from an already-loaded document row."""
values = (
getattr(document, "object_key", None),
getattr(document, "content_type", None),
getattr(document, "byte_size", None),
)
if values == (None, None, None):
return None
if any(value is None for value in values):
raise DocumentObjectMetadataError("Document object metadata is incomplete.")
object_key, content_type, byte_size = _validate_metadata(
document_id=str(getattr(document, "id")),
object_key=str(values[0]),
content_type=str(values[1]),
byte_size=values[2],
)
return StoredDocumentObjectMetadata(
document_id=str(getattr(document, "id")),
user_id=str(getattr(document, "user_id")),
object_key=object_key,
content_type=content_type,
byte_size=byte_size,
)
def attach_owned_document_object_metadata(
*,
user_id: str,
document_id: str,
object_key: str,
content_type: str,
byte_size: int,
) -> StoredDocumentObjectMetadata:
"""Attach generated object metadata after enforcing document ownership."""
from persistence.db import SessionLocal
with SessionLocal() as session:
try:
document = require_owned_document(
session,
user_id=user_id,
document_id=document_id,
)
validated_key, normalized_content_type, validated_byte_size = _validate_metadata(
document_id=str(document.id),
object_key=object_key,
content_type=content_type,
byte_size=byte_size,
)
document.object_key = validated_key
document.content_type = normalized_content_type
document.byte_size = validated_byte_size
session.commit()
return StoredDocumentObjectMetadata(
document_id=str(document.id),
user_id=str(document.user_id),
object_key=validated_key,
content_type=normalized_content_type,
byte_size=validated_byte_size,
)
except Exception:
session.rollback()
raise
def get_owned_document_object_metadata(
*,
user_id: str,
document_id: str,
) -> Optional[StoredDocumentObjectMetadata]:
"""Load metadata only after resolving a document in the user's namespace."""
from persistence.db import SessionLocal
with SessionLocal() as session:
document = require_owned_document(
session,
user_id=user_id,
document_id=document_id,
)
return document_object_metadata_from_row(document)
def get_document_object_metadata(
*,
document_id: str,
) -> Optional[StoredDocumentObjectMetadata]:
"""Internal lookup for a future worker that receives only a document ID."""
from persistence.db import SessionLocal
from persistence.models import Document
if not isinstance(document_id, str) or not document_id.strip():
raise DocumentObjectMetadataError("Document ID must not be empty.")
with SessionLocal() as session:
document = session.execute(
select(Document).where(Document.id == document_id)
).scalar_one_or_none()
if document is None:
return None
return document_object_metadata_from_row(document)
|