diff --git a/bbconf/alembic/versions/c7f3a9e1b204_added_analysis_files_table.py b/bbconf/alembic/versions/c7f3a9e1b204_added_analysis_files_table.py new file mode 100644 index 0000000..b11229e --- /dev/null +++ b/bbconf/alembic/versions/c7f3a9e1b204_added_analysis_files_table.py @@ -0,0 +1,99 @@ +"""Added analysis_files table + +Revision ID: c7f3a9e1b204 +Revises: 845d978eac7d +Create Date: 2026-08-17 21:00:00.000000 + +""" + +from typing import Sequence, Union + +import sqlalchemy as sa +from alembic import op +from sqlalchemy.dialects import postgresql + +# revision identifiers, used by Alembic. +revision: str = "c7f3a9e1b204" +down_revision: Union[str, None] = "845d978eac7d" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + """Upgrade schema.""" + op.create_table( + "analysis_files", + sa.Column("id", sa.Integer(), autoincrement=True, nullable=False), + sa.Column( + "name", + sa.String(), + nullable=False, + comment="Logical name/key, e.g. openSignalMatrix", + ), + sa.Column( + "file_path", + sa.String(), + nullable=False, + comment="S3 object key, relative to the bucket root", + ), + sa.Column( + "file_type", + sa.String(), + nullable=True, + comment="Category, e.g. openSignalMatrix | reference | model", + ), + sa.Column( + "genome", + sa.String(), + nullable=True, + comment="Genome/assembly, e.g. hg38 (optional)", + ), + sa.Column("description", sa.String(), nullable=True), + sa.Column( + "tags", + postgresql.ARRAY(sa.String()), + nullable=True, + comment="Free-form tags", + ), + sa.Column( + "file_size", + sa.Integer(), + nullable=True, + comment="Size of the file in bytes", + ), + sa.Column("checksum", sa.String(), nullable=True, comment="SHA256 of the file"), + sa.Column( + "creation_date", + sa.TIMESTAMP(timezone=True), + nullable=False, + comment="Upload date", + ), + sa.PrimaryKeyConstraint("id"), + ) + op.create_index( + op.f("ix_analysis_files_id"), "analysis_files", ["id"], unique=False + ) + op.create_index( + op.f("ix_analysis_files_name"), "analysis_files", ["name"], unique=False + ) + op.create_index( + op.f("ix_analysis_files_file_type"), + "analysis_files", + ["file_type"], + unique=False, + ) + op.create_index( + op.f("ix_analysis_files_genome"), + "analysis_files", + ["genome"], + unique=False, + ) + + +def downgrade() -> None: + """Downgrade schema.""" + op.drop_index(op.f("ix_analysis_files_genome"), table_name="analysis_files") + op.drop_index(op.f("ix_analysis_files_file_type"), table_name="analysis_files") + op.drop_index(op.f("ix_analysis_files_name"), table_name="analysis_files") + op.drop_index(op.f("ix_analysis_files_id"), table_name="analysis_files") + op.drop_table("analysis_files") diff --git a/bbconf/bbagent.py b/bbconf/bbagent.py index 571e2ff..e4c94ac 100644 --- a/bbconf/bbagent.py +++ b/bbconf/bbagent.py @@ -35,6 +35,7 @@ UsageModel, UsageStats, ) +from bbconf.modules.analysis_files import BedAgentAnalysisFile from bbconf.modules.bedfiles import BedAgentBedFile from bbconf.modules.bedsets import BedAgentBedSet from bbconf.modules.objects import BBObjects @@ -66,6 +67,7 @@ def __init__( self._bedset = BedAgentBedSet(self.config) self._objects = BBObjects(self.config) self._snapshot = BedAgentSnapshot(self.config) + self._analysis_files = BedAgentAnalysisFile(self.config) # get_stats() runs three uncached COUNT queries on the multi-hundred- # thousand-row bed table and is called on hot paths (the stats endpoint @@ -91,6 +93,10 @@ def objects(self) -> BBObjects: def snapshot(self) -> BedAgentSnapshot: return self._snapshot + @property + def analysis_files(self) -> BedAgentAnalysisFile: + return self._analysis_files + def __repr__(self) -> str: repr = f"BedBaseAgent(config={self.config})" repr += f"\n{self.bed}" diff --git a/bbconf/db_utils.py b/bbconf/db_utils.py index 535fc55..8dbf0ba 100644 --- a/bbconf/db_utils.py +++ b/bbconf/db_utils.py @@ -718,6 +718,48 @@ class BedSnapshot(Base): ) +class AnalysisFile(Base): + """ + Registry of standalone analysis files (openSignalMatrix, models, other + analysis inputs) stored in S3. Not tied to any bed file or bedset. + + Append-only: one row per uploaded file, so name-based lookups resolve the + newest matching row (same model as ``bed_snapshots``). This is a new table, + so ``Base.metadata.create_all()`` creates it on the next connection. + """ + + __tablename__ = "analysis_files" + + id: Mapped[int] = mapped_column(primary_key=True, index=True, autoincrement=True) + name: Mapped[str] = mapped_column( + nullable=False, index=True, comment="Logical name/key, e.g. openSignalMatrix" + ) + file_path: Mapped[str] = mapped_column( + nullable=False, comment="S3 object key, relative to the bucket root" + ) + file_type: Mapped[Optional[str]] = mapped_column( + nullable=True, + index=True, + comment="Category, e.g. openSignalMatrix | reference | model", + ) + genome: Mapped[Optional[str]] = mapped_column( + nullable=True, index=True, comment="Genome/assembly, e.g. hg38 (optional)" + ) + description: Mapped[Optional[str]] = mapped_column(nullable=True) + tags: Mapped[Optional[list]] = mapped_column( + ARRAY(String), nullable=True, comment="Free-form tags" + ) + file_size: Mapped[Optional[int]] = mapped_column( + nullable=True, comment="Size of the file in bytes" + ) + checksum: Mapped[Optional[str]] = mapped_column( + nullable=True, comment="SHA256 of the file" + ) + creation_date: Mapped[datetime.datetime] = mapped_column( + default=deliver_update_date, comment="Upload date" + ) + + class BaseEngine: """ A class with base methods, that are used in several classes. diff --git a/bbconf/exceptions.py b/bbconf/exceptions.py index 6991c99..ccb0b37 100644 --- a/bbconf/exceptions.py +++ b/bbconf/exceptions.py @@ -77,6 +77,13 @@ class SnapshotNotFoundError(BedBaseConfError): pass +class AnalysisFileNotFoundError(BedBaseConfError): + """ + Error type for missing analysis file""" + + pass + + class UniverseNotFoundError(BedBaseConfError): """ Error type for missing universe""" diff --git a/bbconf/models/base_models.py b/bbconf/models/base_models.py index 76a38f9..dd1a2a3 100644 --- a/bbconf/models/base_models.py +++ b/bbconf/models/base_models.py @@ -131,3 +131,36 @@ class BedSnapshotResult(BaseModel): class BedSnapshotListResult(BaseModel): count: int results: list[BedSnapshotResult] + + +class AnalysisFileArtifact(BaseModel): + """A standalone analysis file to publish (upload to S3 + record in the database).""" + + path: str # local file path to upload + name: str + file_type: str | None = None + genome: str | None = None + description: str | None = None + tags: list[str] | None = None + file_size: int | None = None + checksum: str | None = None + + +class AnalysisFileResult(BaseModel): + """One registered standalone analysis file.""" + + id: int | None = None + name: str + file_path: str + file_type: str | None = None + genome: str | None = None + description: str | None = None + tags: list[str] | None = None + file_size: int | None = None + checksum: str | None = None + creation_date: datetime.datetime + + +class AnalysisFileListResult(BaseModel): + count: int + results: list[AnalysisFileResult] diff --git a/bbconf/modules/analysis_files.py b/bbconf/modules/analysis_files.py new file mode 100644 index 0000000..1777a40 --- /dev/null +++ b/bbconf/modules/analysis_files.py @@ -0,0 +1,298 @@ +import logging +import os +from datetime import datetime, timezone + +from sqlalchemy import func, select +from sqlalchemy.orm import Session + +from bbconf.config_parser import BedBaseConfig +from bbconf.const import PKG_NAME +from bbconf.db_utils import AnalysisFile +from bbconf.exceptions import AnalysisFileNotFoundError +from bbconf.models.base_models import ( + AnalysisFileArtifact, + AnalysisFileListResult, + AnalysisFileResult, +) + +_LOGGER = logging.getLogger(PKG_NAME) + +# All analysis files live under this single S3 prefix. Not configurable. +ANALYSIS_FILES_S3_PREFIX = "analysis_files" + + +class BedAgentAnalysisFile: + """ + Class that manages standalone analysis files (the ``analysis_files`` index). + + One row per uploaded file (openSignalMatrix, models, other analysis inputs). + These files are global: they are not tied to any bed file or bedset. Adding + a file always uploads it to S3 *and* records it in the database; both writes + live here in bbconf. This class also exposes read (``list`` / ``get`` / + ``get_by_name`` / ``get_by_filename``) and ``delete`` helpers. + """ + + def __init__(self, config: BedBaseConfig): + """ + Initialize BedAgentAnalysisFile. + + Args: + config: Config object. + """ + self.config = config + self._db_engine = self.config.db_engine + + def add( + self, + artifacts: AnalysisFileArtifact | list[AnalysisFileArtifact], + creation_date: datetime | None = None, + ) -> AnalysisFileListResult: + """ + Add analysis files: upload each to S3 and record it in the database. + + Every artifact is uploaded under the fixed ``analysis_files/`` prefix and + then recorded in ``analysis_files``. The index rows are written only + after all uploads succeed, so a partial upload never leaves dangling + rows. + + Args: + artifacts: One artifact or a list of them. Each carries the local + ``path`` to upload plus its ``name`` and file metadata. + creation_date: Upload date recorded on every row + (defaults to now, UTC). + + Returns: + The created analysis-file rows. + """ + if isinstance(artifacts, AnalysisFileArtifact): + artifacts = [artifacts] + if creation_date is None: + creation_date = datetime.now(timezone.utc) + + # Upload everything first; only record rows once all uploads succeed. + uploads: list[tuple[AnalysisFileArtifact, str]] = [] + for artifact in artifacts: + key = f"{ANALYSIS_FILES_S3_PREFIX}/{os.path.basename(artifact.path)}" + self.config.upload_s3(artifact.path, s3_path=key) + uploads.append((artifact, key)) + + results: list[AnalysisFileResult] = [] + with Session(self._db_engine.engine) as session: + for artifact, key in uploads: + row = AnalysisFile( + name=artifact.name, + file_path=key, + file_type=artifact.file_type, + genome=artifact.genome, + description=artifact.description, + tags=artifact.tags, + file_size=artifact.file_size, + checksum=artifact.checksum, + creation_date=creation_date, + ) + session.add(row) + session.flush() + results.append(self._to_result(row)) + session.commit() + + _LOGGER.info(f"Recorded {len(results)} rows in analysis_files") + return AnalysisFileListResult(count=len(results), results=results) + + def delete(self, id: int, remove_s3: bool = True) -> None: + """ + Delete an analysis-file index row. + + Args: + id: Primary key of the analysis-file row. + remove_s3: Also delete the underlying S3 object. + + Returns: + None. + + Raises: + AnalysisFileNotFoundError: If no row with this id exists. + """ + with Session(self._db_engine.engine) as session: + row = session.scalar(select(AnalysisFile).where(AnalysisFile.id == id)) + if row is None: + raise AnalysisFileNotFoundError( + f"Analysis file with id '{id}' not found." + ) + file_path = row.file_path + session.delete(row) + session.commit() + + if remove_s3: + self.config.delete_s3(file_path) + + def list( + self, + file_type: str | None = None, + genome: str | None = None, + tag: str | None = None, + limit: int | None = 100, + offset: int = 0, + ) -> AnalysisFileListResult: + """ + List analysis-file index rows in the database, newest first. + + Args: + file_type: Optional filter on file type. + genome: Optional filter on genome/assembly. + tag: Optional filter; keep only rows whose ``tags`` contain this tag. + limit: Maximum number of rows to return. ``None`` returns all rows. + offset: Number of rows to skip. + + Returns: + List of analysis files and the total matching count. + """ + statement = select(AnalysisFile) + count_statement = select(func.count()).select_from(AnalysisFile) + if file_type is not None: + statement = statement.where(AnalysisFile.file_type == file_type) + count_statement = count_statement.where(AnalysisFile.file_type == file_type) + if genome is not None: + statement = statement.where(AnalysisFile.genome == genome) + count_statement = count_statement.where(AnalysisFile.genome == genome) + if tag is not None: + statement = statement.where(AnalysisFile.tags.any(tag)) + count_statement = count_statement.where(AnalysisFile.tags.any(tag)) + statement = statement.order_by( + AnalysisFile.creation_date.desc(), AnalysisFile.id.desc() + ) + if limit is not None: + statement = statement.limit(limit).offset(offset) + elif offset: + statement = statement.offset(offset) + + with Session(self._db_engine.engine) as session: + total = session.execute(count_statement).scalar_one() + rows = session.scalars(statement).all() + results = [self._to_result(row) for row in rows] + + return AnalysisFileListResult(count=total, results=results) + + def get(self, id: int) -> AnalysisFileResult: + """ + Get a single analysis-file index row by id. + + Args: + id: Primary key of the analysis-file row. + + Returns: + The analysis-file row. + + Raises: + AnalysisFileNotFoundError: If no row with this id exists. + """ + with Session(self._db_engine.engine) as session: + row = session.scalar(select(AnalysisFile).where(AnalysisFile.id == id)) + if row is None: + raise AnalysisFileNotFoundError( + f"Analysis file with id '{id}' not found." + ) + return self._to_result(row) + + def get_by_name(self, name: str, genome: str | None = None) -> AnalysisFileResult: + """ + Resolve an analysis file by its logical name (newest matching row). + + Args: + name: Logical name/key, e.g. ``openSignalMatrix``. + genome: Optional genome/assembly to disambiguate, e.g. ``hg38``. + + Returns: + The newest matching analysis-file row. + + Raises: + AnalysisFileNotFoundError: If no row matches. + """ + statement = select(AnalysisFile).where(AnalysisFile.name == name) + if genome is not None: + statement = statement.where(AnalysisFile.genome == genome) + statement = statement.order_by( + AnalysisFile.creation_date.desc(), AnalysisFile.id.desc() + ) + with Session(self._db_engine.engine) as session: + row = session.scalars(statement).first() + if row is None: + raise AnalysisFileNotFoundError(f"Analysis file '{name}' not found.") + return self._to_result(row) + + def get_by_filename(self, filename: str) -> AnalysisFileResult: + """ + Resolve an analysis file by its file name (the basename of its S3 key). + + Returns the newest row whose ``file_path`` basename equals ``filename``. + Used to round-trip a DRS object-id back to its row. + + Args: + filename: The bare file name, e.g. ``openSignalMatrix_hg38.txt.gz``. + + Returns: + The matching analysis-file row. + + Raises: + AnalysisFileNotFoundError: If no row matches. + """ + filename = os.path.basename(filename) + with Session(self._db_engine.engine) as session: + rows = session.scalars( + select(AnalysisFile) + .where(AnalysisFile.file_path.like(f"%{filename}")) + .order_by(AnalysisFile.creation_date.desc(), AnalysisFile.id.desc()) + ).all() + for row in rows: + if os.path.basename(row.file_path) == filename: + return self._to_result(row) + raise AnalysisFileNotFoundError(f"Analysis file '{filename}' not found.") + + def delete_by_checksum(self, checksum: str, remove_s3: bool = True) -> None: + """ + Delete analysis-file index rows by their checksum. + + Deletes every ``analysis_files`` row whose ``checksum`` matches (a + checksum identifies one file's content) and optionally removes the + underlying S3 objects. + + Args: + checksum: SHA256 checksum of the analysis file. + remove_s3: Also delete the underlying S3 object(s). + + Returns: + None. + + Raises: + AnalysisFileNotFoundError: If no row matches the checksum. + """ + with Session(self._db_engine.engine) as session: + rows = session.scalars( + select(AnalysisFile).where(AnalysisFile.checksum == checksum) + ).all() + if not rows: + raise AnalysisFileNotFoundError( + f"Analysis file with checksum '{checksum}' not found." + ) + file_paths = {row.file_path for row in rows} + for row in rows: + session.delete(row) + session.commit() + + if remove_s3: + for file_path in file_paths: + self.config.delete_s3(file_path) + + @staticmethod + def _to_result(row: AnalysisFile) -> AnalysisFileResult: + return AnalysisFileResult( + id=row.id, + name=row.name, + file_path=row.file_path, + file_type=row.file_type, + genome=row.genome, + description=row.description, + tags=list(row.tags) if row.tags is not None else None, + file_size=row.file_size, + checksum=row.checksum, + creation_date=row.creation_date, + ) diff --git a/tests/test_analysis_files.py b/tests/test_analysis_files.py new file mode 100644 index 0000000..ad39674 --- /dev/null +++ b/tests/test_analysis_files.py @@ -0,0 +1,123 @@ +import pytest + +from bbconf.exceptions import AnalysisFileNotFoundError +from bbconf.models.base_models import AnalysisFileArtifact + +from .conftest import SERVICE_UNAVAILABLE +from .utils import ContextManagerDBTesting + +UPLOAD_TARGET = "bbconf.config_parser.bedbaseconfig.BedBaseConfig.upload_s3" +DELETE_TARGET = "bbconf.config_parser.bedbaseconfig.BedBaseConfig.delete_s3" + + +def _artifact(**overrides) -> AnalysisFileArtifact: + values = dict( + path="/local/openSignalMatrix_hg38.txt.gz", + name="openSignalMatrix", + file_type="openSignalMatrix", + genome="hg38", + description="Open signal matrix for hg38", + tags=["reference", "hg38"], + file_size=12345, + checksum="a" * 64, + ) + values.update(overrides) + return AnalysisFileArtifact(**values) + + +@pytest.mark.skipif(SERVICE_UNAVAILABLE, reason="Database is not available") +class Test_AnalysisFile_Agent: + def test_add(self, bbagent_obj, mocker): + upload_mock = mocker.patch(UPLOAD_TARGET, return_value=True) + with ContextManagerDBTesting(config=bbagent_obj.config, add_data=False): + result = bbagent_obj.analysis_files.add(_artifact()) + + assert upload_mock.called + assert result.count == 1 + row = result.results[0] + assert row.id is not None + assert row.name == "openSignalMatrix" + assert row.genome == "hg38" + assert row.tags == ["reference", "hg38"] + assert row.checksum == "a" * 64 + assert row.file_path == "analysis_files/openSignalMatrix_hg38.txt.gz" + + def test_list_and_filters(self, bbagent_obj, mocker): + mocker.patch(UPLOAD_TARGET, return_value=True) + with ContextManagerDBTesting(config=bbagent_obj.config, add_data=False): + bbagent_obj.analysis_files.add( + [ + _artifact(), + _artifact( + path="/local/openSignalMatrix_mm10.txt.gz", + genome="mm10", + tags=["reference", "mm10"], + checksum="b" * 64, + ), + _artifact( + path="/local/some_model.pt", + name="some_model", + file_type="model", + genome=None, + tags=["model"], + checksum="c" * 64, + ), + ] + ) + + assert bbagent_obj.analysis_files.list().count == 3 + assert ( + bbagent_obj.analysis_files.list(file_type="openSignalMatrix").count == 2 + ) + assert bbagent_obj.analysis_files.list(genome="mm10").count == 1 + assert bbagent_obj.analysis_files.list(tag="model").count == 1 + assert bbagent_obj.analysis_files.list(genome="hg19").count == 0 + + def test_get_and_get_by_name(self, bbagent_obj, mocker): + mocker.patch(UPLOAD_TARGET, return_value=True) + with ContextManagerDBTesting(config=bbagent_obj.config, add_data=False): + added = bbagent_obj.analysis_files.add(_artifact()).results[0] + + by_id = bbagent_obj.analysis_files.get(added.id) + assert by_id.name == "openSignalMatrix" + + by_name = bbagent_obj.analysis_files.get_by_name( + "openSignalMatrix", genome="hg38" + ) + assert by_name.id == added.id + + by_filename = bbagent_obj.analysis_files.get_by_filename( + "openSignalMatrix_hg38.txt.gz" + ) + assert by_filename.id == added.id + + def test_get_missing_raises(self, bbagent_obj): + with ContextManagerDBTesting(config=bbagent_obj.config, add_data=False): + with pytest.raises(AnalysisFileNotFoundError): + bbagent_obj.analysis_files.get(999999) + with pytest.raises(AnalysisFileNotFoundError): + bbagent_obj.analysis_files.get_by_name("does-not-exist") + + def test_delete(self, bbagent_obj, mocker): + mocker.patch(UPLOAD_TARGET, return_value=True) + delete_mock = mocker.patch(DELETE_TARGET, return_value=True) + with ContextManagerDBTesting(config=bbagent_obj.config, add_data=False): + added = bbagent_obj.analysis_files.add(_artifact()).results[0] + + bbagent_obj.analysis_files.delete(added.id) + assert delete_mock.called + assert bbagent_obj.analysis_files.list().count == 0 + with pytest.raises(AnalysisFileNotFoundError): + bbagent_obj.analysis_files.get(added.id) + + def test_delete_by_checksum(self, bbagent_obj, mocker): + mocker.patch(UPLOAD_TARGET, return_value=True) + delete_mock = mocker.patch(DELETE_TARGET, return_value=True) + with ContextManagerDBTesting(config=bbagent_obj.config, add_data=False): + bbagent_obj.analysis_files.add(_artifact()) + + bbagent_obj.analysis_files.delete_by_checksum("a" * 64) + assert delete_mock.called + assert bbagent_obj.analysis_files.list().count == 0 + with pytest.raises(AnalysisFileNotFoundError): + bbagent_obj.analysis_files.delete_by_checksum("a" * 64)