From c90ddebb163bbc4c3403a597db18685388507a40 Mon Sep 17 00:00:00 2001 From: Ioannis Paraskevakos Date: Mon, 13 Jul 2026 13:35:30 -0400 Subject: [PATCH 1/2] Restructuring Archivist architecture to better mirror Librarian. Assisted by Claude Sonnet 5 and Opus 4.8 --- alembic.ini | 149 ++++++++++++++++++ alembic/README | 1 + alembic/env.py | 67 ++++++++ alembic/script.py.mako | 28 ++++ ...aseline_manifest_manifest_entry_archive.py | 86 ++++++++++ pyproject.toml | 2 + src/archivist/__init__.py | 2 +- src/archivist/api/archive.py | 35 +++- src/archivist/api/extract.py | 2 +- .../core/{archive.py => archive_job.py} | 2 +- src/archivist/core/storage.py | 96 ----------- src/archivist/database.py | 2 +- src/archivist/orm/__init__.py | 4 +- src/archivist/orm/archive.py | 120 ++++++++++++++ src/archivist/orm/archivequeue.py | 128 --------------- src/archivist/orm/manifest.py | 52 ++++++ src/archivist/orm/manifest_entry.py | 44 ++++++ src/archivist/storage/base.py | 8 +- src/archivist/storage/storage_disk.py | 6 +- src/archivist/storage/storage_hpss.py | 2 +- src/archivist/tasks/archive.py | 80 ++++++---- tests/conftest.py | 35 ++++ tests/test_api.py | 4 +- tests/test_archive.py | 18 +-- tests/test_archivequeue.py | 64 +++----- tests/test_classes.py | 6 +- tests/test_storage_disk.py | 10 +- tests/test_tasks_archive.py | 28 ++-- 28 files changed, 732 insertions(+), 349 deletions(-) create mode 100644 alembic.ini create mode 100644 alembic/README create mode 100644 alembic/env.py create mode 100644 alembic/script.py.mako create mode 100644 alembic/versions/ff34fa3fb15c_baseline_manifest_manifest_entry_archive.py rename src/archivist/core/{archive.py => archive_job.py} (98%) delete mode 100644 src/archivist/core/storage.py create mode 100644 src/archivist/orm/archive.py delete mode 100644 src/archivist/orm/archivequeue.py create mode 100644 src/archivist/orm/manifest.py create mode 100644 src/archivist/orm/manifest_entry.py diff --git a/alembic.ini b/alembic.ini new file mode 100644 index 0000000..b662638 --- /dev/null +++ b/alembic.ini @@ -0,0 +1,149 @@ +# A generic, single database configuration. + +[alembic] +# path to migration scripts. +# this is typically a path given in POSIX (e.g. forward slashes) +# format, relative to the token %(here)s which refers to the location of this +# ini file +script_location = %(here)s/alembic + +# template used to generate migration file names; The default value is %%(rev)s_%%(slug)s +# Uncomment the line below if you want the files to be prepended with date and time +# see https://alembic.sqlalchemy.org/en/latest/tutorial.html#editing-the-ini-file +# for all available tokens +# file_template = %%(year)d_%%(month).2d_%%(day).2d_%%(hour).2d%%(minute).2d-%%(rev)s_%%(slug)s +# Or organize into date-based subdirectories (requires recursive_version_locations = true) +# file_template = %%(year)d/%%(month).2d/%%(day).2d_%%(hour).2d%%(minute).2d_%%(second).2d_%%(rev)s_%%(slug)s + +# sys.path path, will be prepended to sys.path if present. +# defaults to the current working directory. for multiple paths, the path separator +# is defined by "path_separator" below. +prepend_sys_path = . + + +# timezone to use when rendering the date within the migration file +# as well as the filename. +# If specified, requires the tzdata library which can be installed by adding +# `alembic[tz]` to the pip requirements. +# string value is passed to ZoneInfo() +# leave blank for localtime +# timezone = + +# max length of characters to apply to the "slug" field +# truncate_slug_length = 40 + +# set to 'true' to run the environment during +# the 'revision' command, regardless of autogenerate +# revision_environment = false + +# set to 'true' to allow .pyc and .pyo files without +# a source .py file to be detected as revisions in the +# versions/ directory +# sourceless = false + +# version location specification; This defaults +# to /versions. When using multiple version +# directories, initial revisions must be specified with --version-path. +# The path separator used here should be the separator specified by "path_separator" +# below. +# version_locations = %(here)s/bar:%(here)s/bat:%(here)s/alembic/versions + +# path_separator; This indicates what character is used to split lists of file +# paths, including version_locations and prepend_sys_path within configparser +# files such as alembic.ini. +# The default rendered in new alembic.ini files is "os", which uses os.pathsep +# to provide os-dependent path splitting. +# +# Note that in order to support legacy alembic.ini files, this default does NOT +# take place if path_separator is not present in alembic.ini. If this +# option is omitted entirely, fallback logic is as follows: +# +# 1. Parsing of the version_locations option falls back to using the legacy +# "version_path_separator" key, which if absent then falls back to the legacy +# behavior of splitting on spaces and/or commas. +# 2. Parsing of the prepend_sys_path option falls back to the legacy +# behavior of splitting on spaces, commas, or colons. +# +# Valid values for path_separator are: +# +# path_separator = : +# path_separator = ; +# path_separator = space +# path_separator = newline +# +# Use os.pathsep. Default configuration used for new projects. +path_separator = os + +# set to 'true' to search source files recursively +# in each "version_locations" directory +# new in Alembic version 1.10 +# recursive_version_locations = false + +# the output encoding used when revision files +# are written from script.py.mako +# output_encoding = utf-8 + +# database URL. Archivist's env.py builds the URL from Settings +# (`get_settings().sqlalchemy_database_uri`) rather than reading it here, so +# this key is intentionally left unset. +# sqlalchemy.url = + + +[post_write_hooks] +# post_write_hooks defines scripts or Python functions that are run +# on newly generated revision scripts. See the documentation for further +# detail and examples + +# format using "black" - use the console_scripts runner, against the "black" entrypoint +# hooks = black +# black.type = console_scripts +# black.entrypoint = black +# black.options = -l 79 REVISION_SCRIPT_FILENAME + +# lint with attempts to fix using "ruff" - use the module runner, against the "ruff" module +# hooks = ruff +# ruff.type = module +# ruff.module = ruff +# ruff.options = check --fix REVISION_SCRIPT_FILENAME + +# Alternatively, use the exec runner to execute a binary found on your PATH +# hooks = ruff +# ruff.type = exec +# ruff.executable = ruff +# ruff.options = check --fix REVISION_SCRIPT_FILENAME + +# Logging configuration. This is also consumed by the user-maintained +# env.py script only. +[loggers] +keys = root,sqlalchemy,alembic + +[handlers] +keys = console + +[formatters] +keys = generic + +[logger_root] +level = WARNING +handlers = console +qualname = + +[logger_sqlalchemy] +level = WARNING +handlers = +qualname = sqlalchemy.engine + +[logger_alembic] +level = INFO +handlers = +qualname = alembic + +[handler_console] +class = StreamHandler +args = (sys.stderr,) +level = NOTSET +formatter = generic + +[formatter_generic] +format = %(levelname)-5.5s [%(name)s] %(message)s +datefmt = %H:%M:%S diff --git a/alembic/README b/alembic/README new file mode 100644 index 0000000..2500aa1 --- /dev/null +++ b/alembic/README @@ -0,0 +1 @@ +Generic single-database configuration. diff --git a/alembic/env.py b/alembic/env.py new file mode 100644 index 0000000..d38c1d2 --- /dev/null +++ b/alembic/env.py @@ -0,0 +1,67 @@ +"""Alembic environment for Archivist. + +Wires Alembic to Archivist's settings-driven database configuration. The +target metadata is `database.Base.metadata`; importing `archivist.orm` +registers the `Manifest` / `ManifestEntry` / `Archive` tables on it. Mirrors +the sibling Librarian's `alembic/env.py`. +""" + +# A hack so that `archivist` is importable when alembic is run from the repo +# root: the `src/` layout means the package lives under src/. +import sys + +sys.path.insert(0, "src") + +from logging.config import fileConfig + +from alembic import context +from archivist import orm # noqa: F401 (registers ORM tables on Base.metadata) +from archivist.database import Base, get_engine +from archivist.settings import get_settings + +# this is the Alembic Config object, which provides +# access to the values within the .ini file in use. +config = context.config + +# Interpret the config file for Python logging. +if config.config_file_name is not None: + fileConfig(config.config_file_name) + +# Autogenerate compares this metadata against the live database. +target_metadata = Base.metadata + + +def run_migrations_offline() -> None: + """Run migrations in 'offline' mode -- all we need is a URL.""" + url = get_settings().sqlalchemy_database_uri + context.configure( + url=url, + target_metadata=target_metadata, + literal_binds=True, + dialect_opts={"paramstyle": "named"}, + render_as_batch=True, + ) + + with context.begin_transaction(): + context.run_migrations() + + +def run_migrations_online() -> None: + """Run migrations in 'online' mode -- using the actual Archivist database + connection built from settings. + """ + with get_engine().connect() as connection: + context.configure( + connection=connection, + target_metadata=target_metadata, + render_as_batch=True, + ) + + with context.begin_transaction(): + context.run_migrations() + + +if context.is_offline_mode(): + run_migrations_offline() +else: + run_migrations_online() diff --git a/alembic/script.py.mako b/alembic/script.py.mako new file mode 100644 index 0000000..1101630 --- /dev/null +++ b/alembic/script.py.mako @@ -0,0 +1,28 @@ +"""${message} + +Revision ID: ${up_revision} +Revises: ${down_revision | comma,n} +Create Date: ${create_date} + +""" +from typing import Sequence, Union + +from alembic import op +import sqlalchemy as sa +${imports if imports else ""} + +# revision identifiers, used by Alembic. +revision: str = ${repr(up_revision)} +down_revision: Union[str, Sequence[str], None] = ${repr(down_revision)} +branch_labels: Union[str, Sequence[str], None] = ${repr(branch_labels)} +depends_on: Union[str, Sequence[str], None] = ${repr(depends_on)} + + +def upgrade() -> None: + """Upgrade schema.""" + ${upgrades if upgrades else "pass"} + + +def downgrade() -> None: + """Downgrade schema.""" + ${downgrades if downgrades else "pass"} diff --git a/alembic/versions/ff34fa3fb15c_baseline_manifest_manifest_entry_archive.py b/alembic/versions/ff34fa3fb15c_baseline_manifest_manifest_entry_archive.py new file mode 100644 index 0000000..483a36c --- /dev/null +++ b/alembic/versions/ff34fa3fb15c_baseline_manifest_manifest_entry_archive.py @@ -0,0 +1,86 @@ +"""baseline: manifest, manifest_entry, archive + +Revision ID: ff34fa3fb15c +Revises: +Create Date: 2026-07-13 11:12:15.020976 + +""" + +from typing import Sequence, Union + +import sqlalchemy as sa + +from alembic import op + +# revision identifiers, used by Alembic. +revision: str = "ff34fa3fb15c" + +down_revision: Union[str, Sequence[str], None] = None +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + """Upgrade schema.""" + # ### commands auto generated by Alembic - please adjust! ### + op.create_table( + "manifest", + sa.Column("id", sa.String(length=36), nullable=False), + sa.Column("librarian_name", sa.String(length=256), nullable=False), + sa.Column("created_time", sa.DateTime(), nullable=False), + sa.Column("total_size_bytes", sa.BigInteger(), nullable=False), + sa.Column("file_count", sa.Integer(), nullable=False), + sa.PrimaryKeyConstraint("id"), + ) + op.create_table( + "archive", + sa.Column("id", sa.String(length=36), nullable=False), + sa.Column("created_time", sa.DateTime(), nullable=False), + sa.Column("retries", sa.Integer(), nullable=False), + sa.Column("manifest_id", sa.String(length=36), nullable=False), + sa.Column("archive_root", sa.String(length=512), nullable=False), + sa.Column("archive_path", sa.String(length=2048), nullable=True), + sa.Column("librarian_archive_id", sa.String(length=2048), nullable=True), + sa.Column("consumed", sa.Boolean(), nullable=True), + sa.Column("consumed_time", sa.DateTime(), nullable=True), + sa.Column("completed", sa.Boolean(), nullable=True), + sa.Column("completed_time", sa.DateTime(), nullable=True), + sa.Column("failed", sa.Boolean(), nullable=True), + sa.ForeignKeyConstraint(["manifest_id"], ["manifest.id"], ondelete="CASCADE"), + sa.PrimaryKeyConstraint("id"), + sa.UniqueConstraint("librarian_archive_id"), + sa.UniqueConstraint("manifest_id"), + ) + op.create_table( + "manifest_entry", + sa.Column("id", sa.Integer(), nullable=False), + sa.Column("manifest_id", sa.String(length=36), nullable=False), + sa.Column("name", sa.String(length=1024), nullable=False), + sa.Column("create_time", sa.DateTime(), nullable=False), + sa.Column("size", sa.BigInteger(), nullable=False), + sa.Column("checksum", sa.String(length=256), nullable=False), + sa.Column("uploader", sa.String(length=256), nullable=False), + sa.Column("source", sa.String(length=256), nullable=False), + sa.Column("instance_path", sa.String(length=2048), nullable=False), + sa.Column("instance_create_time", sa.DateTime(), nullable=False), + sa.Column("instance_available", sa.Boolean(), nullable=False), + sa.Column("outgoing_transfer_id", sa.Integer(), nullable=False), + sa.ForeignKeyConstraint(["manifest_id"], ["manifest.id"], ondelete="CASCADE"), + sa.PrimaryKeyConstraint("id"), + ) + with op.batch_alter_table("manifest_entry", schema=None) as batch_op: + batch_op.create_index(batch_op.f("ix_manifest_entry_manifest_id"), ["manifest_id"], unique=False) + + # ### end Alembic commands ### + + +def downgrade() -> None: + """Downgrade schema.""" + # ### commands auto generated by Alembic - please adjust! ### + with op.batch_alter_table("manifest_entry", schema=None) as batch_op: + batch_op.drop_index(batch_op.f("ix_manifest_entry_manifest_id")) + + op.drop_table("manifest_entry") + op.drop_table("archive") + op.drop_table("manifest") + # ### end Alembic commands ### diff --git a/pyproject.toml b/pyproject.toml index 3db50fa..25100ce 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -25,6 +25,7 @@ dependencies = [ "loguru", "uvicorn", "requests>=2.32.5", + "alembic>=1.18.5", ] [project.urls] @@ -39,6 +40,7 @@ dev = [ "coveralls>=4.0.1", "httpx>=0.28.1", "hypothesis>=6.141.1", + "ipython>=9.15.0", "isort>=6.1.0", "pre-commit>=4.3.0", "pytest>=8.4.2", diff --git a/src/archivist/__init__.py b/src/archivist/__init__.py index d86af70..5f632e4 100644 --- a/src/archivist/__init__.py +++ b/src/archivist/__init__.py @@ -4,6 +4,6 @@ """Archive tools for data.""" from ._version import __version__ -from .core.archive import Archive +from .core.archive_job import ArchiveJob from .core.registry import RegistryLibrarian, RegistrySqlite from .storage import StorageDisk, StorageHPSS diff --git a/src/archivist/api/archive.py b/src/archivist/api/archive.py index 4302c8c..89fb369 100644 --- a/src/archivist/api/archive.py +++ b/src/archivist/api/archive.py @@ -7,7 +7,7 @@ from archivist.api import router from archivist.core.models import ManifestFailedResponse, ManifestRequest, ManifestResponse from archivist.database import yield_session -from archivist.orm.archivequeue import ArchiveQueue +from archivist.orm import Archive, Manifest, ManifestEntry from archivist.settings import Settings, get_settings @@ -34,16 +34,37 @@ def archive( manifest_id = uuid.uuid4() + # get_or_create adds the Manifest to the session on the create path and + # returns it; attach entries and the archive job through relationships so + # the FKs resolve and a single commit cascade-persists everything. + manifest = Manifest.get_or_create( + session, + manifest_id=str(manifest_id), + librarian_name=manifest_request.librarian_name, + ) + manifest.entries = [ + ManifestEntry( + name=entry.name, + create_time=entry.create_time, + size=entry.size, + checksum=entry.checksum, + uploader=entry.uploader, + source=entry.source, + instance_path=entry.instance_path, + instance_create_time=entry.instance_create_time, + instance_available=entry.instance_available, + outgoing_transfer_id=entry.outgoing_transfer_id, + ) + for entry in manifest_request.store_files + ] + manifest.total_size_bytes = total_size + manifest.file_count = len(manifest_request.store_files) + logger.info( f"Archiving manifest {manifest_id} from librarian '{manifest_request.librarian_name}': {len(manifest_request.store_files)} file(s), {total_size} bytes total" ) - item = ArchiveQueue.new_item( - manifest_id=str(manifest_id), - manifest=manifest_request.model_dump_json(), - paths=[entry.instance_path for entry in manifest_request.store_files], - root=settings.archive_root, - ) + item = Archive.new_item(manifest=manifest, archive_root=settings.archive_root) session.add(item) session.commit() diff --git a/src/archivist/api/extract.py b/src/archivist/api/extract.py index 2ebe8db..ae6ec02 100644 --- a/src/archivist/api/extract.py +++ b/src/archivist/api/extract.py @@ -4,7 +4,7 @@ from fastapi import Depends, Response from archivist.api import router -from archivist.core.archive import Archive +from archivist.core.archive_job import ArchiveJob from archivist.core.models import ManifestFailedResponse, ManifestRequest, ManifestResponse # from archivist.queue import Queue, get_extract_queue diff --git a/src/archivist/core/archive.py b/src/archivist/core/archive_job.py similarity index 98% rename from src/archivist/core/archive.py rename to src/archivist/core/archive_job.py index 0d5d5b0..b9014fc 100644 --- a/src/archivist/core/archive.py +++ b/src/archivist/core/archive_job.py @@ -7,7 +7,7 @@ from .models import ManifestRequest -class Archive(object): +class ArchiveJob(object): """Class representing a single archive in a storage system. When listing objects in the manifest or paths list, directories can be specified diff --git a/src/archivist/core/storage.py b/src/archivist/core/storage.py deleted file mode 100644 index 8d19011..0000000 --- a/src/archivist/core/storage.py +++ /dev/null @@ -1,96 +0,0 @@ -# Copyright (c) 2025-2026 Simons Observatory. -# Full license can be found in the top level "LICENSE" file. -"""Classes for working with storage systems.""" - -from archivist.core.archive import Archive -from archivist.settings import Settings - - -class Storage(object): - """Base class representing an archive storage system.""" - - def __init__(self, settings: Settings | None = None): - self._settings = settings - - def _store(self, archive): - raise NotImplementedError("Fell through to base class") - - def store(self, archive): - """Store an archive. - - Args: - archive (Archive): The archive to store. - - Returns: - (dict): Metadata specific to the storage system. Used when registering - the archive. - - """ - return self._store(archive) - - def _extract(self, archive_name, storage_info, outdir, paths=None): - raise NotImplementedError("Fell through to base class") - - def extract(self, archive_name, storage_info, outdir, paths=None): - """Extract objects from an archive. - - The archive name and additional storage metadata is used to locate and extract - one or more paths from the archive. `storage_info` should be the same - metadata returned by the call to `store()` and kept in the registry. - - Args: - archive_name (str): The name of the archive to remove. - storage_info (dict): Additional storage-system specific metadata. - outdir (str): The top-level output directory where extracted files - should be placed. - paths (list): Optional list of objects to extract. If None, the - entire archive is extracted. - - Returns: - (None) - - """ - self._extract(archive_name, storage_info, outdir, paths=None) - - -class StorageDisk(Storage): - """Simple filesystem storage. - - This class just keeps archives in subdirectories within a top-level location. - Useful for testing and for archiving to "slow" storage that provides a POSIX - API. - - Args: - directory (str): The top-level location for storing archives. - - """ - - def __init__(self, settings: Settings | None = None): - super().__init__(settings=settings) - self._storage_proc = None - self._archive = None - - def _store(self, archive: Archive): - """Tar the archive files into the storage directory.""" - pass - - def _extract(self, archive_name, storage_info, outdir, paths=None): - """Extract files from tar archives.""" - pass - - -class StorageHPSS(Storage): - """Store archives to HPSS tapes.""" - - def __init__(self): - super().__init__() - - def _store(self, archive): - """Use HTAR to archive files. - The tape list is returned for including in the registry entry. - """ - pass - - def _extract(self, archive_name, storage_info, outdir, paths=None): - """Extract files from HTAR archives.""" - pass diff --git a/src/archivist/database.py b/src/archivist/database.py index 1b4b1c6..d224d3e 100644 --- a/src/archivist/database.py +++ b/src/archivist/database.py @@ -17,7 +17,7 @@ String, create_engine, ) -from sqlalchemy.orm import Session, declarative_base, sessionmaker +from sqlalchemy.orm import Session, declarative_base, relationship, sessionmaker from .settings import get_settings diff --git a/src/archivist/orm/__init__.py b/src/archivist/orm/__init__.py index 2446994..bfe0004 100644 --- a/src/archivist/orm/__init__.py +++ b/src/archivist/orm/__init__.py @@ -1 +1,3 @@ -from .archivequeue import ArchiveQueue # noqa: F401 +from .archive import Archive # noqa: F401 +from .manifest import Manifest # noqa: F401 +from .manifest_entry import ManifestEntry # noqa: F401 diff --git a/src/archivist/orm/archive.py b/src/archivist/orm/archive.py new file mode 100644 index 0000000..058c29e --- /dev/null +++ b/src/archivist/orm/archive.py @@ -0,0 +1,120 @@ +""" +The ORM definition for the queue of archive jobs awaiting processing. + +When the API receives a manifest to archive, it does not perform the +archive operation synchronously. Instead, it places an item in this +queue (backed by the database, so it survives restarts) and a separate +worker thread picks items up and performs the actual archive operation. +""" + +import datetime +import uuid + +from sqlalchemy import select +from sqlalchemy.orm import Session, joinedload, selectinload + +from .. import database as db +from .manifest import Manifest + + +class Archive(db.Base): + """ + The queue of archive jobs awaiting processing. + """ + + __tablename__ = "archive" + + id = db.Column(db.String(36), primary_key=True) # archivist-minted uuid4, decision 1 + "Archivist-minted uuid4 identifying this archive job." + created_time = db.Column(db.DateTime, nullable=False) + "The time this job was added to the queue." + retries = db.Column(db.Integer, nullable=False, default=0) + "The number of times this job has been attempted." + + manifest_id = db.Column( + db.String(36), + db.ForeignKey("manifest.id", ondelete="CASCADE"), + nullable=False, + unique=True, # 1:1, decision 5 + ) + "FK to the manifest being archived (unique -> 1:1)." + manifest = db.relationship("Manifest", back_populates="archive") + "The manifest being archived." + + archive_root = db.Column(db.String(512), nullable=False) # renamed from `root` + "The archive root at submit time; snapshotted since settings can change." + archive_path = db.Column(db.String(2048), nullable=True) + "The destination path, set on completion." + + librarian_archive_id = db.Column(db.String(2048), nullable=True, unique=True) # decision 2 + "Site-specific identifier supplied by the Librarian callback contract." + + consumed = db.Column(db.Boolean, default=False) + "Whether a worker has picked this job up." + consumed_time = db.Column(db.DateTime) + "The time the job was picked up. Useful for pruning." + completed = db.Column(db.Boolean, default=False) + "Whether the job has finished (success or failure)." + completed_time = db.Column(db.DateTime) + "The time the job was marked completed." + failed = db.Column(db.Boolean, default=False) + "Whether the job failed, and that is why it is completed." + + @classmethod + def new_item(cls, manifest: "Manifest", archive_root: str) -> "Archive": + return cls( + id=str(uuid.uuid4()), + manifest=manifest, + archive_root=archive_root, + created_time=datetime.datetime.now(datetime.timezone.utc), + retries=0, + ) + + @classmethod + def dequeue(cls, session: Session) -> "Archive | None": + # Mirror librarian's consume_queue_item: lock-and-skip so workers never + # grab the same row, and filter on BOTH flags (consumed AND completed). + stmt = ( + select(cls) + .options(joinedload(cls.manifest).selectinload(Manifest.entries)) + .with_for_update(skip_locked=True) + .filter_by(consumed=False, completed=False) + .order_by(cls.created_time.asc()) + ) + item = session.execute(stmt).unique().scalar() + if item is not None: + item.consumed = True + item.consumed_time = datetime.datetime.now(datetime.timezone.utc) + session.commit() + return item + + def complete(self, session: Session): + """ + Mark this queue item as successfully completed. + + Parameters + ---------- + session : Session + The database session to use. + """ + + self.completed = True + self.completed_time = datetime.datetime.now(datetime.timezone.utc) + + session.commit() + + def fail(self, session: Session): + """ + Mark this queue item as failed. + + Parameters + ---------- + session : Session + The database session to use. + """ + + self.failed = True + self.completed = True + self.completed_time = datetime.datetime.now(datetime.timezone.utc) + + session.commit() diff --git a/src/archivist/orm/archivequeue.py b/src/archivist/orm/archivequeue.py deleted file mode 100644 index 415d5e1..0000000 --- a/src/archivist/orm/archivequeue.py +++ /dev/null @@ -1,128 +0,0 @@ -""" -The ORM definition for the queue of archive jobs awaiting processing. - -When the API receives a manifest to archive, it does not perform the -archive operation synchronously. Instead, it places an item in this -queue (backed by the database, so it survives restarts) and a separate -worker thread picks items up and performs the actual archive operation. -""" - -import datetime - -from sqlalchemy.orm import Session - -from .. import database as db - - -class ArchiveQueue(db.Base): - """ - The queue of archive jobs awaiting processing. - """ - - __tablename__ = "archive_queue" - - id = db.Column(db.Integer, primary_key=True) - "The unique identifier for this item in the queue." - created_time = db.Column(db.DateTime, nullable=False) - "The time that this item was added to the queue." - retries = db.Column(db.Integer, nullable=False, default=0) - "The number of times this item has been tried to be archived." - - manifest_id = db.Column(db.String(256), nullable=False, unique=True) - "The ID of the manifest associated with this archive." - manifest = db.Column(db.PickleType, nullable=False) - "The raw manifest (JSON) that was submitted for archiving." - paths = db.Column(db.PickleType, nullable=False) - "The list of file / directory paths contained in the archive." - root = db.Column(db.String(512), nullable=False) - "The common root filesystem location for building relative paths." - - consumed = db.Column(db.Boolean, default=False) - "Whether this queue item has been picked up by a worker." - consumed_time = db.Column(db.DateTime) - "The time at which the item was picked up. Useful for pruning." - - completed = db.Column(db.Boolean, default=False) - "Whether this queue item has been entirely completed." - completed_time = db.Column(db.DateTime) - "The time at which the queue item was marked as completed." - - failed = db.Column(db.Boolean, default=False) - "Whether this queue item failed, and that is the reason for completed status." - - @classmethod - def new_item(cls, manifest_id: str, manifest: str, paths: list[str], root: str) -> "ArchiveQueue": - """ - Create a new item in the queue from an incoming archive request. - - Parameters - ---------- - manifest_id : str - The ID of the manifest associated with this archive. - manifest : str - The raw manifest (JSON) that was submitted for archiving. - paths : list[str] - The list of file / directory paths contained in the archive. - root : str - The common root filesystem location for building relative paths. - """ - - return cls( - manifest_id=manifest_id, - manifest=manifest, - paths=paths, - root=root, - created_time=datetime.datetime.now(datetime.timezone.utc), - retries=0, - ) - - @classmethod - def dequeue(cls, session: Session) -> "ArchiveQueue | None": - """ - Pop the oldest unconsumed item off the queue and mark it as consumed. - - Parameters - ---------- - session : Session - The database session to use. - """ - - item = session.query(cls).filter_by(consumed=False).order_by(cls.created_time.asc()).first() - - if item is not None: - item.consumed = True - item.consumed_time = datetime.datetime.now(datetime.timezone.utc) - session.commit() - - return item - - def complete(self, session: Session): - """ - Mark this queue item as successfully completed. - - Parameters - ---------- - session : Session - The database session to use. - """ - - self.completed = True - self.completed_time = datetime.datetime.now(datetime.timezone.utc) - - session.commit() - - def fail(self, session: Session): - """ - Mark this queue item as failed. - - Parameters - ---------- - session : Session - The database session to use. - """ - - self.failed = True - self.completed = True - self.completed_time = datetime.datetime.now(datetime.timezone.utc) - - session.commit() diff --git a/src/archivist/orm/manifest.py b/src/archivist/orm/manifest.py new file mode 100644 index 0000000..dad1b30 --- /dev/null +++ b/src/archivist/orm/manifest.py @@ -0,0 +1,52 @@ +import datetime + +from .. import database as db + + +class Manifest(db.Base): + """ + A manifest submitted by a Librarian: the authoritative record of one + archive request, plus its per-file entries. + """ + + __tablename__ = "manifest" + + id = db.Column(db.String(36), primary_key=True) + "Archivist-minted uuid4 identifying this manifest." + librarian_name = db.Column(db.String(256), nullable=False) + "The name of the Librarian that generated this manifest." + created_time = db.Column(db.DateTime, nullable=False) + "The time this manifest was received." + total_size_bytes = db.Column(db.BigInteger, nullable=False, default=0) + "Denormalized cache of the summed size of all entries." + file_count = db.Column(db.Integer, nullable=False, default=0) + "Denormalized cache of the number of entries." + + entries = db.relationship( + "ManifestEntry", + back_populates="manifest", + cascade="all, delete-orphan", + passive_deletes=True, + ) + "The per-file entries belonging to this manifest." + archive = db.relationship( + "Archive", + back_populates="manifest", + uselist=False, # 1:1, decision 5 + cascade="all, delete-orphan", + passive_deletes=True, + ) + "The single archive job for this manifest (1:1, decision 5)." + + @classmethod + def get_or_create(cls, session, manifest_id, librarian_name) -> "Manifest": + existing = session.get(cls, manifest_id) + if existing is not None: + return existing + manifest = cls( + id=manifest_id, + librarian_name=librarian_name, + created_time=datetime.datetime.now(datetime.timezone.utc), + ) + session.add(manifest) + return manifest diff --git a/src/archivist/orm/manifest_entry.py b/src/archivist/orm/manifest_entry.py new file mode 100644 index 0000000..a657fd9 --- /dev/null +++ b/src/archivist/orm/manifest_entry.py @@ -0,0 +1,44 @@ +from .. import database as db + + +class ManifestEntry(db.Base): + """ + One file's authoritative record within a manifest. One shared table keyed + by the manifest FK -- not a table per manifest. Column set mirrors the + Pydantic `ManifestEntry` in `core/models.py` 1:1. + """ + + __tablename__ = "manifest_entry" + + id = db.Column(db.Integer, primary_key=True) + "Surrogate PK; a single file row has no external identity." + manifest_id = db.Column( + db.String(36), + db.ForeignKey("manifest.id", ondelete="CASCADE"), + nullable=False, + index=True, + ) + "FK back to the owning manifest." + name = db.Column(db.String(1024), nullable=False) + "The name of the file." + create_time = db.Column(db.DateTime, nullable=False) + "The time the file was created." + size = db.Column(db.BigInteger, nullable=False) + "The size of the file in bytes." + checksum = db.Column(db.String(256), nullable=False) + "The sha256 checksum of the file." + uploader = db.Column(db.String(256), nullable=False) + "The uploader of the file." + source = db.Column(db.String(256), nullable=False) + "The source of the file." + instance_path = db.Column(db.String(2048), nullable=False) + "The path to the instance on the store." + instance_create_time = db.Column(db.DateTime, nullable=False) + "The time the instance was created." + instance_available = db.Column(db.Boolean, nullable=False) + "Whether the instance is available." + outgoing_transfer_id = db.Column(db.Integer, nullable=False) + "The ID of the outgoing transfer, if it exists." + + manifest = db.relationship("Manifest", back_populates="entries") + "The owning manifest." diff --git a/src/archivist/storage/base.py b/src/archivist/storage/base.py index 310b896..15ea03f 100644 --- a/src/archivist/storage/base.py +++ b/src/archivist/storage/base.py @@ -1,4 +1,4 @@ -from archivist.core.archive import Archive +from archivist.core.archive_job import ArchiveJob from archivist.settings import Settings @@ -8,14 +8,14 @@ class Storage(object): def __init__(self, settings: Settings | None = None): self._settings = settings - def _store(self, archive): + def _store(self, archive: ArchiveJob) -> "Storage": raise NotImplementedError("Fell through to base class") - def store(self, archive): + def store(self, archive: ArchiveJob) -> "Storage": """Store an archive. Args: - archive (Archive): The archive to store. + archive (ArchiveJob): The archive to store. Returns: (dict): Metadata specific to the storage system. Used when registering diff --git a/src/archivist/storage/storage_disk.py b/src/archivist/storage/storage_disk.py index eba398c..ba8bbd7 100644 --- a/src/archivist/storage/storage_disk.py +++ b/src/archivist/storage/storage_disk.py @@ -5,9 +5,7 @@ from loguru import logger -from archivist.core.archive import Archive -from archivist.database import get_session -from archivist.orm.archivequeue import ArchiveQueue +from archivist.core.archive_job import ArchiveJob from archivist.settings import Settings from archivist.storage.base import Storage @@ -49,7 +47,7 @@ def _copy_files(self, file_list: list[tuple[Path, Path]]) -> None: else: shutil.copy2(src_path, dst_path) - def _store(self, archive: Archive) -> "StorageDisk": + def _store(self, archive: ArchiveJob) -> "StorageDisk": """Copy the archive's files into a per-manifest subdirectory of archive_root.""" self._archive = archive paths = self._archive.create_archive() diff --git a/src/archivist/storage/storage_hpss.py b/src/archivist/storage/storage_hpss.py index 8989fa4..6663cb3 100644 --- a/src/archivist/storage/storage_hpss.py +++ b/src/archivist/storage/storage_hpss.py @@ -1,4 +1,4 @@ -from archivist.core.archive import Archive +from archivist.core.archive_job import ArchiveJob from archivist.settings import Settings from archivist.storage.base import Storage diff --git a/src/archivist/tasks/archive.py b/src/archivist/tasks/archive.py index d943a39..6ad7dd5 100644 --- a/src/archivist/tasks/archive.py +++ b/src/archivist/tasks/archive.py @@ -1,69 +1,93 @@ -import json import queue from time import sleep +from typing import Callable import loguru -from sqlalchemy import inspect +from sqlalchemy.orm import Session from archivist.core import storage_factory -from archivist.core.archive import Archive +from archivist.core.archive_job import ArchiveJob +from archivist.core.models import ManifestEntry from archivist.database import get_session -from archivist.orm.archivequeue import ArchiveQueue +from archivist.orm.archive import Archive from archivist.queue import get_status_queue from archivist.settings import get_settings -def start_archive(librarian_name: str, store_files: bool = True) -> None: +def start_archive( + librarian_name: str, + store_files: bool = True, + session_maker: Callable[[], Session] = get_session, +) -> bool: """ - Start an archive process by sending a request to the archivist server. + Dequeue the next archive job and hand it to a storage backend. Called directly from the background worker thread loop (not a FastAPI route), so dependencies are resolved here rather than via `Depends`. + Takes ``session_maker`` (rather than calling ``get_session`` inline) so a + future task-based runner can inject it; returns ``True`` when a job was + picked up and ``False`` when the queue was empty, mirroring librarian's + ``consume_queue_item``. """ status_queue = get_status_queue() settings = get_settings() - session = get_session() + session = session_maker() try: - archive_item = ArchiveQueue.dequeue(session) - if archive_item is not None: - archive_dict = {c.key: getattr(archive_item, c.key) for c in inspect(archive_item).mapper.column_attrs} - loguru.logger.info(f"Archive Item: {archive_dict}, {type(archive_item)}") - archive = Archive( - manifest=json.loads(archive_dict["manifest"]), - manifest_id=archive_dict["manifest_id"], - local_root=settings.local_root, - archive_root=settings.archive_root, - type=settings.archive_type, - ) - storage = storage_factory[settings.archive_type](settings=settings) - storage_task = storage.store(archive) - status_queue.enqueue(storage_task) + archive_item = Archive.dequeue(session) + if archive_item is None: + return False + + manifest = { + "store_files": [ + ManifestEntry.model_validate(entry, from_attributes=True).model_dump() + for entry in archive_item.manifest.entries + ] + } + loguru.logger.info(f"Archive Item: {manifest}, {type(archive_item)}") + archive = ArchiveJob( + manifest=manifest, + manifest_id=archive_item.manifest_id, + local_root=settings.local_root, + archive_root=archive_item.archive_root, + type=settings.archive_type, + ) + storage = storage_factory[settings.archive_type](settings=settings) + storage_task = storage.store(archive) + status_queue.enqueue(storage_task) + return True finally: session.close() -def process_status_queue() -> None: +def process_status_queue(session_maker: Callable[[], Session] = get_session) -> bool: """ Process the status queue to check for completed or failed archive tasks. Called directly from the background worker thread loop (not a FastAPI route), so dependencies are resolved here rather than via `Depends`. + Returns ``True`` when a task was handled and ``False`` when the queue was + empty, mirroring librarian's ``check_on_consumed``. """ status_queue = get_status_queue() try: storage_task = status_queue.dequeue(block=False) + except queue.Empty: + return False + + try: loguru.logger.debug(f"Processing status queue for storage task: {storage_task._archive}") if storage_task.future.done(): loguru.logger.debug(f"Storage task for archive {storage_task._archive.manifest_id} future completed.") - session = get_session() + session = session_maker() try: - item = session.query(ArchiveQueue).filter_by(manifest_id=storage_task._archive.manifest_id).first() + item = session.query(Archive).filter_by(manifest_id=storage_task._archive.manifest_id).first() loguru.logger.debug( - f"Retrieved ArchiveQueue item for manifest_id {storage_task._archive.manifest_id}: {item}" + f"Retrieved Archive item for manifest_id {storage_task._archive.manifest_id}: {item}" ) if item is None: - return + return True try: + item.archive_path = storage_task._archive.archive_root item.complete(session) loguru.logger.info(f"Archive {storage_task._archive.manifest_id} completed successfully.") except Exception: @@ -78,8 +102,8 @@ def process_status_queue() -> None: status_queue.enqueue(storage_task) status_queue.task_done() - except queue.Empty: - pass + return True except Exception as e: loguru.logger.error(f"Error processing status queue: {e}") sleep(1) # Sleep for a short duration to avoid busy waiting + return True diff --git a/tests/conftest.py b/tests/conftest.py index 5643039..555d388 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -114,6 +114,41 @@ def make_manifest_entry(**overrides): return entry +def make_archive_item( + session, + manifest_id="m1", + *, + librarian_name="test-librarian", + archive_root="/root", + entries=None, +): + """Persist a `Manifest` (+ `ManifestEntry` rows) and its `Archive` job. + + Mirrors what the `/archive` route does, so worker/ORM tests can build a + ready-to-dequeue job in one call. `entries` is a list of dicts as produced + by `make_manifest_entry`; defaults to a single entry. + """ + + from archivist.orm import Archive, Manifest, ManifestEntry + + if entries is None: + entries = [make_manifest_entry()] + + manifest = Manifest.get_or_create( + session, + manifest_id=manifest_id, + librarian_name=librarian_name, + ) + manifest.entries = [ManifestEntry(**entry) for entry in entries] + manifest.total_size_bytes = sum(entry["size"] for entry in entries) + manifest.file_count = len(entries) + + item = Archive.new_item(manifest=manifest, archive_root=str(archive_root)) + session.add(item) + session.commit() + return item + + def make_manifest_entry_json(**overrides): """Like `make_manifest_entry`, but with datetimes as ISO strings for use as raw HTTP JSON bodies.""" diff --git a/tests/test_api.py b/tests/test_api.py index dc4a95c..9c7ef39 100644 --- a/tests/test_api.py +++ b/tests/test_api.py @@ -20,7 +20,7 @@ from archivist.api import health_router from archivist.api import router as api_router from archivist.database import yield_session -from archivist.orm.archivequeue import ArchiveQueue +from archivist.orm.archive import Archive from archivist.settings import get_settings @@ -81,7 +81,7 @@ def test_archive_persists_queue_item(self): response = self.client.post("/api/v1/archive", json=payload) manifest_id = response.json()["manifest_id"] - item = self.session.query(ArchiveQueue).filter_by(manifest_id=manifest_id).one() + item = self.session.query(Archive).filter_by(manifest_id=manifest_id).one() self.assertFalse(item.consumed) self.assertFalse(item.completed) diff --git a/tests/test_archive.py b/tests/test_archive.py index 282b083..5858262 100644 --- a/tests/test_archive.py +++ b/tests/test_archive.py @@ -8,12 +8,12 @@ from conftest import make_manifest_entry -from archivist.core.archive import Archive +from archivist.core.archive_job import ArchiveJob class TestArchiveProperties(unittest.TestCase): def test_defaults_are_none(self): - archive = Archive() + archive = ArchiveJob() self.assertIsNone(archive.manifest) self.assertIsNone(archive.manifest_id) self.assertIsNone(archive.local_root) @@ -22,7 +22,7 @@ def test_defaults_are_none(self): def test_constructor_sets_properties(self): manifest = {"store_files": []} - archive = Archive( + archive = ArchiveJob( manifest=manifest, manifest_id="abc", local_root="/local", @@ -35,7 +35,7 @@ def test_constructor_sets_properties(self): self.assertEqual(archive.archive_root, "/archive") def test_repr_includes_key_fields(self): - archive = Archive(manifest_id="abc", type="posix") + archive = ArchiveJob(manifest_id="abc", type="posix") text = repr(archive) self.assertIn("abc", text) self.assertIn("posix", text) @@ -48,7 +48,7 @@ def test_builds_src_dst_pairs_relative_to_local_root(self): make_manifest_entry(instance_path="/local/sub/file.txt"), ] } - archive = Archive( + archive = ArchiveJob( manifest=manifest, local_root="/local", archive_root="/archive", @@ -69,24 +69,24 @@ def test_multiple_entries_preserve_order(self): make_manifest_entry(instance_path="/local/b.txt"), ] } - archive = Archive(manifest=manifest, local_root="/local", archive_root="/archive", type="posix") + archive = ArchiveJob(manifest=manifest, local_root="/local", archive_root="/archive", type="posix") pairs = archive.create_archive() self.assertEqual([str(src) for src, _ in pairs], ["/local/a.txt", "/local/b.txt"]) def test_empty_store_files_returns_empty_list(self): - archive = Archive(manifest={"store_files": []}, local_root="/local", archive_root="/archive", type="posix") + archive = ArchiveJob(manifest={"store_files": []}, local_root="/local", archive_root="/archive", type="posix") self.assertEqual(archive.create_archive(), []) class TestCreateArchiveOtherTypes(unittest.TestCase): def test_hpss_returns_none(self): - archive = Archive(manifest={"store_files": []}, type="hpss") + archive = ArchiveJob(manifest={"store_files": []}, type="hpss") self.assertIsNone(archive.create_archive()) def test_unknown_type_returns_none(self): - archive = Archive(manifest={"store_files": []}, type="something-else") + archive = ArchiveJob(manifest={"store_files": []}, type="something-else") self.assertIsNone(archive.create_archive()) diff --git a/tests/test_archivequeue.py b/tests/test_archivequeue.py index 630bf16..69589ea 100644 --- a/tests/test_archivequeue.py +++ b/tests/test_archivequeue.py @@ -1,13 +1,14 @@ # Copyright (c) 2025-2026 Simons Observatory. # Full license can be found in the top level "LICENSE" file. -"""Tests for archivist.orm.archivequeue.ArchiveQueue, against a real sqlite DB.""" +"""Tests for archivist.orm.archive.Archive, against a real sqlite DB.""" import unittest import pytest +from conftest import make_archive_item -from archivist.orm.archivequeue import ArchiveQueue +from archivist.orm import Archive, Manifest @pytest.mark.usefixtures("db_session") @@ -21,44 +22,41 @@ def _inject_session(self, db_session): class TestNewItem(TestArchiveQueueBase): def test_new_item_defaults(self): - item = ArchiveQueue.new_item(manifest_id="m1", manifest="{}", paths=["/a"], root="/root") - self.assertEqual(item.manifest_id, "m1") - self.assertEqual(item.paths, ["/a"]) - self.assertEqual(item.root, "/root") + manifest = Manifest.get_or_create(self.session, manifest_id="m1", librarian_name="lib") + item = Archive.new_item(manifest=manifest, archive_root="/root") + self.assertIs(item.manifest, manifest) + self.assertEqual(item.archive_root, "/root") self.assertEqual(item.retries, 0) self.assertFalse(item.consumed) self.assertFalse(item.completed) self.assertFalse(item.failed) def test_new_item_persists(self): - item = ArchiveQueue.new_item(manifest_id="m1", manifest="{}", paths=["/a"], root="/root") - self.session.add(item) - self.session.commit() + make_archive_item(self.session, manifest_id="m1", archive_root="/root") - fetched = self.session.query(ArchiveQueue).filter_by(manifest_id="m1").one() - self.assertEqual(fetched.root, "/root") + fetched = self.session.query(Archive).filter_by(manifest_id="m1").one() + self.assertEqual(fetched.archive_root, "/root") def test_manifest_id_must_be_unique(self): from sqlalchemy.exc import IntegrityError - self.session.add(ArchiveQueue.new_item(manifest_id="dup", manifest="{}", paths=[], root="/root")) - self.session.commit() + make_archive_item(self.session, manifest_id="dup", archive_root="/root") - self.session.add(ArchiveQueue.new_item(manifest_id="dup", manifest="{}", paths=[], root="/root")) + # A second archive job on the same manifest violates the unique FK (1:1). + manifest = Manifest.get_or_create(self.session, manifest_id="dup", librarian_name="lib") + self.session.add(Archive.new_item(manifest=manifest, archive_root="/root")) with self.assertRaises(IntegrityError): self.session.commit() class TestDequeue(TestArchiveQueueBase): def test_dequeue_returns_none_when_empty(self): - self.assertIsNone(ArchiveQueue.dequeue(self.session)) + self.assertIsNone(Archive.dequeue(self.session)) def test_dequeue_marks_item_consumed(self): - item = ArchiveQueue.new_item(manifest_id="m1", manifest="{}", paths=[], root="/root") - self.session.add(item) - self.session.commit() + make_archive_item(self.session, manifest_id="m1") - dequeued = ArchiveQueue.dequeue(self.session) + dequeued = Archive.dequeue(self.session) self.assertEqual(dequeued.manifest_id, "m1") self.assertTrue(dequeued.consumed) @@ -67,35 +65,25 @@ def test_dequeue_marks_item_consumed(self): def test_dequeue_returns_oldest_first(self): import time - first = ArchiveQueue.new_item(manifest_id="first", manifest="{}", paths=[], root="/root") - self.session.add(first) - self.session.commit() - + make_archive_item(self.session, manifest_id="first") time.sleep(0.01) + make_archive_item(self.session, manifest_id="second") - second = ArchiveQueue.new_item(manifest_id="second", manifest="{}", paths=[], root="/root") - self.session.add(second) - self.session.commit() - - dequeued = ArchiveQueue.dequeue(self.session) + dequeued = Archive.dequeue(self.session) self.assertEqual(dequeued.manifest_id, "first") def test_dequeue_skips_already_consumed_items(self): - item = ArchiveQueue.new_item(manifest_id="m1", manifest="{}", paths=[], root="/root") - self.session.add(item) - self.session.commit() + make_archive_item(self.session, manifest_id="m1") - ArchiveQueue.dequeue(self.session) + Archive.dequeue(self.session) - self.assertIsNone(ArchiveQueue.dequeue(self.session)) + self.assertIsNone(Archive.dequeue(self.session)) class TestCompleteAndFail(TestArchiveQueueBase): def test_complete_sets_flags(self): - item = ArchiveQueue.new_item(manifest_id="m1", manifest="{}", paths=[], root="/root") - self.session.add(item) - self.session.commit() + item = make_archive_item(self.session, manifest_id="m1") item.complete(self.session) @@ -104,9 +92,7 @@ def test_complete_sets_flags(self): self.assertIsNotNone(item.completed_time) def test_fail_sets_flags(self): - item = ArchiveQueue.new_item(manifest_id="m1", manifest="{}", paths=[], root="/root") - self.session.add(item) - self.session.commit() + item = make_archive_item(self.session, manifest_id="m1") item.fail(self.session) diff --git a/tests/test_classes.py b/tests/test_classes.py index 1fdc8ac..722ac22 100644 --- a/tests/test_classes.py +++ b/tests/test_classes.py @@ -6,7 +6,7 @@ import unittest import archivist -from archivist import Archive, RegistryLibrarian, RegistrySqlite, StorageDisk, StorageHPSS +from archivist import ArchiveJob, RegistryLibrarian, RegistrySqlite, StorageDisk, StorageHPSS class TestPackageImports(unittest.TestCase): @@ -14,13 +14,13 @@ def test_version_is_exposed(self): self.assertIsInstance(archivist.__version__, str) def test_construction(self): - archive = Archive() + archive_job = ArchiveJob() reg_lib = RegistryLibrarian() reg_sql = RegistrySqlite() store_disk = StorageDisk() store_hpss = StorageHPSS() - self.assertIsInstance(archive, Archive) + self.assertIsInstance(archive_job, ArchiveJob) self.assertIsInstance(reg_lib, RegistryLibrarian) self.assertIsInstance(reg_sql, RegistrySqlite) self.assertIsInstance(store_disk, StorageDisk) diff --git a/tests/test_storage_disk.py b/tests/test_storage_disk.py index 8dc0a2f..0665098 100644 --- a/tests/test_storage_disk.py +++ b/tests/test_storage_disk.py @@ -8,7 +8,7 @@ import pytest from conftest import make_manifest_entry -from archivist.core.archive import Archive +from archivist.core.archive_job import ArchiveJob from archivist.storage.storage_disk import StorageDisk @@ -27,7 +27,7 @@ def test_store_copies_file_to_archive_root(self): source.write_text("hello") manifest = {"store_files": [make_manifest_entry(instance_path=str(source))]} - archive = Archive( + archive = ArchiveJob( manifest=manifest, local_root=str(self.local_root), archive_root=str(self.archive_root), @@ -49,7 +49,7 @@ def test_store_copies_nested_file_preserving_relative_path(self): source.write_text("nested-content") manifest = {"store_files": [make_manifest_entry(instance_path=str(source))]} - archive = Archive( + archive = ArchiveJob( manifest=manifest, local_root=str(self.local_root), archive_root=str(self.archive_root), @@ -69,7 +69,7 @@ def test_store_returns_self_with_archive_and_future(self): source.write_text("hello") manifest = {"store_files": [make_manifest_entry(instance_path=str(source))]} - archive = Archive( + archive = ArchiveJob( manifest=manifest, local_root=str(self.local_root), archive_root=str(self.archive_root), @@ -92,7 +92,7 @@ def test_store_copies_directory_tree(self): (src_dir / "inner.txt").write_text("inner-content") manifest = {"store_files": [make_manifest_entry(instance_path=str(src_dir))]} - archive = Archive( + archive = ArchiveJob( manifest=manifest, local_root=str(self.local_root), archive_root=str(self.archive_root), diff --git a/tests/test_tasks_archive.py b/tests/test_tasks_archive.py index 1df5ca5..7ff8b88 100644 --- a/tests/test_tasks_archive.py +++ b/tests/test_tasks_archive.py @@ -9,10 +9,9 @@ import unittest.mock import pytest -from conftest import make_manifest_entry +from conftest import make_archive_item, make_manifest_entry -from archivist.core.models import ManifestRequest -from archivist.orm.archivequeue import ArchiveQueue +from archivist.orm.archive import Archive from archivist.queue import get_status_queue from archivist.tasks.archive import process_status_queue, start_archive @@ -27,19 +26,12 @@ def _inject(self, db_session, use_settings, local_root, archive_root): self.archive_root = archive_root def _enqueue_manifest_for(self, source_path, manifest_id="m1"): - request = ManifestRequest( - librarian_name="test-librarian", - store_files=[make_manifest_entry(instance_path=str(source_path))], - ) - item = ArchiveQueue.new_item( + return make_archive_item( + self.session, manifest_id=manifest_id, - manifest=request.model_dump_json(), - paths=[str(source_path)], - root=str(self.archive_root), + archive_root=str(self.archive_root), + entries=[make_manifest_entry(instance_path=str(source_path))], ) - self.session.add(item) - self.session.commit() - return item class TestStartArchive(TestTasksArchiveBase): @@ -56,7 +48,7 @@ def test_start_archive_dequeues_and_enqueues_status(self): self.assertEqual(get_status_queue().size, 1) - item = self.session.query(ArchiveQueue).filter_by(manifest_id="m1").one() + item = self.session.query(Archive).filter_by(manifest_id="m1").one() self.assertTrue(item.consumed) def test_start_archive_actually_copies_file(self): @@ -89,7 +81,7 @@ def test_process_status_queue_completes_finished_task(self): while time.time() < deadline: process_status_queue() self.session.expire_all() - item = self.session.query(ArchiveQueue).filter_by(manifest_id="m1").one() + item = self.session.query(Archive).filter_by(manifest_id="m1").one() if item.completed: break time.sleep(0.05) @@ -150,11 +142,11 @@ def test_process_status_queue_marks_item_failed_when_complete_raises(self): time.sleep(0.05) get_status_queue().enqueue(task) - with unittest.mock.patch.object(ArchiveQueue, "complete", side_effect=RuntimeError("boom")): + with unittest.mock.patch.object(Archive, "complete", side_effect=RuntimeError("boom")): process_status_queue() self.session.expire_all() - item = self.session.query(ArchiveQueue).filter_by(manifest_id="m1").one() + item = self.session.query(Archive).filter_by(manifest_id="m1").one() self.assertTrue(item.completed) self.assertTrue(item.failed) From 7be34efa1db1bb1c55b6f145b4bccc9510ea2fae Mon Sep 17 00:00:00 2001 From: Ioannis Paraskevakos Date: Mon, 13 Jul 2026 13:47:56 -0400 Subject: [PATCH 2/2] chore: remove dead comments --- src/archivist/api/extract.py | 3 --- src/archivist/orm/archive.py | 10 ++++------ src/archivist/orm/manifest.py | 2 +- src/archivist/tasks/archive.py | 3 +-- 4 files changed, 6 insertions(+), 12 deletions(-) diff --git a/src/archivist/api/extract.py b/src/archivist/api/extract.py index ae6ec02..7ccacb3 100644 --- a/src/archivist/api/extract.py +++ b/src/archivist/api/extract.py @@ -6,8 +6,6 @@ from archivist.api import router from archivist.core.archive_job import ArchiveJob from archivist.core.models import ManifestFailedResponse, ManifestRequest, ManifestResponse - -# from archivist.queue import Queue, get_extract_queue from archivist.settings import Settings, get_settings @@ -15,7 +13,6 @@ def extract( manifest_request: ManifestRequest, response: Response, - # queue: Queue = Depends(get_extract_queue), settings: Settings = Depends(get_settings), ): """ diff --git a/src/archivist/orm/archive.py b/src/archivist/orm/archive.py index 058c29e..6940e9b 100644 --- a/src/archivist/orm/archive.py +++ b/src/archivist/orm/archive.py @@ -24,7 +24,7 @@ class Archive(db.Base): __tablename__ = "archive" - id = db.Column(db.String(36), primary_key=True) # archivist-minted uuid4, decision 1 + id = db.Column(db.String(36), primary_key=True) # archivist-minted uuid4 "Archivist-minted uuid4 identifying this archive job." created_time = db.Column(db.DateTime, nullable=False) "The time this job was added to the queue." @@ -35,18 +35,18 @@ class Archive(db.Base): db.String(36), db.ForeignKey("manifest.id", ondelete="CASCADE"), nullable=False, - unique=True, # 1:1, decision 5 + unique=True, # 1:1 ) "FK to the manifest being archived (unique -> 1:1)." manifest = db.relationship("Manifest", back_populates="archive") "The manifest being archived." - archive_root = db.Column(db.String(512), nullable=False) # renamed from `root` + archive_root = db.Column(db.String(512), nullable=False) "The archive root at submit time; snapshotted since settings can change." archive_path = db.Column(db.String(2048), nullable=True) "The destination path, set on completion." - librarian_archive_id = db.Column(db.String(2048), nullable=True, unique=True) # decision 2 + librarian_archive_id = db.Column(db.String(2048), nullable=True, unique=True) "Site-specific identifier supplied by the Librarian callback contract." consumed = db.Column(db.Boolean, default=False) @@ -72,8 +72,6 @@ def new_item(cls, manifest: "Manifest", archive_root: str) -> "Archive": @classmethod def dequeue(cls, session: Session) -> "Archive | None": - # Mirror librarian's consume_queue_item: lock-and-skip so workers never - # grab the same row, and filter on BOTH flags (consumed AND completed). stmt = ( select(cls) .options(joinedload(cls.manifest).selectinload(Manifest.entries)) diff --git a/src/archivist/orm/manifest.py b/src/archivist/orm/manifest.py index dad1b30..c67d63d 100644 --- a/src/archivist/orm/manifest.py +++ b/src/archivist/orm/manifest.py @@ -36,7 +36,7 @@ class Manifest(db.Base): cascade="all, delete-orphan", passive_deletes=True, ) - "The single archive job for this manifest (1:1, decision 5)." + "The single archive job for this manifest." @classmethod def get_or_create(cls, session, manifest_id, librarian_name) -> "Manifest": diff --git a/src/archivist/tasks/archive.py b/src/archivist/tasks/archive.py index 6ad7dd5..a8b123e 100644 --- a/src/archivist/tasks/archive.py +++ b/src/archivist/tasks/archive.py @@ -26,8 +26,7 @@ def start_archive( route), so dependencies are resolved here rather than via `Depends`. Takes ``session_maker`` (rather than calling ``get_session`` inline) so a future task-based runner can inject it; returns ``True`` when a job was - picked up and ``False`` when the queue was empty, mirroring librarian's - ``consume_queue_item``. + picked up and ``False`` when the queue was empty. """ status_queue = get_status_queue() settings = get_settings()