Feat/worker threads

pyproject.toml

@@ -110,5 +110,6 @@ fixture-parentheses = false
 mark-parentheses = false
 
 [tool.isort]
+profile = "black"
 line_length = 120
 skip = ["_version.py"]

src/archivist/__init__.py

@@ -6,4 +6,4 @@
 from ._version import __version__
 from .core.archive import Archive
 from .core.registry import RegistryLibrarian, RegistrySqlite
-from .core.storage import StorageDisk, StorageHPSS
+from .storage import StorageDisk, StorageHPSS

src/archivist/__main__.py

@@ -4,6 +4,7 @@
 from pathlib import Path
 
 import click
+import loguru
 import requests
 
 from archivist.core.models import ManifestEntry, ManifestFailedResponse, ManifestRequest, ManifestResponse
@@ -34,12 +35,6 @@ def main(ctx, config):
 @click.option(
     "--source-path", type=click.Path(exists=True), required=True, help="The source path of the file to archive"
 )
-@click.option(
-    "--dest-path",
-    type=click.Path(exists=False),
-    required=True,
-    help="The destination path where the file should be archived",
-)
 @click.option(
     "--librarian-name",
     type=str,
@@ -48,18 +43,17 @@ def main(ctx, config):
     default="librarian",
 )
 @click.pass_context
-def archive(ctx, source_path, dest_path, librarian_name):
+def archive(ctx, source_path, librarian_name):
     """Command to archive a file.
     For example: `archivist -c config archive "--source-path /path/to/source --dest-path /path/to/dest"`
     """
     # Here you would implement the logic to handle the archiving process
     # For now, we will just print the source and destination paths
-    click.echo(f"Archiving file from {source_path} to {dest_path}")
+    click.echo(f"Archiving file from {source_path}")
 
     settings = get_settings()
 
     source_root = Path(source_path)
-    dest_root = Path(dest_path) if dest_path is not None else "source_root"
 
     if source_root.is_file():
         files = [source_root]
@@ -74,7 +68,7 @@ def archive(ctx, source_path, dest_path, librarian_name):
     for file_path in files:
         stat = file_path.stat()
         relative_path = file_path.relative_to(walk_root)
-
+        loguru.logger.info(f"Archiving file: {file_path}, relative path: {relative_path}")
         store_files.append(
             ManifestEntry(
                 name=file_path.name,
@@ -83,7 +77,7 @@ def archive(ctx, source_path, dest_path, librarian_name):
                 checksum=_checksum(file_path),
                 uploader=uploader,
                 source=str(source_root),
-                instance_path=str(dest_root / relative_path),
+                instance_path=str(file_path.absolute()),
                 instance_create_time=datetime.now(),
                 instance_available=True,
                 outgoing_transfer_id=0,

src/archivist/api/archive.py

@@ -2,18 +2,20 @@
 
 from fastapi import Depends, Response
 from loguru import logger
+from sqlalchemy.orm import Session
 
 from archivist.api import router
-from archivist.core.models import Archive, ManifestFailedResponse, ManifestRequest, ManifestResponse
-from archivist.queue import ArchiveQueue, get_archive_queue
+from archivist.core.models import ManifestFailedResponse, ManifestRequest, ManifestResponse
+from archivist.database import yield_session
+from archivist.orm.archivequeue import ArchiveQueue
 from archivist.settings import Settings, get_settings
 
 
 @router.post("/archive", response_model=ManifestResponse | ManifestFailedResponse)
 def archive(
     manifest_request: ManifestRequest,
     response: Response,
-    queue: ArchiveQueue = Depends(get_archive_queue),
+    session: Session = Depends(yield_session),
     settings: Settings = Depends(get_settings),
 ):
     """
@@ -36,15 +38,14 @@ def archive(
         f"Archiving manifest {manifest_id} from librarian '{manifest_request.librarian_name}': {len(manifest_request.store_files)} file(s), {total_size} bytes total"
     )
 
-    archive = Archive(
+    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.storage_root,  # Example root path
+        root=settings.archive_root,
     )
-
-    # TODO: Add the archive to a queue for processing
-    # queue.enqueue_archive(archive)
+    session.add(item)
+    session.commit()
 
     return ManifestResponse(
         manifest_id=str(manifest_id),

src/archivist/api/extract.py

@@ -4,16 +4,18 @@
 from fastapi import Depends, Response
 
 from archivist.api import router
-from archivist.core.models import Archive, ManifestFailedResponse, ManifestRequest, ManifestResponse
-from archivist.queue import ExtractQueue, get_extract_queue
+from archivist.core.archive import Archive
+from archivist.core.models import ManifestFailedResponse, ManifestRequest, ManifestResponse
+
+# from archivist.queue import Queue, get_extract_queue
 from archivist.settings import Settings, get_settings
 
 
 @router.post("/extract", response_model=ManifestResponse | ManifestFailedResponse)
 def extract(
     manifest_request: ManifestRequest,
     response: Response,
-    queue: ExtractQueue = Depends(get_extract_queue),
+    # queue: Queue = Depends(get_extract_queue),
     settings: Settings = Depends(get_settings),
 ):
     """

src/archivist/core/__init__.py

@@ -1,3 +1 @@
-from .storage import StorageDisk, StorageHPSS
-
-storage_factory = {"hpss:": StorageHPSS, "posix": StorageDisk}
+from archivist.storage import StorageDisk, StorageHPSS, storage_factory  # noqa: F401

src/archivist/core/archive.py

@@ -2,7 +2,9 @@
 # Full license can be found in the top level "LICENSE" file.
 """Classes for working with archives."""
 
-import subprocess
+from pathlib import Path
+
+from .models import ManifestRequest
 
 
 class Archive(object):
@@ -16,38 +18,60 @@ class Archive(object):
     Args:
         manifest (str):  Path to a manifest file listing the files / directories
             contained in the archive.
-        paths (list):  Alternatively list the full set of files / directories contained
-            in the archive.
         root (str):  The common root filesystem location for building relative paths
             for all objects.
 
     """
 
-    def __init__(self, manifest=None, paths=None, root=None, type=None):
+    def __init__(
+        self,
+        manifest: dict | None = None,
+        manifest_id: str | None = None,
+        local_root: str | None = None,
+        archive_root: str | None = None,
+        type: str | None = None,
+    ):
         self._manifest = manifest
-        self._paths = paths
-        self._root = root
+        self._manifest_id = manifest_id
+        self._local_root = local_root
+        self._archive_root = archive_root
         self._type = type
         self._checksum = None  # Placeholder for checksum value
 
     @property
-    def paths(self):
-        return self._paths
+    def local_root(self):
+        return self._local_root
 
     @property
-    def root(self):
-        return self._root
+    def archive_root(self):
+        return self._archive_root
 
     @property
     def checksum(self):
-        """Return the checksum of the archive."""
+        return self._checksum
+
+    @property
+    def manifest(self):
+        return self._manifest
+
+    @property
+    def manifest_id(self):
+        return self._manifest_id
+
+    def __repr__(self):
+        return f"<Archive manifest_id={self._manifest_id} type={self._type}>, <manifest={self._manifest}>, <local_root={self._local_root}>, <archive_root={self._archive_root}>"
 
     def create_archive(self):
         """Create the archive based on the manifest or paths."""
         # Placeholder for archive creation logic
         if self._type == "posix":
-            # Implement logic for creating a POSIX archive
-            pass
+            src_dst_paths = []
+            for entry in self._manifest["store_files"]:
+                src_path = Path(entry["instance_path"])
+                dst_path = Path(self._archive_root) / src_path.relative_to(self._local_root)
+                src_dst_paths.append((src_path, dst_path))
+            return src_dst_paths
+
         elif self._type == "hpss":
             # Implement logic for creating an HPSS archive
             # Call subprocess to create the archive using HPSS commands

src/archivist/core/models.py

@@ -48,17 +48,3 @@ class ManifestResponse(BaseModel):
 class ManifestFailedResponse(BaseModel):
     error: str
     "The error message indicating why the manifest request failed."
-
-
-class Archive(BaseModel):
-    manifest_id: str
-    "The ID of the manifest associated with this archive."
-
-    manifest: str
-    "Path to a manifest file listing the files / directories contained in the archive."
-
-    paths: list[str]
-    "Alternatively list the full set of files / directories contained in the archive."
-
-    root: str
-    "The common root filesystem location for building relative paths for all objects."

src/archivist/core/storage.py

@@ -2,12 +2,15 @@
 # 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):
-        pass
+    def __init__(self, settings: Settings | None = None):
+        self._settings = settings
 
     def _store(self, archive):
         raise NotImplementedError("Fell through to base class")
@@ -62,11 +65,12 @@ class StorageDisk(Storage):
 
     """
 
-    def __init__(self, directory):
-        self._directory = directory
-        super().__init__()
+    def __init__(self, settings: Settings | None = None):
+        super().__init__(settings=settings)
+        self._storage_proc = None
+        self._archive = None
 
-    def _store(self, archive):
+    def _store(self, archive: Archive):
         """Tar the archive files into the storage directory."""
         pass
 

src/archivist/database.py

@@ -0,0 +1,98 @@
+"""
+Core database runner for SQLAlchemy.
+"""
+
+from typing import Generator
+
+from loguru import logger
+from sqlalchemy import (
+    BigInteger,
+    Boolean,
+    Column,
+    DateTime,
+    Enum,
+    ForeignKey,
+    Integer,
+    PickleType,
+    String,
+    create_engine,
+)
+from sqlalchemy.orm import Session, declarative_base, sessionmaker
+
+from .settings import get_settings
+
+# Engine and sessionmaker are created lazily, on first use, rather than at
+# import time. Settings (in particular the database path / ARCHIVIST_CONFIG_PATH)
+# may not be finalized yet when this module is first imported -- e.g. the CLI
+# sets ARCHIVIST_CONFIG_PATH from `--config` inside a click callback, which runs
+# after all top-level imports have already happened.
+_engine = None
+_SessionMaker = None
+
+
+def get_engine():
+    """
+    Returns the (lazily created) SQLAlchemy engine, built from the current settings.
+    """
+
+    global _engine
+
+    if _engine is None:
+        settings = get_settings()
+        logger.info("Starting database engine.")
+        _engine = create_engine(
+            settings.sqlalchemy_database_uri,
+            # Required for async and SQLite
+            connect_args=({"check_same_thread": False} if "sqlite" in settings.sqlalchemy_database_uri else {}),
+        )
+
+    return _engine
+
+
+def get_sessionmaker() -> sessionmaker:
+    """
+    Returns the (lazily created) sessionmaker, bound to the lazily created engine.
+    """
+
+    global _SessionMaker
+
+    if _SessionMaker is None:
+        logger.info("Creating database session.")
+        _SessionMaker = sessionmaker(bind=get_engine(), autocommit=False, autoflush=False)
+
+    return _SessionMaker
+
+
+def yield_session() -> "Generator[Session, None, None]":
+    """
+    Yields a new databse session.
+    """
+
+    session = get_sessionmaker()()
+    try:
+        yield session
+    finally:
+        session.close()
+
+
+def get_session() -> "Session":
+    """
+    Returns a new database session. Unlike yield_session, it is
+    your responsibility to close the session.
+    """
+
+    return get_sessionmaker()()
+
+
+Base = declarative_base()
+
+
+def create_all() -> None:
+    """
+    Create all tables that don't already exist. Imports the ORM models
+    first so their table definitions are registered with `Base`.
+    """
+
+    from . import orm  # noqa: F401
+
+    Base.metadata.create_all(get_engine())

src/archivist/orm/__init__.py

@@ -0,0 +1 @@
+from .archivequeue import ArchiveQueue  # noqa: F401