generated from john/python-template
service updates
This commit is contained in:
@@ -3,8 +3,8 @@ from collections.abc import Sequence
|
|||||||
from dataclasses import dataclass
|
from dataclasses import dataclass
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
from uuid import UUID
|
from uuid import UUID
|
||||||
from uuid import uuid4
|
|
||||||
|
|
||||||
|
from sqlalchemy.exc import IntegrityError
|
||||||
from sqlalchemy.orm import selectinload
|
from sqlalchemy.orm import selectinload
|
||||||
from sqlmodel import select
|
from sqlmodel import select
|
||||||
from sqlmodel.ext.asyncio.session import AsyncSession
|
from sqlmodel.ext.asyncio.session import AsyncSession
|
||||||
@@ -17,11 +17,10 @@ from ..models import Document
|
|||||||
from ..models import Job
|
from ..models import Job
|
||||||
from ..models import JobStatus
|
from ..models import JobStatus
|
||||||
from .base import ServiceBase
|
from .base import ServiceBase
|
||||||
|
from .store import store_file
|
||||||
|
|
||||||
logger = logging.getLogger(__name__)
|
logger = logging.getLogger(__name__)
|
||||||
|
|
||||||
SUPPORTED_UPLOAD_EXTENSIONS = {".jpg", ".jpeg", ".png", ".tif", ".tiff", ".pdf"}
|
|
||||||
|
|
||||||
|
|
||||||
class DocumentError(AppError):
|
class DocumentError(AppError):
|
||||||
"""Raised when document operations fail."""
|
"""Raised when document operations fail."""
|
||||||
@@ -35,6 +34,10 @@ class UploadError(DocumentError):
|
|||||||
"""Raised when uploaded content cannot be persisted safely."""
|
"""Raised when uploaded content cannot be persisted safely."""
|
||||||
|
|
||||||
|
|
||||||
|
class DocumentAlreadyExistsError(DocumentError):
|
||||||
|
"""Raised when a document with the same filename already exists in the database."""
|
||||||
|
|
||||||
|
|
||||||
@dataclass(frozen=True)
|
@dataclass(frozen=True)
|
||||||
class UploadJobResult:
|
class UploadJobResult:
|
||||||
"""Summary of created upload records."""
|
"""Summary of created upload records."""
|
||||||
@@ -48,11 +51,22 @@ class UploadJobResult:
|
|||||||
class DocumentService(ServiceBase):
|
class DocumentService(ServiceBase):
|
||||||
"""Thin service class for managing documents in the database."""
|
"""Thin service class for managing documents in the database."""
|
||||||
|
|
||||||
|
#
|
||||||
|
# CRUD Operations
|
||||||
|
#
|
||||||
|
|
||||||
async def create_document(self, document: Document) -> Document:
|
async def create_document(self, document: Document) -> Document:
|
||||||
"""Create a new document in the database."""
|
"""Create a new document in the database."""
|
||||||
async with self.session_factory() as session:
|
async with self.session_factory() as session:
|
||||||
session.add(document)
|
session.add(document)
|
||||||
await session.commit()
|
try:
|
||||||
|
await session.commit()
|
||||||
|
except IntegrityError as exc:
|
||||||
|
raise DocumentAlreadyExistsError(
|
||||||
|
f"Document with id {document.id} already exists",
|
||||||
|
category=ErrorCategory.VALIDATION,
|
||||||
|
suggestion="Rename the file and try again.",
|
||||||
|
) from exc
|
||||||
await session.refresh(document)
|
await session.refresh(document)
|
||||||
return document
|
return document
|
||||||
|
|
||||||
@@ -95,73 +109,22 @@ class DocumentService(ServiceBase):
|
|||||||
await session.delete(document)
|
await session.delete(document)
|
||||||
await session.commit()
|
await session.commit()
|
||||||
|
|
||||||
|
# Query Operations
|
||||||
|
|
||||||
async def query_documents(self, *, filename: str | None = None) -> Sequence[Document]:
|
async def query_documents(self, *, filename: str | None = None) -> Sequence[Document]:
|
||||||
"""Query documents from the database based on provided filters."""
|
"""Query documents from the database based on provided filters."""
|
||||||
async with self.session_factory() as session:
|
async with self.session_factory() as session:
|
||||||
query = select(Document)
|
query = select(Document)
|
||||||
if filename is not None:
|
if filename is not None:
|
||||||
query = query.where(Document.filename == filename)
|
query = query.where(Document.filename == filename)
|
||||||
return (await session.exec(query)).all()
|
result = await session.exec(query)
|
||||||
|
return result.all()
|
||||||
|
|
||||||
async def list_documents(self) -> Sequence[Document]:
|
async def list_documents(self) -> Sequence[Document]:
|
||||||
"""List all documents in the database."""
|
"""List all documents in the database."""
|
||||||
async with self.session_factory() as session:
|
async with self.session_factory() as session:
|
||||||
return (await session.exec(select(Document))).all()
|
result = await session.exec(select(Document))
|
||||||
|
return result.all()
|
||||||
async def upload_file(
|
|
||||||
self,
|
|
||||||
*,
|
|
||||||
filename: str,
|
|
||||||
file_bytes: bytes,
|
|
||||||
) -> UploadJobResult:
|
|
||||||
"""Persist an uploaded file and create document/job records."""
|
|
||||||
stored_path = store_file(filename=filename, file_bytes=file_bytes, settings=self.settings)
|
|
||||||
try:
|
|
||||||
document, job = await _create_upload_records(
|
|
||||||
session=self.session_factory(),
|
|
||||||
original_filename=filename,
|
|
||||||
stored_path=stored_path,
|
|
||||||
)
|
|
||||||
except Exception as exc:
|
|
||||||
_best_effort_delete(stored_path)
|
|
||||||
raise UploadError(
|
|
||||||
"Failed to create upload database records",
|
|
||||||
category=ErrorCategory.INFRA_TRANSIENT,
|
|
||||||
suggestion="Retry upload. If this keeps happening, verify database availability.",
|
|
||||||
retriable=True,
|
|
||||||
) from exc
|
|
||||||
|
|
||||||
logger.info("Created upload job document_id=%s job_id=%s", document.id, job.id)
|
|
||||||
return UploadJobResult(
|
|
||||||
document_id=document.id,
|
|
||||||
job_id=job.id,
|
|
||||||
stored_path=stored_path,
|
|
||||||
original_filename=Path(filename).name,
|
|
||||||
)
|
|
||||||
|
|
||||||
|
|
||||||
def store_file(*, filename: str, file_bytes: bytes, settings: Settings | None = None) -> Path:
|
|
||||||
"""Persist an uploaded file to the configured upload directory."""
|
|
||||||
runtime_settings = settings or get_settings()
|
|
||||||
_validate_upload(filename=filename, file_bytes=file_bytes)
|
|
||||||
|
|
||||||
upload_dir = runtime_settings.upload_dir
|
|
||||||
upload_dir.mkdir(parents=True, exist_ok=True)
|
|
||||||
|
|
||||||
stored_name = _build_stored_filename(filename)
|
|
||||||
stored_path = upload_dir / stored_name
|
|
||||||
|
|
||||||
try:
|
|
||||||
stored_path.write_bytes(file_bytes)
|
|
||||||
except OSError as exc:
|
|
||||||
raise UploadError(
|
|
||||||
"Failed to persist upload file",
|
|
||||||
category=ErrorCategory.INFRA_PERSISTENT,
|
|
||||||
suggestion="Check upload directory permissions and available disk space, then retry.",
|
|
||||||
) from exc
|
|
||||||
|
|
||||||
logger.info("Stored uploaded file: %s", stored_path)
|
|
||||||
return stored_path
|
|
||||||
|
|
||||||
|
|
||||||
async def create_upload_job(
|
async def create_upload_job(
|
||||||
@@ -169,10 +132,15 @@ async def create_upload_job(
|
|||||||
filename: str,
|
filename: str,
|
||||||
file_bytes: bytes,
|
file_bytes: bytes,
|
||||||
session: AsyncSession,
|
session: AsyncSession,
|
||||||
settings: Settings,
|
settings: Settings | None = None,
|
||||||
) -> UploadJobResult:
|
) -> UploadJobResult:
|
||||||
"""Create upload-backed document and queued job records."""
|
"""Create upload-backed document and queued job records."""
|
||||||
stored_path = store_file(filename=filename, file_bytes=file_bytes, settings=settings)
|
runtime_settings = settings or get_settings()
|
||||||
|
stored_path = store_file(
|
||||||
|
filename=filename,
|
||||||
|
file_bytes=file_bytes,
|
||||||
|
settings=runtime_settings,
|
||||||
|
)
|
||||||
try:
|
try:
|
||||||
document, job = await _create_upload_records(
|
document, job = await _create_upload_records(
|
||||||
session=session,
|
session=session,
|
||||||
@@ -205,7 +173,7 @@ async def _create_upload_records(
|
|||||||
) -> tuple[Document, Job]:
|
) -> tuple[Document, Job]:
|
||||||
document = Document(
|
document = Document(
|
||||||
filename=Path(original_filename).name,
|
filename=Path(original_filename).name,
|
||||||
file_path=stored_path,
|
file_path=str(stored_path),
|
||||||
)
|
)
|
||||||
session.add(document)
|
session.add(document)
|
||||||
await session.flush()
|
await session.flush()
|
||||||
@@ -221,36 +189,6 @@ async def _create_upload_records(
|
|||||||
return document, job
|
return document, job
|
||||||
|
|
||||||
|
|
||||||
def _validate_upload(*, filename: str, file_bytes: bytes) -> None:
|
|
||||||
if not file_bytes:
|
|
||||||
raise UploadError(
|
|
||||||
"Upload payload is empty",
|
|
||||||
category=ErrorCategory.VALIDATION,
|
|
||||||
suggestion="Select a non-empty file and try again.",
|
|
||||||
)
|
|
||||||
|
|
||||||
safe_name = Path(filename).name
|
|
||||||
if not safe_name:
|
|
||||||
raise UploadError(
|
|
||||||
"Upload filename is required",
|
|
||||||
category=ErrorCategory.VALIDATION,
|
|
||||||
suggestion="Choose a file with a valid filename and retry.",
|
|
||||||
)
|
|
||||||
|
|
||||||
suffix = Path(safe_name).suffix.lower()
|
|
||||||
if suffix not in SUPPORTED_UPLOAD_EXTENSIONS:
|
|
||||||
raise UploadError(
|
|
||||||
f"Unsupported upload extension: {suffix}",
|
|
||||||
category=ErrorCategory.USER_INPUT,
|
|
||||||
suggestion="Upload JPG, JPEG, PNG, TIFF, or PDF files only.",
|
|
||||||
)
|
|
||||||
|
|
||||||
|
|
||||||
def _build_stored_filename(filename: str) -> str:
|
|
||||||
safe_name = Path(filename).name
|
|
||||||
return f"{uuid4()}_{safe_name}"
|
|
||||||
|
|
||||||
|
|
||||||
def _best_effort_delete(path: Path) -> None:
|
def _best_effort_delete(path: Path) -> None:
|
||||||
try:
|
try:
|
||||||
if path.exists():
|
if path.exists():
|
||||||
|
|||||||
@@ -13,6 +13,10 @@ from .base import ServiceBase
|
|||||||
class JobService(ServiceBase):
|
class JobService(ServiceBase):
|
||||||
"""Thin service class for managing jobs in the database."""
|
"""Thin service class for managing jobs in the database."""
|
||||||
|
|
||||||
|
#
|
||||||
|
# CRUD Operations
|
||||||
|
#
|
||||||
|
|
||||||
async def create_job(self, job: Job, session: AsyncSession | None = None) -> Job:
|
async def create_job(self, job: Job, session: AsyncSession | None = None) -> Job:
|
||||||
"""Create a new job in the database."""
|
"""Create a new job in the database."""
|
||||||
async with self._session_scope(session) as _session:
|
async with self._session_scope(session) as _session:
|
||||||
@@ -52,6 +56,8 @@ class JobService(ServiceBase):
|
|||||||
await _session.delete(job)
|
await _session.delete(job)
|
||||||
await _session.commit()
|
await _session.commit()
|
||||||
|
|
||||||
|
# Query Operations
|
||||||
|
|
||||||
async def query_jobs(
|
async def query_jobs(
|
||||||
self,
|
self,
|
||||||
*,
|
*,
|
||||||
@@ -67,10 +73,15 @@ class JobService(ServiceBase):
|
|||||||
query = query.where(Job.document.filename == filename)
|
query = query.where(Job.document.filename == filename)
|
||||||
return (await session.exec(query)).all()
|
return (await session.exec(query)).all()
|
||||||
|
|
||||||
async def list_jobs(self, session: AsyncSession | None = None) -> Sequence[Job]:
|
async def list_jobs(self, session: AsyncSession | None = None, *, load_docs: bool = False) -> Sequence[Job]:
|
||||||
"""List all jobs in the database."""
|
"""List all jobs in the database."""
|
||||||
async with self._session_scope(session) as _session:
|
async with self._session_scope(session) as _session:
|
||||||
return (await _session.exec(select(Job))).all()
|
query = select(Job)
|
||||||
|
if load_docs:
|
||||||
|
query = query.options(selectinload(Job.document)) # pyright: ignore[reportArgumentType]
|
||||||
|
return (await _session.exec(query)).all()
|
||||||
|
|
||||||
|
# Other Operations
|
||||||
|
|
||||||
async def mark_job_status(
|
async def mark_job_status(
|
||||||
self,
|
self,
|
||||||
|
|||||||
@@ -0,0 +1,70 @@
|
|||||||
|
import logging
|
||||||
|
from pathlib import Path
|
||||||
|
from uuid import uuid4
|
||||||
|
|
||||||
|
from ..config import Settings
|
||||||
|
from ..config import get_settings
|
||||||
|
from ..errors import AppError
|
||||||
|
from ..errors import ErrorCategory
|
||||||
|
|
||||||
|
logger = logging.getLogger(__name__)
|
||||||
|
|
||||||
|
SUPPORTED_UPLOAD_EXTENSIONS = {".jpg", ".jpeg", ".png", ".tif", ".tiff", ".pdf"}
|
||||||
|
|
||||||
|
|
||||||
|
class UploadError(AppError):
|
||||||
|
"""Raised when uploaded content cannot be persisted safely."""
|
||||||
|
|
||||||
|
|
||||||
|
def store_file(*, filename: str, file_bytes: bytes, settings: Settings | None = None) -> Path:
|
||||||
|
"""Persist an uploaded file to the configured upload directory."""
|
||||||
|
runtime_settings = settings or get_settings()
|
||||||
|
_validate_upload(filename=filename, file_bytes=file_bytes)
|
||||||
|
|
||||||
|
upload_dir = runtime_settings.upload_dir
|
||||||
|
upload_dir.mkdir(parents=True, exist_ok=True)
|
||||||
|
|
||||||
|
stored_name = _build_stored_filename(filename)
|
||||||
|
stored_path = upload_dir / stored_name
|
||||||
|
|
||||||
|
try:
|
||||||
|
stored_path.write_bytes(file_bytes)
|
||||||
|
except OSError as exc:
|
||||||
|
raise UploadError(
|
||||||
|
"Failed to persist upload file",
|
||||||
|
category=ErrorCategory.INFRA_PERSISTENT,
|
||||||
|
suggestion="Check upload directory permissions and available disk space, then retry.",
|
||||||
|
) from exc
|
||||||
|
|
||||||
|
logger.info("Stored uploaded file: %s", stored_path)
|
||||||
|
return stored_path
|
||||||
|
|
||||||
|
|
||||||
|
def _validate_upload(*, filename: str, file_bytes: bytes) -> None:
|
||||||
|
if not file_bytes:
|
||||||
|
raise UploadError(
|
||||||
|
"Upload payload is empty",
|
||||||
|
category=ErrorCategory.VALIDATION,
|
||||||
|
suggestion="Select a non-empty file and try again.",
|
||||||
|
)
|
||||||
|
|
||||||
|
safe_name = Path(filename).name
|
||||||
|
if not safe_name:
|
||||||
|
raise UploadError(
|
||||||
|
"Upload filename is required",
|
||||||
|
category=ErrorCategory.VALIDATION,
|
||||||
|
suggestion="Choose a file with a valid filename and retry.",
|
||||||
|
)
|
||||||
|
|
||||||
|
suffix = Path(safe_name).suffix.lower()
|
||||||
|
if suffix not in SUPPORTED_UPLOAD_EXTENSIONS:
|
||||||
|
raise UploadError(
|
||||||
|
f"Unsupported upload extension: {suffix}",
|
||||||
|
category=ErrorCategory.USER_INPUT,
|
||||||
|
suggestion="Upload JPG, JPEG, PNG, TIFF, or PDF files only.",
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def _build_stored_filename(filename: str) -> str:
|
||||||
|
safe_name = Path(filename).name
|
||||||
|
return f"{uuid4()}_{safe_name}"
|
||||||
Reference in New Issue
Block a user