diff --git a/.github/pull_request_template.md b/.github/pull_request_template.md index a24b8bf0..fdc90a74 100644 --- a/.github/pull_request_template.md +++ b/.github/pull_request_template.md @@ -3,5 +3,6 @@ ## TODO: +- [ ] ❗ If this PR includes a new database schema migration, following steps are completed: (README)[README.md] - [ ] Version of pepdbagent updated in `__version__.py` file - [ ] Changelog updated \ No newline at end of file diff --git a/.github/workflows/run-pytest.yml b/.github/workflows/run-pytest.yml index 3abcaa95..d2d6c3ce 100644 --- a/.github/workflows/run-pytest.yml +++ b/.github/workflows/run-pytest.yml @@ -15,6 +15,7 @@ jobs: python-version: ["3.10", "3.13"] os: [ubuntu-latest] # can't use macOS when using service containers or container jobs runs-on: ${{ matrix.os }} + services: postgres: image: postgres diff --git a/README.md b/README.md index 227e40f1..6211ac19 100644 --- a/README.md +++ b/README.md @@ -48,11 +48,62 @@ from bbconf import BedBaseAgent agent = BedBaseAgent(config="config.yaml") # Access submodules -agent.bed # BED file operations -agent.bedset # BED set operations -agent.objects # Generic object/file operations +agent.bed # BED file operations +agent.bedset # BED set operations +agent.objects # Generic object/file operations # Get platform statistics stats = agent.get_stats() print(stats.bedfiles_number, stats.bedsets_number) ``` + +## Database migrations + +`bbconf` uses [Alembic](https://alembic.sqlalchemy.org/) to version the database +schema. The migration scripts live in `bbconf/alembic`, and `alembic.ini` (repo +root) is used for local CLI work. The first (baseline) revision is +`8b0b706d0827`; it reproduces exactly the schema that `Base.metadata.create_all()` +builds, including the `pg_trgm` extension and the trigram / partial / expression +indexes. + +To update schema for desirable database, use different database url in `alembic.ini`, +otherwise run test database + +### Creating a new revision + +After changing the models in `bbconf/db_utils.py`: + +```bash +alembic revision --autogenerate -m "Describe your change" +``` + +Review the generated file. Alembic cannot autogenerate a few constructs used by +bbconf — the `pg_trgm` extension and expression-based indexes may need a manual +`op.execute(...)` — so always check the diff before committing. + +### Applying migrations + +```bash +alembic upgrade head # upgrade to the latest revision +alembic downgrade -1 # roll back one revision +alembic current # show the DB's current revision +``` + +### Running migrations automatically + +To upgrade the database to `head` automatically when `bbconf` starts, set +`run_migrations: true` under the `database` section of the config file: + +```yaml +database: + host: localhost + port: 5432 + user: postgres + password: docker + database: bedbase + run_migrations: true +``` + +> **Note:** enable this only after the database has been stamped/upgraded to a +> known revision. Turning it on against an un-stamped existing database will fail +> on startup, because the baseline revision creates tables that already exist. diff --git a/alembic.ini b/alembic.ini new file mode 100644 index 00000000..5f4c58b6 --- /dev/null +++ b/alembic.ini @@ -0,0 +1,128 @@ +# A generic, single database configuration. +# +# This file is used only for local development / CLI work +# (e.g. `alembic revision --autogenerate`, `alembic upgrade head`). +# At runtime, bbconf builds the Alembic config programmatically in +# `BaseEngine.run_db_migration()` and does NOT read this file. + +[alembic] +# path to migration scripts +# Use forward slashes (/) also on windows to provide an os agnostic path +script_location = ./bbconf/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 + +# sys.path path, will be prepended to sys.path if present. +# defaults to the current working directory. +prepend_sys_path = . + +# timezone to use when rendering the date within the migration file +# as well as the filename. +# If specified, requires the python>=3.9 or backports.zoneinfo library and tzdata library. +# Any required deps can 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 alembic/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 "version_path_separator" below. +# version_locations = %(here)s/bar:%(here)s/bat:alembic/versions + +# version path separator; As mentioned above, this is the character used to split +# version_locations. The default within new alembic.ini files is "os", which uses os.pathsep. +# If this key is omitted entirely, it falls back to the legacy behavior of splitting on spaces and/or commas. +# Valid values for version_path_separator are: +# +# version_path_separator = : +# version_path_separator = ; +# version_path_separator = space +# version_path_separator = newline +# +# Use os.pathsep. Default configuration used for new projects. +version_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 + +# Local development connection string. Override with `-x` or edit as needed. +# Runtime migrations use the URL built from the bbconf config instead. + +### !!!! Change this code to desirable database!!!! +sqlalchemy.url = postgresql+psycopg://postgres:docker@localhost:5432/bedbase + + +[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 exec runner, execute a binary +# hooks = ruff +# ruff.type = exec +# ruff.executable = %(here)s/.venv/bin/ruff +# ruff.options = check --fix REVISION_SCRIPT_FILENAME + +# Logging configuration +[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/bbconf/alembic/README b/bbconf/alembic/README new file mode 100644 index 00000000..2500aa1b --- /dev/null +++ b/bbconf/alembic/README @@ -0,0 +1 @@ +Generic single-database configuration. diff --git a/bbconf/alembic/__init__.py b/bbconf/alembic/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/bbconf/alembic/env.py b/bbconf/alembic/env.py new file mode 100644 index 00000000..e2b9fecb --- /dev/null +++ b/bbconf/alembic/env.py @@ -0,0 +1,77 @@ +from logging.config import fileConfig + +from alembic import context +from sqlalchemy import engine_from_config, pool + +# 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. +# This line sets up loggers basically. +if config.config_file_name is not None: + fileConfig(config.config_file_name) + +# add your model's MetaData object here +# for 'autogenerate' support. +# Importing from bbconf.db_utils also registers the custom @compiles types +# (BIGSERIAL, JSON->JSONB, ARRAY) and the pg_trgm extension DDL event, so the +# metadata compiles to exactly the same DDL that create_all() produces. +from bbconf.db_utils import Base + +target_metadata = Base.metadata + +# other values from the config, defined by the needs of env.py, +# can be acquired: +# my_important_option = config.get_main_option("my_important_option") +# ... etc. + + +def run_migrations_offline() -> None: + """Run migrations in 'offline' mode. + + This configures the context with just a URL + and not an Engine, though an Engine is acceptable + here as well. By skipping the Engine creation + we don't even need a DBAPI to be available. + + Calls to context.execute() here emit the given string to the + script output. + + """ + url = config.get_main_option("sqlalchemy.url") + context.configure( + url=url, + target_metadata=target_metadata, + literal_binds=True, + dialect_opts={"paramstyle": "named"}, + ) + + with context.begin_transaction(): + context.run_migrations() + + +def run_migrations_online() -> None: + """Run migrations in 'online' mode. + + In this scenario we need to create an Engine + and associate a connection with the context. + + """ + connectable = engine_from_config( + config.get_section(config.config_ini_section, {}), + prefix="sqlalchemy.", + poolclass=pool.NullPool, + ) + + with connectable.connect() as connection: + context.configure(connection=connection, target_metadata=target_metadata) + + with context.begin_transaction(): + context.run_migrations() + + +if context.is_offline_mode(): + run_migrations_offline() +else: + run_migrations_online() diff --git a/bbconf/alembic/script.py.mako b/bbconf/alembic/script.py.mako new file mode 100644 index 00000000..51a73aa6 --- /dev/null +++ b/bbconf/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 + +import sqlalchemy as sa +from alembic import op +${imports if imports else ""} + +# revision identifiers, used by Alembic. +revision: str = ${repr(up_revision)} +down_revision: Union[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/bbconf/alembic/versions/845d978eac7d_added_genomic_distribution_json_plots.py b/bbconf/alembic/versions/845d978eac7d_added_genomic_distribution_json_plots.py new file mode 100644 index 00000000..fc486f6f --- /dev/null +++ b/bbconf/alembic/versions/845d978eac7d_added_genomic_distribution_json_plots.py @@ -0,0 +1,60 @@ +"""Added genomic distribution json plots + +Revision ID: 845d978eac7d +Revises: 8b0b706d0827 +Create Date: 2026-08-16 18:02:03.058352 + +""" + +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 = "845d978eac7d" +down_revision: Union[str, None] = "8b0b706d0827" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + """Upgrade schema.""" + + op.drop_column("bed", "pephub") + op.add_column( + "bed_stats", + sa.Column( + "distributions", + postgresql.JSONB(astext_type=sa.Text()), + nullable=True, + comment="Full distribution arrays from gtars genomicdist (JSONB)", + ), + ) + op.add_column( + "bedsets", + sa.Column( + "bedset_stats", + postgresql.JSONB(astext_type=sa.Text()), + nullable=True, + comment="Pre-aggregated distribution statistics from gtars (JSONB)", + ), + ) + + +def downgrade() -> None: + """Downgrade schema.""" + # ### commands auto generated by Alembic - please adjust! ### + op.drop_column("bedsets", "bedset_stats") + op.drop_column("bed_stats", "distributions") + op.add_column( + "bed", + sa.Column( + "pephub", + sa.BOOLEAN(), + autoincrement=False, + nullable=False, + comment="Whether sample was added to pephub", + ), + ) diff --git a/bbconf/alembic/versions/8b0b706d0827_initial_migration.py b/bbconf/alembic/versions/8b0b706d0827_initial_migration.py new file mode 100644 index 00000000..24f76169 --- /dev/null +++ b/bbconf/alembic/versions/8b0b706d0827_initial_migration.py @@ -0,0 +1,742 @@ +"""Initial migration + +Revision ID: 8b0b706d0827 +Revises: +Create Date: 2026-08-14 23:09:22.903899 + +""" + +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 = "8b0b706d0827" +down_revision: Union[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! ### + # pg_trgm backs the trigram GIN indexes on `bedsets` (ix_bedsets_name_trgm / + # ix_bedsets_description_trgm). In the ORM this is created by a before_create + # DDL event on the bedsets table; autogenerate does not emit it, so it is + # added here by hand. Must run before those indexes are created. + op.execute("CREATE EXTENSION IF NOT EXISTS pg_trgm") + op.create_table( + "bed_snapshots", + sa.Column("id", sa.Integer(), autoincrement=True, nullable=False), + sa.Column( + "file_path", + sa.String(), + nullable=False, + comment="S3 object key, relative to the bucket root", + ), + sa.Column( + "file_type", + sa.String(), + nullable=False, + comment="metadata | bedsets | bedset_membership | manifest", + ), + sa.Column( + "creation_date", + sa.TIMESTAMP(timezone=True), + nullable=False, + comment="Build date of the export", + ), + sa.Column( + "record_count", + sa.Integer(), + nullable=True, + comment="Rows actually written to the file", + ), + 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( + "schema_version", + sa.Integer(), + nullable=True, + comment="Export schema version", + ), + sa.PrimaryKeyConstraint("id"), + ) + op.create_index(op.f("ix_bed_snapshots_id"), "bed_snapshots", ["id"], unique=False) + op.create_table( + "bedsets", + sa.Column("id", sa.String(), nullable=False), + sa.Column("name", sa.String(), nullable=False, comment="Name of the bedset"), + sa.Column( + "description", + sa.String(), + nullable=True, + comment="Description of the bedset", + ), + sa.Column( + "summary", sa.String(), nullable=True, comment="Summary of the bedset" + ), + sa.Column("submission_date", sa.TIMESTAMP(timezone=True), nullable=False), + sa.Column("last_update_date", sa.TIMESTAMP(timezone=True), nullable=True), + sa.Column( + "md5sum", sa.String(), nullable=True, comment="MD5 sum of the bedset" + ), + sa.Column( + "bedset_means", + postgresql.JSON(astext_type=sa.Text()), + nullable=True, + comment="Mean values of the bedset", + ), + sa.Column( + "bedset_standard_deviation", + postgresql.JSON(astext_type=sa.Text()), + nullable=True, + comment="Median values of the bedset", + ), + sa.Column( + "bedfile_count", + sa.Integer(), + nullable=False, + comment="Number of bedfiles in the bedset (denormalized count)", + ), + sa.Column("author", sa.String(), nullable=True, comment="Author of the bedset"), + sa.Column("source", sa.String(), nullable=True, comment="Source of the bedset"), + sa.Column( + "processed", + sa.Boolean(), + nullable=False, + comment="Whether the bedset was processed", + ), + sa.PrimaryKeyConstraint("id"), + ) + op.create_index( + "ix_bedsets_description_trgm", + "bedsets", + ["description"], + unique=False, + postgresql_using="gin", + postgresql_ops={"description": "gin_trgm_ops"}, + ) + op.create_index(op.f("ix_bedsets_id"), "bedsets", ["id"], unique=False) + op.create_index( + "ix_bedsets_name_trgm", + "bedsets", + ["name"], + unique=False, + postgresql_using="gin", + postgresql_ops={"name": "gin_trgm_ops"}, + ) + op.create_index( + "ix_bedsets_unprocessed", + "bedsets", + ["id"], + unique=False, + postgresql_where=sa.text("processed = false"), + ) + op.create_table( + "geo_gse_status", + sa.Column("id", sa.Integer(), autoincrement=True, nullable=False), + sa.Column("gse", sa.String(), nullable=False, comment="GSE number"), + sa.Column( + "status", sa.String(), nullable=False, comment="Status of the GEO project" + ), + sa.Column("submission_date", sa.TIMESTAMP(timezone=True), nullable=False), + sa.Column( + "number_of_files", sa.Integer(), nullable=False, comment="Number of files" + ), + sa.Column( + "number_of_success", + sa.Integer(), + nullable=False, + comment="Number of success", + ), + sa.Column( + "number_of_skips", sa.Integer(), nullable=False, comment="Number of skips" + ), + sa.Column( + "number_of_fails", sa.Integer(), nullable=False, comment="Number of fails" + ), + sa.Column("error", sa.String(), nullable=True, comment="Error message"), + sa.PrimaryKeyConstraint("id"), + sa.UniqueConstraint("gse"), + ) + op.create_index( + op.f("ix_geo_gse_status_id"), "geo_gse_status", ["id"], unique=False + ) + op.create_table( + "licenses", + sa.Column("id", sa.String(), nullable=False), + sa.Column("shorthand", sa.String(), nullable=True, comment="License shorthand"), + sa.Column("label", sa.String(), nullable=False, comment="License label"), + sa.Column( + "description", sa.String(), nullable=False, comment="License description" + ), + sa.PrimaryKeyConstraint("id"), + ) + op.create_index(op.f("ix_licenses_id"), "licenses", ["id"], unique=False) + op.create_table( + "reference_genomes", + sa.Column("digest", sa.String(), nullable=False), + sa.Column( + "alias", sa.String(), nullable=False, comment="Name of the reference genome" + ), + sa.PrimaryKeyConstraint("digest"), + ) + op.create_index( + op.f("ix_reference_genomes_digest"), + "reference_genomes", + ["digest"], + unique=False, + ) + op.create_table( + "usage_files", + sa.Column("id", sa.Integer(), autoincrement=True, nullable=False), + sa.Column("file_path", sa.String(), nullable=False, comment="Path to the file"), + sa.Column("count", sa.Integer(), nullable=False, comment="Number of downloads"), + sa.Column( + "date_from", + sa.TIMESTAMP(timezone=True), + nullable=False, + comment="Date from", + ), + sa.Column( + "date_to", sa.TIMESTAMP(timezone=True), nullable=False, comment="Date to" + ), + sa.PrimaryKeyConstraint("id"), + ) + op.create_index(op.f("ix_usage_files_id"), "usage_files", ["id"], unique=False) + op.create_table( + "usage_search", + sa.Column("id", sa.Integer(), autoincrement=True, nullable=False), + sa.Column("query", sa.String(), nullable=False, comment="Search query"), + sa.Column( + "type", + sa.String(), + nullable=False, + comment="Type of the search. Bed/Bedset", + ), + sa.Column("count", sa.Integer(), nullable=False, comment="Number of searches"), + sa.Column( + "date_from", + sa.TIMESTAMP(timezone=True), + nullable=False, + comment="Date from", + ), + sa.Column( + "date_to", sa.TIMESTAMP(timezone=True), nullable=False, comment="Date to" + ), + sa.PrimaryKeyConstraint("id"), + ) + op.create_index(op.f("ix_usage_search_id"), "usage_search", ["id"], unique=False) + op.create_table( + "bed", + sa.Column("id", sa.String(), nullable=False), + sa.Column("name", sa.String(), nullable=True), + sa.Column("genome_alias", sa.String(), nullable=True), + sa.Column("genome_digest", sa.String(), nullable=True), + sa.Column("description", sa.String(), nullable=True), + sa.Column("bed_compliance", sa.String(), nullable=False), + sa.Column("data_format", sa.String(), nullable=False), + sa.Column("compliant_columns", sa.Integer(), nullable=False), + sa.Column("non_compliant_columns", sa.Integer(), nullable=False), + sa.Column( + "header", + sa.String(), + nullable=True, + comment="Header of the bed file, it if was provided.", + ), + sa.Column( + "indexed", + sa.Boolean(), + nullable=False, + comment="Whether sample was added to qdrant", + ), + sa.Column( + "file_indexed", + sa.Boolean(), + nullable=False, + comment="Whether file was tokenized and added to the vector database", + ), + sa.Column( + "pephub", + sa.Boolean(), + nullable=False, + comment="Whether sample was added to pephub", + ), + sa.Column("submission_date", sa.TIMESTAMP(timezone=True), nullable=False), + sa.Column("last_update_date", sa.TIMESTAMP(timezone=True), nullable=True), + sa.Column("is_universe", sa.Boolean(), nullable=True), + sa.Column("license_id", sa.String(), nullable=True), + sa.Column( + "processed", + sa.Boolean(), + nullable=False, + comment="Whether the bed file was processed", + ), + sa.ForeignKeyConstraint(["license_id"], ["licenses.id"], ondelete="CASCADE"), + sa.PrimaryKeyConstraint("id"), + ) + op.create_index( + "genome_alias_index", + "bed", + ["genome_alias"], + unique=False, + postgresql_with={"deduplicate_items": "true"}, + ) + op.create_index(op.f("ix_bed_id"), "bed", ["id"], unique=False) + op.create_index(op.f("ix_bed_license_id"), "bed", ["license_id"], unique=False) + op.create_index( + "ix_bed_not_file_indexed", + "bed", + ["id"], + unique=False, + postgresql_where=sa.text("file_indexed = false"), + ) + op.create_index( + "ix_bed_not_indexed", + "bed", + ["id"], + unique=False, + postgresql_where=sa.text("indexed = false"), + ) + op.create_index( + "ix_bed_submission_date", + "bed", + [sa.literal_column("submission_date DESC"), sa.literal_column("id")], + unique=False, + ) + op.create_index( + "ix_bed_unprocessed", + "bed", + ["id"], + unique=False, + postgresql_where=sa.text("processed = false"), + ) + op.create_table( + "geo_gsm_status", + sa.Column("id", sa.Integer(), autoincrement=True, nullable=False), + sa.Column("gse_status_id", sa.Integer(), nullable=False), + sa.Column("gsm", sa.String(), nullable=False, comment="GSM number"), + sa.Column("sample_name", sa.String(), nullable=False), + sa.Column( + "status", sa.String(), nullable=False, comment="Status of the GEO sample" + ), + sa.Column("error", sa.String(), nullable=True, comment="Error message"), + sa.Column( + "source_submission_date", + sa.TIMESTAMP(timezone=True), + nullable=True, + comment="Submission date of the source", + ), + sa.Column("submission_date", sa.TIMESTAMP(timezone=True), nullable=False), + sa.Column("bed_id", sa.String(), nullable=True, comment="Bed identifier"), + sa.Column( + "file_size", sa.BigInteger(), nullable=False, comment="Size of the file" + ), + sa.Column("genome", sa.String(), nullable=True, comment="Genome"), + sa.ForeignKeyConstraint( + ["gse_status_id"], ["geo_gse_status.id"], ondelete="CASCADE" + ), + sa.PrimaryKeyConstraint("id"), + ) + op.create_index( + op.f("ix_geo_gsm_status_bed_id"), "geo_gsm_status", ["bed_id"], unique=False + ) + op.create_index( + op.f("ix_geo_gsm_status_gse_status_id"), + "geo_gsm_status", + ["gse_status_id"], + unique=False, + ) + op.create_index( + op.f("ix_geo_gsm_status_id"), "geo_gsm_status", ["id"], unique=False + ) + op.create_table( + "usage_bedset_meta", + sa.Column("id", sa.Integer(), autoincrement=True, nullable=False), + sa.Column("bedset_id", sa.String(), nullable=True), + sa.Column("count", sa.Integer(), nullable=False, comment="Number of visits"), + sa.Column( + "date_from", + sa.TIMESTAMP(timezone=True), + nullable=False, + comment="Date from", + ), + sa.Column( + "date_to", sa.TIMESTAMP(timezone=True), nullable=False, comment="Date to" + ), + sa.ForeignKeyConstraint(["bedset_id"], ["bedsets.id"], ondelete="CASCADE"), + sa.PrimaryKeyConstraint("id"), + ) + op.create_index( + op.f("ix_usage_bedset_meta_bedset_id"), + "usage_bedset_meta", + ["bedset_id"], + unique=False, + ) + op.create_index( + op.f("ix_usage_bedset_meta_id"), "usage_bedset_meta", ["id"], unique=False + ) + op.create_table( + "bed_metadata", + sa.Column("species_name", sa.String(), nullable=False, comment="Organism name"), + sa.Column( + "species_id", sa.String(), nullable=True, comment="Organism taxon id" + ), + sa.Column( + "genotype", sa.String(), nullable=True, comment="Genotype of the sample" + ), + sa.Column( + "phenotype", sa.String(), nullable=True, comment="Phenotype of the sample" + ), + sa.Column( + "cell_type", + sa.String(), + nullable=True, + comment="Specific kind of cell with distinct characteristics found in an organism. e.g. Neurons, Hepatocytes, Adipocytes", + ), + sa.Column( + "cell_line", + sa.String(), + nullable=True, + comment="Population of cells derived from a single cell and cultured in the lab for extended use, e.g. HeLa, HepG2, k562", + ), + sa.Column("tissue", sa.String(), nullable=True, comment="Tissue type"), + sa.Column( + "library_source", + sa.String(), + nullable=True, + comment="Library source (e.g. genomic, transcriptomic)", + ), + sa.Column( + "assay", + sa.String(), + nullable=True, + comment="Experimental protocol (e.g. ChIP-seq)", + ), + sa.Column( + "antibody", sa.String(), nullable=True, comment="Antibody used in the assay" + ), + sa.Column( + "target", + sa.String(), + nullable=True, + comment="Target of the assay (e.g. H3K4me3)", + ), + sa.Column( + "treatment", + sa.String(), + nullable=True, + comment="Treatment of the sample (e.g. drug treatment)", + ), + sa.Column( + "original_file_name", + sa.String(), + nullable=True, + comment="Original file name", + ), + sa.Column( + "global_sample_id", + postgresql.ARRAY(sa.String()), + nullable=True, + comment="Global sample identifier. e.g. GSM000", + ), + sa.Column( + "global_experiment_id", + postgresql.ARRAY(sa.String()), + nullable=True, + comment="Global experiment identifier. e.g. GSE000", + ), + sa.Column("id", sa.String(), nullable=False), + sa.ForeignKeyConstraint(["id"], ["bed.id"], ondelete="CASCADE"), + sa.PrimaryKeyConstraint("id"), + ) + op.create_index(op.f("ix_bed_metadata_id"), "bed_metadata", ["id"], unique=False) + op.create_table( + "bed_stats", + sa.Column("id", sa.String(), nullable=False), + sa.Column("number_of_regions", sa.Float(), nullable=True), + sa.Column("gc_content", sa.Float(), nullable=True), + sa.Column("median_tss_dist", sa.Float(), nullable=True), + sa.Column("mean_region_width", sa.Float(), nullable=True), + sa.Column("exon_frequency", sa.Float(), nullable=True), + sa.Column("intron_frequency", sa.Float(), nullable=True), + sa.Column("promoterprox_frequency", sa.Float(), nullable=True), + sa.Column("intergenic_frequency", sa.Float(), nullable=True), + sa.Column("promotercore_frequency", sa.Float(), nullable=True), + sa.Column("fiveutr_frequency", sa.Float(), nullable=True), + sa.Column("threeutr_frequency", sa.Float(), nullable=True), + sa.Column("fiveutr_percentage", sa.Float(), nullable=True), + sa.Column("threeutr_percentage", sa.Float(), nullable=True), + sa.Column("promoterprox_percentage", sa.Float(), nullable=True), + sa.Column("exon_percentage", sa.Float(), nullable=True), + sa.Column("intron_percentage", sa.Float(), nullable=True), + sa.Column("intergenic_percentage", sa.Float(), nullable=True), + sa.Column("promotercore_percentage", sa.Float(), nullable=True), + sa.Column("tssdist", sa.Float(), nullable=True), + sa.ForeignKeyConstraint(["id"], ["bed.id"], ondelete="CASCADE"), + sa.PrimaryKeyConstraint("id"), + ) + op.create_index(op.f("ix_bed_stats_id"), "bed_stats", ["id"], unique=False) + op.create_index( + "ix_bed_stats_missing_regions", + "bed_stats", + ["id"], + unique=False, + postgresql_where=sa.text("number_of_regions IS NULL"), + ) + op.create_table( + "bedfile_bedset_relation", + sa.Column("bedset_id", sa.String(), nullable=False), + sa.Column("bedfile_id", sa.String(), nullable=False), + sa.ForeignKeyConstraint(["bedfile_id"], ["bed.id"], ondelete="CASCADE"), + sa.ForeignKeyConstraint(["bedset_id"], ["bedsets.id"], ondelete="CASCADE"), + sa.PrimaryKeyConstraint("bedset_id", "bedfile_id"), + ) + op.create_index( + op.f("ix_bedfile_bedset_relation_bedfile_id"), + "bedfile_bedset_relation", + ["bedfile_id"], + unique=False, + ) + op.create_table( + "files", + sa.Column("id", sa.Integer(), nullable=False), + sa.Column( + "name", + sa.String(), + nullable=False, + comment="Name of the file, e.g. bed, bigBed", + ), + sa.Column( + "file_digest", + sa.String(), + nullable=True, + comment="Digest of the file. Mainly used for bed file.", + ), + sa.Column("title", sa.String(), nullable=True), + sa.Column( + "type", + sa.String(), + nullable=False, + comment="Type of the object, e.g. file, plot, ...", + ), + sa.Column("path", sa.String(), nullable=False), + sa.Column( + "path_thumbnail", + sa.String(), + nullable=True, + comment="Thumbnail path of the file", + ), + sa.Column("description", sa.String(), nullable=True), + sa.Column("size", sa.Integer(), nullable=True, comment="Size of the file"), + sa.Column("bedfile_id", sa.String(), nullable=True), + sa.Column("bedset_id", sa.String(), nullable=True), + sa.ForeignKeyConstraint(["bedfile_id"], ["bed.id"], ondelete="CASCADE"), + sa.ForeignKeyConstraint(["bedset_id"], ["bedsets.id"], ondelete="CASCADE"), + sa.PrimaryKeyConstraint("id"), + sa.UniqueConstraint("name", "bedfile_id"), + sa.UniqueConstraint("name", "bedset_id"), + ) + op.create_index(op.f("ix_files_bedfile_id"), "files", ["bedfile_id"], unique=False) + op.create_index(op.f("ix_files_bedset_id"), "files", ["bedset_id"], unique=False) + op.create_index(op.f("ix_files_id"), "files", ["id"], unique=False) + op.create_table( + "genome_ref_stats", + sa.Column("id", sa.Integer(), nullable=False), + sa.Column("bed_id", sa.String(), nullable=False), + sa.Column("provided_genome", sa.String(), nullable=False), + sa.Column( + "compared_genome", sa.String(), nullable=False, comment="Compared Genome" + ), + sa.Column("genome_digest", sa.String(), nullable=False), + sa.Column("xs", sa.Float(), nullable=True), + sa.Column("oobr", sa.Float(), nullable=True), + sa.Column("sequence_fit", sa.Float(), nullable=True), + sa.Column("assigned_points", sa.Integer(), nullable=False), + sa.Column("tier_ranking", sa.Integer(), nullable=False), + sa.ForeignKeyConstraint(["bed_id"], ["bed.id"], ondelete="CASCADE"), + sa.ForeignKeyConstraint( + ["genome_digest"], ["reference_genomes.digest"], ondelete="CASCADE" + ), + sa.PrimaryKeyConstraint("id"), + sa.UniqueConstraint("bed_id", "compared_genome"), + ) + op.create_index( + op.f("ix_genome_ref_stats_bed_id"), "genome_ref_stats", ["bed_id"], unique=False + ) + op.create_index( + op.f("ix_genome_ref_stats_id"), "genome_ref_stats", ["id"], unique=False + ) + op.create_table( + "universes", + sa.Column("id", sa.String(), nullable=False), + sa.Column( + "method", + sa.String(), + nullable=True, + comment="Method used to create the universe", + ), + sa.Column("bedset_id", sa.String(), nullable=True), + sa.ForeignKeyConstraint(["bedset_id"], ["bedsets.id"], ondelete="CASCADE"), + sa.ForeignKeyConstraint(["id"], ["bed.id"], ondelete="CASCADE"), + sa.PrimaryKeyConstraint("id"), + ) + op.create_index( + op.f("ix_universes_bedset_id"), "universes", ["bedset_id"], unique=False + ) + op.create_index(op.f("ix_universes_id"), "universes", ["id"], unique=False) + op.create_table( + "usage_bed_meta", + sa.Column("id", sa.Integer(), autoincrement=True, nullable=False), + sa.Column("bed_id", sa.String(), nullable=True), + sa.Column("count", sa.Integer(), nullable=False, comment="Number of visits"), + sa.Column( + "date_from", + sa.TIMESTAMP(timezone=True), + nullable=False, + comment="Date from", + ), + sa.Column( + "date_to", sa.TIMESTAMP(timezone=True), nullable=False, comment="Date to" + ), + sa.ForeignKeyConstraint(["bed_id"], ["bed.id"], ondelete="CASCADE"), + sa.PrimaryKeyConstraint("id"), + ) + op.create_index( + op.f("ix_usage_bed_meta_bed_id"), "usage_bed_meta", ["bed_id"], unique=False + ) + op.create_index( + op.f("ix_usage_bed_meta_id"), "usage_bed_meta", ["id"], unique=False + ) + op.create_table( + "tokenized_bed", + sa.Column("bed_id", sa.String(), nullable=False), + sa.Column("universe_id", sa.String(), nullable=False), + sa.Column( + "path", + sa.String(), + nullable=False, + comment="Path to the tokenized bed file", + ), + sa.ForeignKeyConstraint(["bed_id"], ["bed.id"], ondelete="CASCADE"), + sa.ForeignKeyConstraint(["universe_id"], ["universes.id"], ondelete="CASCADE"), + sa.PrimaryKeyConstraint("bed_id", "universe_id"), + ) + op.create_index( + op.f("ix_tokenized_bed_bed_id"), "tokenized_bed", ["bed_id"], unique=False + ) + op.create_index( + op.f("ix_tokenized_bed_universe_id"), + "tokenized_bed", + ["universe_id"], + unique=False, + ) + # ### end Alembic commands ### + + +def downgrade() -> None: + """Downgrade schema.""" + # ### commands auto generated by Alembic - please adjust! ### + op.drop_index(op.f("ix_tokenized_bed_universe_id"), table_name="tokenized_bed") + op.drop_index(op.f("ix_tokenized_bed_bed_id"), table_name="tokenized_bed") + op.drop_table("tokenized_bed") + op.drop_index(op.f("ix_usage_bed_meta_id"), table_name="usage_bed_meta") + op.drop_index(op.f("ix_usage_bed_meta_bed_id"), table_name="usage_bed_meta") + op.drop_table("usage_bed_meta") + op.drop_index(op.f("ix_universes_id"), table_name="universes") + op.drop_index(op.f("ix_universes_bedset_id"), table_name="universes") + op.drop_table("universes") + op.drop_index(op.f("ix_genome_ref_stats_id"), table_name="genome_ref_stats") + op.drop_index(op.f("ix_genome_ref_stats_bed_id"), table_name="genome_ref_stats") + op.drop_table("genome_ref_stats") + op.drop_index(op.f("ix_files_id"), table_name="files") + op.drop_index(op.f("ix_files_bedset_id"), table_name="files") + op.drop_index(op.f("ix_files_bedfile_id"), table_name="files") + op.drop_table("files") + op.drop_index( + op.f("ix_bedfile_bedset_relation_bedfile_id"), + table_name="bedfile_bedset_relation", + ) + op.drop_table("bedfile_bedset_relation") + op.drop_index( + "ix_bed_stats_missing_regions", + table_name="bed_stats", + postgresql_where=sa.text("number_of_regions IS NULL"), + ) + op.drop_index(op.f("ix_bed_stats_id"), table_name="bed_stats") + op.drop_table("bed_stats") + op.drop_index(op.f("ix_bed_metadata_id"), table_name="bed_metadata") + op.drop_table("bed_metadata") + op.drop_index(op.f("ix_usage_bedset_meta_id"), table_name="usage_bedset_meta") + op.drop_index( + op.f("ix_usage_bedset_meta_bedset_id"), table_name="usage_bedset_meta" + ) + op.drop_table("usage_bedset_meta") + op.drop_index(op.f("ix_geo_gsm_status_id"), table_name="geo_gsm_status") + op.drop_index(op.f("ix_geo_gsm_status_gse_status_id"), table_name="geo_gsm_status") + op.drop_index(op.f("ix_geo_gsm_status_bed_id"), table_name="geo_gsm_status") + op.drop_table("geo_gsm_status") + op.drop_index( + "ix_bed_unprocessed", + table_name="bed", + postgresql_where=sa.text("processed = false"), + ) + op.drop_index("ix_bed_submission_date", table_name="bed") + op.drop_index( + "ix_bed_not_indexed", + table_name="bed", + postgresql_where=sa.text("indexed = false"), + ) + op.drop_index( + "ix_bed_not_file_indexed", + table_name="bed", + postgresql_where=sa.text("file_indexed = false"), + ) + op.drop_index(op.f("ix_bed_license_id"), table_name="bed") + op.drop_index(op.f("ix_bed_id"), table_name="bed") + op.drop_index( + "genome_alias_index", + table_name="bed", + postgresql_with={"deduplicate_items": "true"}, + ) + op.drop_table("bed") + op.drop_index(op.f("ix_usage_search_id"), table_name="usage_search") + op.drop_table("usage_search") + op.drop_index(op.f("ix_usage_files_id"), table_name="usage_files") + op.drop_table("usage_files") + op.drop_index(op.f("ix_reference_genomes_digest"), table_name="reference_genomes") + op.drop_table("reference_genomes") + op.drop_index(op.f("ix_licenses_id"), table_name="licenses") + op.drop_table("licenses") + op.drop_index(op.f("ix_geo_gse_status_id"), table_name="geo_gse_status") + op.drop_table("geo_gse_status") + op.drop_index( + "ix_bedsets_unprocessed", + table_name="bedsets", + postgresql_where=sa.text("processed = false"), + ) + op.drop_index( + "ix_bedsets_name_trgm", + table_name="bedsets", + postgresql_using="gin", + postgresql_ops={"name": "gin_trgm_ops"}, + ) + op.drop_index(op.f("ix_bedsets_id"), table_name="bedsets") + op.drop_index( + "ix_bedsets_description_trgm", + table_name="bedsets", + postgresql_using="gin", + postgresql_ops={"description": "gin_trgm_ops"}, + ) + op.drop_table("bedsets") + op.drop_index(op.f("ix_bed_snapshots_id"), table_name="bed_snapshots") + op.drop_table("bed_snapshots") + # ### end Alembic commands ### diff --git a/bbconf/alembic/versions/__init__.py b/bbconf/alembic/versions/__init__.py new file mode 100644 index 00000000..e69de29b 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 00000000..b11229e1 --- /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 6248f6d8..e4c94ac8 100644 --- a/bbconf/bbagent.py +++ b/bbconf/bbagent.py @@ -1,9 +1,11 @@ import logging import statistics +import threading from functools import cached_property from pathlib import Path import numpy as np +from cachetools import TTLCache from sqlalchemy.engine import ScalarResult from sqlalchemy.orm import Session from sqlalchemy.sql import and_, distinct, func, or_, select @@ -33,9 +35,11 @@ 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 +from bbconf.modules.snapshots import BedAgentSnapshot from .const import PKG_NAME @@ -62,6 +66,16 @@ def __init__( self._bed = BedAgentBedFile(self.config, self) 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 + # plus the neighbours/list/search result builders). Cache the result + # with a TTL so those paths do not hit the database on every request. + # The lock guards the cache dict only, never the DB query itself. + self._stats_cache = TTLCache(maxsize=1, ttl=3600) + self._stats_lock = threading.Lock() @property def bed(self) -> BedAgentBedFile: @@ -75,6 +89,14 @@ def bedset(self) -> BedAgentBedSet: def objects(self) -> BBObjects: return self._objects + @property + 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}" @@ -86,9 +108,17 @@ def get_stats(self) -> StatsReturn: """ Get statistics for a bed file. + The result is cached with a TTL because this runs three COUNT queries + against the large bed table and is called on hot API paths. + Returns: Statistics. """ + with self._stats_lock: + cached = self._stats_cache.get("stats") + if cached is not None: + return cached + with Session(self.config.db_engine.engine) as session: number_of_bed = session.execute(select(func.count(Bed.id))).one()[0] number_of_bedset = session.execute(select(func.count(BedSets.id))).one()[0] @@ -97,12 +127,17 @@ def get_stats(self) -> StatsReturn: select(func.count(distinct(Bed.genome_alias))) ).one()[0] - return StatsReturn( + stats = StatsReturn( bedfiles_number=number_of_bed, bedsets_number=number_of_bedset, genomes_number=number_of_genomes, ) + with self._stats_lock: + self._stats_cache["stats"] = stats + + return stats + def get_detailed_stats(self, concise: bool = False) -> FileStats: """ Get comprehensive statistics for all bed files. @@ -116,11 +151,29 @@ def get_detailed_stats(self, concise: bool = False) -> FileStats: _LOGGER.info("Getting detailed statistics for all bed files") + numeric_stats_statement = ( + select( + BedStats.number_of_regions, + BedStats.mean_region_width, + Files.size, + ) + .select_from(Bed) + .join(BedStats, BedStats.id == Bed.id) + .join(Files, Files.bedfile_id == Bed.id) + .where( + Files.name == "bed_file", + BedStats.number_of_regions.is_not(None), + BedStats.mean_region_width.is_not(None), + Files.size.is_not(None), + ) + ) + with Session(self.config.db_engine.engine) as session: bed_compliance = { f[0]: f[1] for f in session.execute( select(Bed.bed_compliance, func.count(Bed.bed_compliance)) + .where(Bed.bed_compliance.is_not(None)) .group_by(Bed.bed_compliance) .order_by(func.count(Bed.bed_compliance).desc()) ).all() @@ -129,6 +182,7 @@ def get_detailed_stats(self, concise: bool = False) -> FileStats: f[0]: f[1] for f in session.execute( select(Bed.data_format, func.count(Bed.data_format)) + .where(Bed.data_format.is_not(None)) .group_by(Bed.data_format) .order_by(func.count(Bed.data_format).desc()) ).all() @@ -137,6 +191,7 @@ def get_detailed_stats(self, concise: bool = False) -> FileStats: f[0]: f[1] for f in session.execute( select(Bed.genome_alias, func.count(Bed.genome_alias)) + .where(Bed.genome_alias.is_not(None)) .group_by(Bed.genome_alias) .order_by(func.count(Bed.genome_alias).desc()) ).all() @@ -147,6 +202,7 @@ def get_detailed_stats(self, concise: bool = False) -> FileStats: select( BedMetadata.species_name, func.count(BedMetadata.species_name) ) + .where(BedMetadata.species_name.is_not(None)) .group_by(BedMetadata.species_name) .order_by(func.count(BedMetadata.species_name).desc()) ).all() @@ -155,6 +211,7 @@ def get_detailed_stats(self, concise: bool = False) -> FileStats: f[0]: f[1] for f in session.execute( select(BedMetadata.assay, func.count(BedMetadata.assay)) + .where(BedMetadata.assay.is_not(None)) .group_by(BedMetadata.assay) .order_by(func.count(BedMetadata.assay).desc()) ).all() @@ -163,28 +220,29 @@ def get_detailed_stats(self, concise: bool = False) -> FileStats: f[0]: f[1] for f in session.execute( select(BedMetadata.cell_line, func.count(BedMetadata.cell_line)) + .where(BedMetadata.cell_line.is_not(None)) .group_by(BedMetadata.cell_line) .order_by(func.count(BedMetadata.cell_line).desc()) ).all() } - slice_value = 20 + bed_comments = self._stats_comments(session) + geo_status = self._stats_geo_status(session) + + numeric_rows = session.execute(numeric_stats_statement).all() - bed_comments = self._stats_comments(session) - geo_status = self._stats_geo_status(session) + geo_stats = self._get_geo_stats(session) - bedfiles_info = self.bed_files_info() + slice_value = 20 - number_of_regions = [bed.number_of_regions for bed in bedfiles_info.files] - list_mean_width = [bed.mean_region_width for bed in bedfiles_info.files] - list_file_size = [bed.file_size for bed in bedfiles_info.files] + number_of_regions = [row[0] for row in numeric_rows] + list_mean_width = [row[1] for row in numeric_rows] + list_file_size = [row[2] for row in numeric_rows] number_of_regions_bins = self._bin_number_of_regions(number_of_regions) list_mean_width_bins = self._bin_mean_region_width(list_mean_width) list_file_size_bins = self._bin_file_size(list_file_size) - geo_stats = self._get_geo_stats(session) - if concise: bed_compliance_concise = dict(list(bed_compliance.items())[0:slice_value]) bed_compliance_concise["other"] = sum( @@ -688,8 +746,14 @@ def _bin_number_of_regions(self, number_of_regions: list) -> BinValues: return BinValues( bins=n_region_bin_edges, counts=n_region_counts, - mean=round(statistics.mean(number_of_regions), 2), - median=round(statistics.median(number_of_regions), 2), + mean=round(statistics.mean(number_of_regions), 2) + if number_of_regions + else 0, + median=( + round(statistics.median(number_of_regions), 2) + if number_of_regions + else 0 + ), ) def _bin_mean_region_width(self, mean_region_widths: list) -> BinValues: @@ -719,8 +783,16 @@ def _bin_mean_region_width(self, mean_region_widths: list) -> BinValues: return BinValues( bins=mean_reg_width_bin_edges, counts=mean_reg_width_counts, - mean=round(statistics.mean(mean_region_widths), 2), - median=round(statistics.median(mean_region_widths), 2), + mean=( + round(statistics.mean(mean_region_widths), 2) + if mean_region_widths + else 0 + ), + median=( + round(statistics.median(mean_region_widths), 2) + if mean_region_widths + else 0 + ), ) def _bin_file_size(self, list_file_size: list) -> BinValues: @@ -751,8 +823,16 @@ def _bin_file_size(self, list_file_size: list) -> BinValues: return BinValues( bins=file_size_bin_edges, counts=file_size_counts, - mean=round(statistics.mean(filtered_list_file_size), 2), - median=round(statistics.median(filtered_list_file_size), 2), + mean=( + round(statistics.mean(filtered_list_file_size), 2) + if filtered_list_file_size + else 0 + ), + median=( + round(statistics.median(filtered_list_file_size), 2) + if filtered_list_file_size + else 0 + ), ) def _get_geo_stats(self, sa_session: Session) -> GEOStatistics: @@ -805,8 +885,8 @@ def _get_geo_stats(self, sa_session: Session) -> GEOStatistics: file_sizes=BinValues( bins=list(file_size_bin_edges), counts=file_size_counts.astype(int).tolist(), - mean=round(statistics.mean(file_sizes), 2), - median=round(statistics.median(file_sizes), 2), + mean=round(statistics.mean(file_sizes), 2) if file_sizes else 0, + median=round(statistics.median(file_sizes), 2) if file_sizes else 0, ), ) diff --git a/bbconf/config_parser/bedbaseconfig.py b/bbconf/config_parser/bedbaseconfig.py index ef9db9e6..f66880f2 100644 --- a/bbconf/config_parser/bedbaseconfig.py +++ b/bbconf/config_parser/bedbaseconfig.py @@ -20,7 +20,6 @@ from geniml.search.backends import BiVectorBackend, QdrantBackend from geniml.search.interfaces import BiVectorSearchInterface from geniml.search.query2vec import BED2Vec -from pephubclient import PEPHubClient from qdrant_client import QdrantClient, models from sentence_transformers import SparseEncoder from umap import UMAP @@ -115,7 +114,6 @@ def __init__(self, config: Path | str, init_ml: bool = True): self.umap_encoder: UMAP | None = None self.sparse_encoder = None - self._phc = self._init_pephubclient() self._boto3_client = self._init_boto3_client() @staticmethod @@ -131,19 +129,7 @@ def _read_config_file(config_path: str) -> ConfigFile: """ _config = yacman.YAMLConfigManager.from_yaml_file(filepath=config_path).exp - - config_dict = {} - for field_name, annotation in ConfigFile.model_fields.items(): - try: - config_dict[field_name] = annotation.annotation( - **_config.get(field_name) - ) - except TypeError: - # TODO: this should be more specific - config_dict[field_name] = annotation.annotation() - - return ConfigFile(**config_dict) - # return ConfigFile.from_yaml(Path(config_path)) + return ConfigFile(**_config) @property def config(self) -> ConfigFile: @@ -165,16 +151,6 @@ def db_engine(self) -> BaseEngine: """ return self._db_engine - @property - def phc(self) -> PEPHubClient: - """ - Get PEPHub client. - - Returns: - PEPHub client. - """ - return self._phc - @property def boto3_client(self) -> boto3.client: """ @@ -227,6 +203,7 @@ def _init_db_engine(self) -> BaseEngine: user=self._config.database.user, password=self._config.database.password, drivername=f"{self._config.database.dialect}+{self._config.database.driver}", + run_migrations=self._config.database.run_migrations, ) def _init_qdrant_client(self) -> QdrantClient: @@ -682,24 +659,6 @@ def delete_files_s3(self, files: list[FileModel]) -> None: self.delete_s3(file.path_thumbnail) return None - @staticmethod - def _init_pephubclient() -> PEPHubClient | None: - """ - Create Pephub client object using credentials provided in config file. - - Returns: - PephubClient. - """ - - # try: - # _LOGGER.info("Initializing PEPHub client...") - # return PEPHubClient() - # except Exception as e: - # _LOGGER.error(f"Error in creating PephubClient object: {e}") - # warnings.warn(f"Error in creating PephubClient object: {e}", UserWarning) - # return None - return None - def get_prefixed_uri(self, postfix: str, access_id: str) -> str: """ Return uri with correct prefix (schema). diff --git a/bbconf/config_parser/const.py b/bbconf/config_parser/const.py index 61aad4ec..91877ef2 100644 --- a/bbconf/config_parser/const.py +++ b/bbconf/config_parser/const.py @@ -17,10 +17,6 @@ DEFAULT_SPARSE_MODEL = "prithivida/Splade_PP_en_v2" DEFAULT_REGION2_VEC_MODEL = "databio/r2v_encoder-ChIP-atlas-hg38" -DEFAULT_PEPHUB_NAMESPACE = "databio" -DEFAULT_PEPHUB_NAME = "bedbase_all" -DEFAULT_PEPHUB_TAG = "default" - DEFAULT_S3_BUCKET = "bedbase" diff --git a/bbconf/config_parser/models.py b/bbconf/config_parser/models.py index 8f1d37ec..69cd5e57 100644 --- a/bbconf/config_parser/models.py +++ b/bbconf/config_parser/models.py @@ -1,5 +1,6 @@ import logging from pathlib import Path +from typing import Literal from pydantic import BaseModel, ConfigDict, computed_field, field_validator from yacman import load_yaml @@ -9,9 +10,6 @@ DEFAULT_DB_DRIVER, DEFAULT_DB_NAME, DEFAULT_DB_PORT, - DEFAULT_PEPHUB_NAME, - DEFAULT_PEPHUB_NAMESPACE, - DEFAULT_PEPHUB_TAG, DEFAULT_QDRANT_BIVEC_COLLECTION_NAME, DEFAULT_QDRANT_FILE_COLLECTION_NAME, DEFAULT_QDRANT_HYBRID_COLLECTION_NAME, @@ -35,6 +33,7 @@ class ConfigDB(BaseModel): database: str = DEFAULT_DB_NAME dialect: str = DEFAULT_DB_DIALECT driver: str | None = DEFAULT_DB_DRIVER + run_migrations: bool = False model_config = ConfigDict(extra="forbid") @@ -74,14 +73,14 @@ class ConfigPath(BaseModel): class AccessMethodsStruct(BaseModel): type: str - description: str = None + description: str | None = None prefix: str class AccessMethods(BaseModel): - http: AccessMethodsStruct = None - s3: AccessMethodsStruct = None - local: AccessMethodsStruct = None + http: AccessMethodsStruct | None = None + s3: AccessMethodsStruct | None = None + local: AccessMethodsStruct | None = None class ConfigS3(BaseModel): @@ -120,20 +119,25 @@ def modify_access(self) -> bool: return False -class ConfigPepHubClient(BaseModel): - namespace: str | None = DEFAULT_PEPHUB_NAMESPACE - name: str | None = DEFAULT_PEPHUB_NAME - tag: str | None = DEFAULT_PEPHUB_TAG +class ConfigAnalysis(BaseModel): + """Analysis backend configuration. + + Controls which statistics engine is used for BED file analysis. + """ + + backend: Literal["r", "gtars"] = "r" + + model_config = ConfigDict(extra="forbid") class ConfigFile(BaseModel): database: ConfigDB - qdrant: ConfigQdrant = None + qdrant: ConfigQdrant | None = None server: ConfigServer path: ConfigPath - access_methods: AccessMethods = None - s3: ConfigS3 = None - phc: ConfigPepHubClient = None + access_methods: AccessMethods | None = None + s3: ConfigS3 | None = None + analysis: ConfigAnalysis | None = ConfigAnalysis() model_config = ConfigDict(extra="allow") diff --git a/bbconf/config_parser/utils.py b/bbconf/config_parser/utils.py index cba2e364..10376d09 100644 --- a/bbconf/config_parser/utils.py +++ b/bbconf/config_parser/utils.py @@ -1,7 +1,6 @@ import logging import yacman -from pephubclient.helpers import MessageHandler as m from pydantic_core._pydantic_core import ValidationError from bbconf.config_parser.models import ConfigFile @@ -28,7 +27,7 @@ def config_analyzer(config_path: str) -> bool: _LOGGER.info(f"Analyzing the configuration file {config_path}...") - _config = yacman.YAMLConfigManager(filepath=config_path).exp + _config = yacman.YAMLConfigManager.from_yaml_file(filepath=config_path).exp config_dict = {} for field_name, annotation in ConfigFile.model_fields.items(): @@ -56,6 +55,6 @@ def config_analyzer(config_path: str) -> bool: ) return False - m.print_success("Configuration file is valid! ") + _LOGGER.info("Configuration file is valid!") return True diff --git a/bbconf/db_utils.py b/bbconf/db_utils.py index 050c7e61..8dbf0ba1 100644 --- a/bbconf/db_utils.py +++ b/bbconf/db_utils.py @@ -1,20 +1,26 @@ import datetime import logging +import os from typing import Optional import pandas as pd +from alembic import command +from alembic.config import Config from sqlalchemy import ( + DDL, TIMESTAMP, BigInteger, ForeignKey, + Index, Result, Select, String, UniqueConstraint, event, select, + text, ) -from sqlalchemy.dialects.postgresql import ARRAY, JSON +from sqlalchemy.dialects.postgresql import ARRAY, JSON, JSONB from sqlalchemy.engine import URL, Engine, create_engine from sqlalchemy.event import listens_for from sqlalchemy.exc import IntegrityError, ProgrammingError @@ -105,9 +111,6 @@ class Bed(Base): default=False, comment="Whether file was tokenized and added to the vector database", ) - pephub: Mapped[bool] = mapped_column( - default=False, comment="Whether sample was added to pephub" - ) submission_date: Mapped[datetime.datetime] = mapped_column( default=deliver_update_date @@ -152,6 +155,30 @@ class Bed(Base): default=False, comment="Whether the bed file was processed" ) + __table_args__ = ( + # Backs the genome filter (Bed.genome_alias == genome) and the + # GROUP BY genome_alias aggregations. Historically created by hand in the + # live DB; declared here so a fresh create_all() reproduces it exactly + # (name and btree deduplication included). + Index( + "genome_alias_index", + "genome_alias", + postgresql_with={"deduplicate_items": "true"}, + ), + # Backs get_recent_beds / list_beds(order_by="submission_date"): + # ORDER BY submission_date DESC, id ASC LIMIT n. + Index("ix_bed_submission_date", text("submission_date DESC"), text("id")), + # Partial indexes for the background-worker backlog scans. They stay + # small and get faster as each queue drains toward empty. + Index("ix_bed_unprocessed", "id", postgresql_where=text("processed = false")), + Index("ix_bed_not_indexed", "id", postgresql_where=text("indexed = false")), + Index( + "ix_bed_not_file_indexed", + "id", + postgresql_where=text("file_indexed = false"), + ), + ) + class BedMetadata(Base): __tablename__ = "bed_metadata" @@ -255,8 +282,23 @@ class BedStats(Base): promotercore_percentage: Mapped[Optional[float]] tssdist: Mapped[Optional[float]] + distributions: Mapped[Optional[dict]] = mapped_column( + JSONB, + nullable=True, + comment="Full distribution arrays from gtars genomicdist (JSONB)", + ) + bed: Mapped["Bed"] = relationship("Bed", back_populates="stats") + __table_args__ = ( + # Backs the "beds missing computed stats" worker scan. + Index( + "ix_bed_stats_missing_regions", + "id", + postgresql_where=text("number_of_regions IS NULL"), + ), + ) + class Files(Base): __tablename__ = "files" @@ -304,7 +346,7 @@ class BedFileBedSetRelation(Base): ForeignKey("bedsets.id", ondelete="CASCADE"), primary_key=True ) bedfile_id: Mapped[str] = mapped_column( - ForeignKey("bed.id", ondelete="CASCADE"), primary_key=True + ForeignKey("bed.id", ondelete="CASCADE"), primary_key=True, index=True ) bedset: Mapped["BedSets"] = relationship("BedSets", back_populates="bedfiles") @@ -337,6 +379,15 @@ class BedSets(Base): bedset_standard_deviation: Mapped[Optional[dict]] = mapped_column( JSON, comment="Median values of the bedset" ) + bedset_stats: Mapped[Optional[dict]] = mapped_column( + JSONB, + nullable=True, + comment="Pre-aggregated distribution statistics from gtars (JSONB)", + ) + + bedfile_count: Mapped[int] = mapped_column( + default=0, comment="Number of bedfiles in the bedset (denormalized count)" + ) bedfiles: Mapped[list["BedFileBedSetRelation"]] = relationship( "BedFileBedSetRelation", back_populates="bedset", cascade="all, delete-orphan" @@ -351,6 +402,40 @@ class BedSets(Base): default=False, comment="Whether the bedset was processed" ) + __table_args__ = ( + # Backs the "unprocessed bedsets" worker scan. + Index( + "ix_bedsets_unprocessed", "id", postgresql_where=text("processed = false") + ), + # Trigram GIN indexes for the bedset search (get_ids_list), which filters + # on name/description with ILIKE '%query%'. A leading-wildcard ILIKE + # cannot use a btree index at all, so pg_trgm is the only thing that + # avoids a full table scan here. Needs the pg_trgm extension, which the + # before_create listener below creates on first creation of this table. + Index( + "ix_bedsets_name_trgm", + "name", + postgresql_using="gin", + postgresql_ops={"name": "gin_trgm_ops"}, + ), + Index( + "ix_bedsets_description_trgm", + "description", + postgresql_using="gin", + postgresql_ops={"description": "gin_trgm_ops"}, + ), + ) + + +# Make the pg_trgm extension available before the bedsets trigram GIN indexes are +# built. Scoped to this table's creation so it runs only on first-time schema +# creation (not on every startup) and only on PostgreSQL. +event.listen( + BedSets.__table__, + "before_create", + DDL("CREATE EXTENSION IF NOT EXISTS pg_trgm").execute_if(dialect="postgresql"), +) + class Universes(Base): __tablename__ = "universes" @@ -597,6 +682,84 @@ class UsageSearch(Base): date_to: Mapped[datetime.datetime] = mapped_column(comment="Date to") +class BedSnapshot(Base): + """ + Index of bulk metadata exports published to S3. + + One row per published artifact (metadata / bedsets / membership / manifest). + The exporter writes a row after a successful upload; the /v1/bed/exports + endpoint reads them newest-first. This is a new table, so + Base.metadata.create_all() creates it on the next connection. + """ + + __tablename__ = "bed_snapshots" + + id: Mapped[int] = mapped_column(primary_key=True, index=True, autoincrement=True) + file_path: Mapped[str] = mapped_column( + nullable=False, comment="S3 object key, relative to the bucket root" + ) + file_type: Mapped[str] = mapped_column( + nullable=False, comment="metadata | bedsets | bedset_membership | manifest" + ) + creation_date: Mapped[datetime.datetime] = mapped_column( + default=deliver_update_date, comment="Build date of the export" + ) + record_count: Mapped[Optional[int]] = mapped_column( + nullable=True, comment="Rows actually written to the file" + ) + 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" + ) + schema_version: Mapped[Optional[int]] = mapped_column( + nullable=True, comment="Export schema version" + ) + + +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. @@ -613,6 +776,7 @@ def __init__( drivername: str = POSTGRES_DIALECT, dsn: str | None = None, echo: bool = False, + run_migrations: bool = False, ): """ Initialize connection to the bedbase database. You can use the basic connection parameters @@ -627,6 +791,9 @@ def __init__( drivername: Driver used in connection. dsn: Libpq connection string using the dsn parameter (e.g. 'postgresql://user_name:password@host_name:port/db_name'). + run_migrations: Upgrade the database to the latest Alembic revision + (``head``) before connecting. Safe on an already-migrated or + pre-existing database (the initial revision is idempotent). """ if not dsn: dsn = URL.create( @@ -638,6 +805,13 @@ def __init__( drivername=drivername, ) + if run_migrations: + if isinstance(dsn, str): + migration_url = dsn + else: + migration_url = dsn.render_as_string(hide_password=False) + self.run_db_migration(migration_url) + self._engine = create_engine(dsn, echo=echo) self.create_schema(self._engine) self.check_db_connection() @@ -679,6 +853,25 @@ def delete_schema(self, engine=None) -> None: Base.metadata.drop_all(engine) return None + def run_db_migration(self, database_url: str) -> None: + """ + Upgrade the database to the latest Alembic revision (``head``). + + The Alembic config is built programmatically so the package does not + depend on the repo-root ``alembic.ini`` at runtime. + + Args: + database_url: SQLAlchemy connection URL (with password) to migrate. + """ + script_location = os.path.join(os.path.dirname(__file__), "alembic") + + alembic_cfg = Config() + alembic_cfg.set_main_option("script_location", script_location) + alembic_cfg.set_main_option("sqlalchemy.url", database_url) + + _LOGGER.info("Running database migrations to the latest revision...") + command.upgrade(alembic_cfg, "head") + def session_execute(self, statement: Select) -> Result: """ Execute statement using sqlalchemy statement. diff --git a/bbconf/exceptions.py b/bbconf/exceptions.py index 3ad393cc..ccb0b37d 100644 --- a/bbconf/exceptions.py +++ b/bbconf/exceptions.py @@ -70,6 +70,20 @@ class BedSetExistsError(BedBaseConfError): pass +class SnapshotNotFoundError(BedBaseConfError): + """ + Error type for missing snapshot""" + + 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 ef3efd59..dd1a2a37 100644 --- a/bbconf/models/base_models.py +++ b/bbconf/models/base_models.py @@ -103,3 +103,64 @@ class FileStats(BaseModel): file_size: BinValues number_of_regions: BinValues geo: GEOStatistics + + +class BedSnapshotArtifact(BaseModel): + """A built snapshot file to publish (upload to S3 + record in the database).""" + + path: str # local file path to upload + file_type: str + record_count: int | None = None + file_size: int | None = None + checksum: str | None = None + schema_version: int | None = None + + +class BedSnapshotResult(BaseModel): + """One published bulk-export artifact.""" + + file_path: str + file_type: str + creation_date: datetime.datetime + record_count: int | None = None + file_size: int | None = None + checksum: str | None = None + schema_version: int | None = None + + +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/models/bed_models.py b/bbconf/models/bed_models.py index c7056492..1103594f 100644 --- a/bbconf/models/bed_models.py +++ b/bbconf/models/bed_models.py @@ -74,34 +74,9 @@ class BedStatsModel(BaseModel): promoterprox_frequency: float | None = None promoterprox_percentage: float | None = None - model_config = ConfigDict(extra="ignore", populate_by_name=True) - - -class BedPEPHub(BaseModel): - sample_name: str | None = "" - genome: str | None = "" - organism: str | None = "" - species_id: str | None = "" - cell_type: str | None = "" - cell_line: str | None = "" - assay: str | None = Field("", description="Experimental protocol (e.g. ChIP-seq)") - library_source: str | None = Field( - "", description="Library source (e.g. genomic, transcriptomic)" - ) - genotype: str | None = Field("", description="Genotype of the sample") - target: str | None = Field("", description="Target of the assay (e.g. H3K4me3)") - antibody: str | None = Field("", description="Antibody used in the assay") - treatment: str | None = Field( - "", description="Treatment of the sample (e.g. drug treatment)" - ) - tissue: str | None = Field("", description="Tissue type") - global_sample_id: str | None = Field("", description="Global sample identifier") - global_experiment_id: str | None = Field( - "", description="Global experiment identifier" - ) - description: str | None = Field("", description="Description of the sample") + distributions: dict | None = None - model_config = ConfigDict(extra="allow", populate_by_name=True) + model_config = ConfigDict(extra="ignore", populate_by_name=True) class StandardMeta(BaseModel): @@ -165,10 +140,6 @@ def ensure_list(cls, v: str | list[str]) -> list[str]: raise ValueError("values must be a string or a list of strings") -class BedPEPHubRestrict(BedPEPHub): - model_config = ConfigDict(extra="ignore") - - class BedMetadataBasic(BedClassification): id: str name: str | None = "" @@ -190,6 +161,7 @@ class BedSetMinimal(BaseModel): id: str name: str | None = None description: str | None = None + bedfile_count: int = 0 class BedMetadataAll(BedMetadataBasic): @@ -197,7 +169,6 @@ class BedMetadataAll(BedMetadataBasic): plots: BedPlots | None = None files: BedFiles | None = None universe_metadata: UniverseMetadata | None = None - raw_metadata: BedPEPHub | BedPEPHubRestrict | None = None bedsets: list[BedSetMinimal] | None = None diff --git a/bbconf/models/bedset_models.py b/bbconf/models/bedset_models.py index ca074a99..8acbf141 100644 --- a/bbconf/models/bedset_models.py +++ b/bbconf/models/bedset_models.py @@ -1,4 +1,5 @@ import datetime +from typing import Optional from pydantic import BaseModel, ConfigDict, model_validator @@ -7,10 +8,34 @@ class BedSetStats(BaseModel): + """Bedset statistics: mean/sd of scalar columns. + + Populated from bedset_means and bedset_standard_deviation database columns. + """ + mean: BedStatsModel = None sd: BedStatsModel = None +class BedSetDistributions(BaseModel): + """Collection-level aggregated distribution statistics for a bedset. + + Stored in the bedset_stats JSONB database column. Populated when + member bed files have been processed with the gtars analysis backend. + """ + + n_files: int = 0 + composition: Optional[dict] = None + scalar_summaries: Optional[dict] = None + tss_histogram: Optional[dict] = None + widths_histogram: Optional[dict] = None + neighbor_distances: Optional[dict] = None + gc_content: Optional[dict] = None + region_distribution: Optional[dict] = None + partitions: Optional[dict] = None + chromosome_summaries: Optional[dict] = None + + class BedSetPlots(BaseModel): region_commonality: FileModel = None @@ -24,10 +49,12 @@ class BedSetMetadata(BaseModel): submission_date: datetime.datetime = None last_update_date: datetime.datetime = None statistics: BedSetStats | None = None + distributions: BedSetDistributions | None = None plots: BedSetPlots | None = None description: str = None summary: str = None bed_ids: list[str] = None + bedfile_count: int = 0 author: str | None = None source: str | None = None diff --git a/bbconf/modules/analysis_files.py b/bbconf/modules/analysis_files.py new file mode 100644 index 00000000..1777a400 --- /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/bbconf/modules/bedfiles.py b/bbconf/modules/bedfiles.py index c45d0bea..7a594f46 100644 --- a/bbconf/modules/bedfiles.py +++ b/bbconf/modules/bedfiles.py @@ -6,7 +6,6 @@ from geniml.bbclient import BBClient from geniml.search.backends import QdrantBackend from gtars.models import RegionSet as GRegionSet -from pephubclient.exceptions import ResponseError from pydantic import BaseModel from qdrant_client import models from qdrant_client.http.exceptions import UnexpectedResponse @@ -15,7 +14,7 @@ from sqlalchemy import and_, cast, delete, func, or_, select from sqlalchemy.dialects import postgresql from sqlalchemy.exc import IntegrityError -from sqlalchemy.orm import Session, aliased +from sqlalchemy.orm import Session, aliased, selectinload from sqlalchemy.orm.attributes import flag_modified from tqdm import tqdm @@ -52,8 +51,6 @@ BedListSearchResult, BedMetadataAll, BedMetadataBasic, - BedPEPHub, - BedPEPHubRestrict, BedPlots, BedSetMinimal, BedStatsModel, @@ -100,27 +97,62 @@ def get(self, identifier: str, full: bool = False) -> BedMetadataAll: Args: identifier: Bed file identifier. - full: If True, return full metadata, including statistics, files, and raw metadata from pephub. + full: If True, return full metadata, including statistics and files. Returns: BED file metadata. """ statement = select(Bed).where(and_(Bed.id == identifier)) - bed_plots = BedPlots() - bed_files = BedFiles() - with Session(self._sa_engine) as session: bed_object = session.scalar(statement) if not bed_object: raise BEDFileNotFoundError(f"Bed file with id: {identifier} not found.") - if full: - for result in bed_object.files: - # PLOTS - if result.name in BedPlots.model_fields: + return self._build_metadata(bed_object, full=full) + + def _build_metadata(self, bed_object: Bed, full: bool = False) -> BedMetadataAll: + """ + Build a BedMetadataAll model from a Bed ORM object. + + For ``full=True`` this assembles plots, files, stats, bedsets, and + universe metadata (which lazy-load relationships, so the caller must + keep the SQLAlchemy session open). For ``full=False`` only scalar + columns and the (joined-loaded) annotations are accessed, so the + Bed object may be detached from its session. + + Args: + bed_object: Bed ORM object to build metadata from. + full: If True, return full metadata, including statistics and files. + + Returns: + BED file metadata. + """ + identifier = bed_object.id + + bed_plots = BedPlots() + bed_files = BedFiles() + + if full: + for result in bed_object.files: + # PLOTS + if result.name in BedPlots.model_fields: + setattr( + bed_plots, + result.name, + FileModel( + **result.__dict__, + object_id=f"bed.{identifier}.{result.name}", + access_methods=self.config.construct_access_method_list( + result.path + ), + ), + ) + # FILES + elif result.name in BedFiles.model_fields: + ( setattr( - bed_plots, + bed_files, result.name, FileModel( **result.__dict__, @@ -129,64 +161,35 @@ def get(self, identifier: str, full: bool = False) -> BedMetadataAll: result.path ), ), - ) - # FILES - elif result.name in BedFiles.model_fields: - ( - setattr( - bed_files, - result.name, - FileModel( - **result.__dict__, - object_id=f"bed.{identifier}.{result.name}", - access_methods=self.config.construct_access_method_list( - result.path - ), - ), - ), - ) - - else: - _LOGGER.error( - f"Unknown file type: {result.name}. And is not in the model fields. Skipping.." - ) - bed_stats = BedStatsModel(**bed_object.stats.__dict__) - bed_bedsets = [] - for relation in bed_object.bedsets: - bed_bedsets.append( - BedSetMinimal( - id=relation.bedset.id, - description=relation.bedset.description, - name=relation.bedset.name, - ) + ), ) - if bed_object.universe: - universe_meta = UniverseMetadata(**bed_object.universe.__dict__) else: - universe_meta = UniverseMetadata() - else: - bed_plots = None - bed_files = None - bed_stats = None - universe_meta = None - bed_bedsets = [] - - try: - if full: - bed_metadata = BedPEPHubRestrict( - **self.config.phc.sample.get( - namespace=self.config.config.phc.namespace, - name=self.config.config.phc.name, - tag=self.config.config.phc.tag, - sample_name=identifier, + _LOGGER.error( + f"Unknown file type: {result.name}. And is not in the model fields. Skipping.." + ) + bed_stats = BedStatsModel(**bed_object.stats.__dict__) + bed_bedsets = [] + for relation in bed_object.bedsets: + bed_bedsets.append( + BedSetMinimal( + id=relation.bedset.id, + description=relation.bedset.description, + name=relation.bedset.name, + bedfile_count=relation.bedset.bedfile_count, ) ) + + if bed_object.universe: + universe_meta = UniverseMetadata(**bed_object.universe.__dict__) else: - bed_metadata = None - except Exception as e: - _LOGGER.warning(f"Could not retrieve metadata from pephub. Error: {e}") - bed_metadata = None + universe_meta = UniverseMetadata() + else: + bed_plots = None + bed_files = None + bed_stats = None + universe_meta = None + bed_bedsets = [] return BedMetadataAll( id=bed_object.id, @@ -197,7 +200,6 @@ def get(self, identifier: str, full: bool = False) -> BedMetadataAll: description=bed_object.description, submission_date=bed_object.submission_date, last_update_date=bed_object.last_update_date, - raw_metadata=bed_metadata, genome_alias=bed_object.genome_alias, genome_digest=bed_object.genome_digest, bed_compliance=bed_object.bed_compliance, @@ -290,17 +292,32 @@ def get_neighbours( limit=limit, offset=offset, ) - result_list = [] - for result in results.points: - result_id = result.id.replace("-", "") - result_list.append( - QdrantSearchResult( - id=result_id, - payload=result.payload, - score=result.score, - metadata=self.get(result_id, full=False), - ) + # Hydrate all neighbours with a single batched query instead of one + # SELECT per neighbour (was an N+1). annotations is joined-loaded, + # but selectinload keeps that explicit for this detached-object path. + ids = [result.id.replace("-", "") for result in results.points] + with Session(self._sa_engine) as session: + beds = { + bed.id: bed + for bed in session.scalars( + select(Bed) + .where(Bed.id.in_(ids)) + .options(selectinload(Bed.annotations)) + ).all() + } + result_list = [ + QdrantSearchResult( + id=result.id.replace("-", ""), + payload=result.payload, + score=result.score, + metadata=self._build_metadata( + beds[result.id.replace("-", "")], full=False + ), ) + for result in results.points + # skip stale Qdrant points that no longer exist in the database + if result.id.replace("-", "") in beds + ] except UnexpectedResponse as err: _LOGGER.error( f"Qdrant request failed. Error: {err}. Returning empty result set." @@ -347,28 +364,6 @@ def get_files(self, identifier: str) -> BedFiles: ) return bed_files - def get_raw_metadata(self, identifier: str) -> BedPEPHub: - """ - Get file metadata by identifier. - - Args: - identifier: Bed file identifier. - - Returns: - BED file raw metadata. - """ - try: - bed_metadata = self.config.phc.sample.get( - namespace=self.config.config.phc.namespace, - name=self.config.config.phc.name, - tag=self.config.config.phc.tag, - sample_name=identifier, - ) - except Exception as e: - _LOGGER.warning(f"Could not retrieve metadata from pephub. Error: {e}") - bed_metadata = {} - return BedPEPHubRestrict(**bed_metadata) - def get_classification(self, identifier: str) -> BedClassification: """ Get file classification by identifier. @@ -470,7 +465,7 @@ def get_ids_list( and_(Bed.bed_compliance == bed_compliance) ) - statement = statement.limit(limit).offset(offset) + statement = statement.order_by(Bed.id).limit(limit).offset(offset) result_list = [] with Session(self._sa_engine) as session: @@ -557,7 +552,6 @@ def add( ref_validation: dict[str, BaseModel] | None = None, license_id: str = DEFAULT_LICENSE, upload_qdrant: bool = False, - upload_pephub: bool = False, upload_s3: bool = False, local_path: str = None, overwrite: bool = False, @@ -570,7 +564,7 @@ def add( Args: identifier: Bed file identifier. stats: Bed file results {statistics, plots, files, metadata}. - metadata: Bed file metadata (will be saved in pephub). + metadata: Bed file metadata. plots: Bed file plots. files: Bed file files. classification: Bed file classification. @@ -578,11 +572,10 @@ def add( license_id: Bed file license id (default: 'DUO:0000042'). Full list of licenses: https://raw.githubusercontent.com/EBISPOT/DUO/master/duo.csv upload_qdrant: Add bed file to qdrant indexes. - upload_pephub: Add bed file to pephub. upload_s3: Upload files to s3. local_path: Local path to the output files. overwrite: Overwrite bed file if it already exists. - nofail: Do not raise an error for error in pephub/s3/qdrant or record exists and not overwrite. + nofail: Do not raise an error for error in s3/qdrant or record exists and not overwrite. processed: True if bedfile was processed and statistics and plots were calculated. Returns: @@ -634,23 +627,6 @@ def add( bed_metadata = StandardMeta(**metadata) classification = BedClassification(**classification) - if upload_pephub: - pephub_metadata = BedPEPHub(**metadata) - try: - self.upload_pephub( - identifier, - pephub_metadata.model_dump(exclude=set("input_file")), - overwrite, - ) - except Exception as e: - _LOGGER.warning( - f"Could not upload to pephub. Error: {e}. nofail: {nofail}" - ) - upload_pephub = False - if not nofail: - raise e - else: - _LOGGER.info("upload_pephub set to false. Skipping pephub..") if upload_qdrant: if classification.genome_alias == "hg38": @@ -686,7 +662,6 @@ def add( description=bed_metadata.description, license_id=license_id, indexed=upload_qdrant, - pephub=upload_pephub, processed=processed, ) session.add(new_bed) @@ -758,7 +733,6 @@ def update( ref_validation: dict[str, BaseModel] | None = None, license_id: str = DEFAULT_LICENSE, upload_qdrant: bool = False, - upload_pephub: bool = False, upload_s3: bool = True, local_path: str = None, overwrite: bool = False, @@ -771,18 +745,17 @@ def update( Args: identifier: Bed file identifier. stats: Bed file results {statistics, plots, files, metadata}. - metadata: Bed file metadata (will be saved in pephub). + metadata: Bed file metadata. plots: Bed file plots. files: Bed file files. classification: Bed file classification. ref_validation: Reference validation data. RefGenValidModel. license_id: Bed file license id (default: 'DUO:0000042'). upload_qdrant: Add bed file to qdrant indexes. - upload_pephub: Add bed file to pephub. upload_s3: Upload files to s3. local_path: Local path to the output files. overwrite: Overwrite bed file if it already exists. - nofail: Do not raise an error for error in pephub/s3/qdrant or record exists and not overwrite. + nofail: Do not raise an error for error in s3/qdrant or record exists and not overwrite. processed: True if bedfile was processed and statistics and plots were calculated. Returns: @@ -806,19 +779,6 @@ def update( bed_metadata = StandardMeta(**metadata if metadata else {}) classification = BedClassification(**classification if classification else {}) - if upload_pephub and metadata: - metadata = BedPEPHub(**metadata) - try: - self.update_pephub(identifier, metadata.model_dump(), overwrite) - except Exception as e: - _LOGGER.warning( - f"Could not upload to pephub. Error: {e}. nofail: {nofail}" - ) - if not nofail: - raise e - else: - _LOGGER.info("upload_pephub set to false. Skipping pephub..") - if upload_qdrant: if classification.genome_alias == "hg38": _LOGGER.info(f"Uploading bed file to qdrant.. [{identifier}]") @@ -1152,65 +1112,15 @@ def delete(self, identifier: str) -> None: bed_object = session.scalar(statement) files = [FileModel(**k.__dict__) for k in bed_object.files] - delete_pephub = bed_object.pephub delete_qdrant = bed_object.indexed session.delete(bed_object) session.commit() - if delete_pephub: - self.delete_pephub_sample(identifier) if delete_qdrant: self.delete_qdrant_point(identifier) self.config.delete_files_s3(files) - def upload_pephub(self, identifier: str, metadata: dict, overwrite: bool = False): - if not metadata: - _LOGGER.warning("No metadata provided. Skipping pephub upload..") - return False - self.config.phc.sample.create( - namespace=self.config.config.phc.namespace, - name=self.config.config.phc.name, - tag=self.config.config.phc.tag, - sample_name=identifier, - sample_dict=metadata, - overwrite=overwrite, - ) - - def update_pephub( - self, identifier: str, metadata: dict, overwrite: bool = False - ) -> None: - try: - if not metadata: - _LOGGER.warning("No metadata provided. Skipping pephub upload..") - return None - self.config.phc.sample.update( - namespace=self.config.config.phc.namespace, - name=self.config.config.phc.name, - tag=self.config.config.phc.tag, - sample_name=identifier, - sample_dict=metadata, - ) - except ResponseError as e: - _LOGGER.warning(f"Could not update pephub. Error: {e}") - - def delete_pephub_sample(self, identifier: str): - """ - Delete sample from pephub. - - Args: - identifier: Bed file identifier. - """ - try: - self.config.phc.sample.remove( - namespace=self.config.config.phc.namespace, - name=self.config.config.phc.name, - tag=self.config.config.phc.tag, - sample_name=identifier, - ) - except ResponseError as e: - _LOGGER.warning(f"Could not delete from pephub. Error: {e}") - def upload_file_qdrant( self, bed_id: str, @@ -1382,8 +1292,15 @@ def bed_to_bed_search( continue if result_meta: results_list.append(QdrantSearchResult(**result, metadata=result_meta)) + + # Count of the searchable pool (indexed bed vectors), not the total number + # of bed files in the database (which overcounts unindexed genomes). + count = self.config.qdrant_client.count( + collection_name=self.config.config.qdrant.file_collection, + exact=True, + ).count return BedListSearchResult( - count=self.bb_agent.get_stats().bedfiles_number, + count=count, limit=limit, offset=offset, results=results_list, diff --git a/bbconf/modules/bedsets.py b/bbconf/modules/bedsets.py index 83566ac3..5b0d38c7 100644 --- a/bbconf/modules/bedsets.py +++ b/bbconf/modules/bedsets.py @@ -89,6 +89,7 @@ def get(self, identifier: str, full: bool = False) -> BedSetMetadata: statistics=stats, plots=plots, bed_ids=list_of_bedfiles, + bedfile_count=bedset_obj.bedfile_count, submission_date=bedset_obj.submission_date, last_update_date=bedset_obj.last_update_date, author=bedset_obj.author, @@ -305,7 +306,6 @@ def create( statistics: bool = False, annotation: dict | None = None, plots: dict | None = None, - upload_pephub: bool = False, upload_s3: bool = False, local_path: str = "", no_fail: bool = False, @@ -323,7 +323,6 @@ def create( statistics: Calculate statistics for bedset. annotation: Bedset annotation (author, source). plots: Dictionary with plots. - upload_pephub: Upload bedset to pephub (create view in pephub). upload_s3: Upload bedset to s3. local_path: Local path to the output files. no_fail: Do not raise an error if bedset already exists. @@ -356,13 +355,8 @@ def create( if not isinstance(annotation, dict): annotation = {} - if upload_pephub: - try: - self._create_pephub_view(identifier, description, bedid_list, no_fail) - except Exception as e: - _LOGGER.error(f"Failed to create view in pephub: {e}") - if not no_fail: - raise e + if no_fail: + bedid_list = list(set(bedid_list)) new_bedset = BedSets( id=identifier, @@ -375,6 +369,7 @@ def create( author=annotation.get("author"), source=annotation.get("source"), processed=processed, + bedfile_count=len(bedid_list), ) if upload_s3: @@ -387,8 +382,6 @@ def create( with Session(self._db_engine.engine) as session: session.add(new_bedset) - if no_fail: - bedid_list = list(set(bedid_list)) for bedfile in bedid_list: session.add( BedFileBedSetRelation(bedset_id=identifier, bedfile_id=bedfile) @@ -427,7 +420,11 @@ def _calculate_statistics(self, bed_ids: list[str]) -> BedSetStats: """ _LOGGER.info("Calculating bedset statistics") - numeric_columns = BedStatsModel.model_fields + numeric_columns = [ + name + for name, field in BedStatsModel.model_fields.items() + if field.annotation in (float, float | None) + ] bedset_sd = {} bedset_mean = {} @@ -459,47 +456,15 @@ def _calculate_statistics(self, bed_ids: list[str]) -> BedSetStats: _LOGGER.info("Bedset statistics were calculated successfully") return bedset_stats - def _create_pephub_view( - self, - bedset_id: str, - description: str = None, - bed_ids: list = None, - nofail: bool = False, - ) -> None: - """ - Create view in pephub for bedset. - - Args: - bedset_id: Bedset identifier. - description: Bedset description. - bed_ids: List of bed file identifiers. - nofail: Do not raise an error if sample not found. - - Returns: - None. - """ - - _LOGGER.info(f"Creating view in pephub for bedset '{bedset_id}'") - try: - self.config.phc.view.create( - namespace=self.config.config.phc.namespace, - name=self.config.config.phc.name, - tag=self.config.config.phc.tag, - view_name=bedset_id, - # description=description, - sample_list=bed_ids, - ) - except Exception as e: - _LOGGER.error(f"Failed to create view in pephub: {e}") - if not nofail: - raise e - return None - def get_ids_list( - self, query: str = None, limit: int = 10, offset: int = 0 + self, query: str | None = None, limit: int = 10, offset: int = 0 ) -> BedSetListResult: """ - Get list of bedsets from the database. + Find (search) bedsets from the database. + + Use `get(identifier)` to + fetch a single bedset's member ids. `bedfile_count` is populated + directly from the denormalized column, so it's free. Args: query: Search query. @@ -509,7 +474,7 @@ def get_ids_list( Returns: List of bedsets. """ - statement = select(BedSets.id) + statement = select(BedSets) count_statement = select(func.count(BedSets.id)) if query: query = query.strip() @@ -528,12 +493,24 @@ def get_ids_list( ) with Session(self._db_engine.engine) as session: - bedset_list = session.execute(statement.limit(limit).offset(offset)) + bedset_list = session.scalars(statement.limit(limit).offset(offset)) bedset_count = session.execute(count_statement).one() - result_list = [] - for bedset_id in bedset_list: - result_list.append(self.get(bedset_id[0])) + result_list = [ + BedSetMetadata( + id=bedset_obj.id, + name=bedset_obj.name, + description=bedset_obj.description, + md5sum=bedset_obj.md5sum, + bedfile_count=bedset_obj.bedfile_count, + submission_date=bedset_obj.submission_date, + last_update_date=bedset_obj.last_update_date, + author=bedset_obj.author, + source=bedset_obj.source, + ) + for bedset_obj in bedset_list + ] + return BedSetListResult( count=bedset_count[0], limit=limit, @@ -601,35 +578,9 @@ def delete(self, identifier: str) -> None: session.delete(bedset_obj) session.commit() - self.delete_phc_view(identifier, nofail=True) if files: self.config.delete_files_s3(files) - def delete_phc_view(self, identifier: str, nofail: bool = False) -> None: - """ - Delete view in pephub. - - Args: - identifier: Bedset identifier. - nofail: Do not raise an error if view not found. - - Returns: - None. - """ - _LOGGER.info(f"Deleting view in pephub for bedset '{identifier}'") - try: - self.config.phc.view.delete( - namespace=self.config.config.phc.namespace, - name=self.config.config.phc.name, - tag=self.config.config.phc.tag, - view_name=identifier, - ) - except Exception as e: - _LOGGER.error(f"Failed to delete view in pephub: {e}") - if not nofail: - raise e - return None - def exists(self, identifier: str) -> bool: """ Check if bedset exists in the database. @@ -688,6 +639,7 @@ def get_unprocessed(self, limit: int = 100, offset: int = 0) -> BedSetListResult statistics=None, plots=None, bed_ids=list_of_bedfiles, + bedfile_count=bedset_obj.bedfile_count, submission_date=bedset_obj.submission_date, last_update_date=bedset_obj.last_update_date, author=bedset_obj.author, diff --git a/bbconf/modules/snapshots.py b/bbconf/modules/snapshots.py new file mode 100644 index 00000000..49e7ee2c --- /dev/null +++ b/bbconf/modules/snapshots.py @@ -0,0 +1,260 @@ +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 BedSnapshot +from bbconf.exceptions import SnapshotNotFoundError +from bbconf.models.base_models import ( + BedSnapshotArtifact, + BedSnapshotListResult, + BedSnapshotResult, +) + +_LOGGER = logging.getLogger(PKG_NAME) + +# All snapshots live under this single S3 prefix. Not configurable. +SNAPSHOT_S3_PREFIX = "snapshot" + + +class BedAgentSnapshot: + """ + Class that manages bulk-export snapshots (the ``bed_snapshots`` index). + + One row per published artifact (metadata / bedsets / bedset_membership / + manifest). Adding a snapshot always uploads the file to S3 *and* records it + in the database; both writes live here in bbconf. This class also exposes + read (``list`` / ``get``) and ``delete`` helpers. + """ + + def __init__(self, config: BedBaseConfig): + """ + Initialize BedAgentSnapshot. + + Args: + config: Config object. + """ + self.config = config + self._db_engine = self.config.db_engine + + def add( + self, + artifacts: BedSnapshotArtifact | list[BedSnapshotArtifact], + creation_date: datetime | None = None, + ) -> BedSnapshotListResult: + """ + Add snapshot artifacts: upload each to S3 and record it in the database. + + Every artifact is uploaded under the fixed ``snapshot/`` prefix and then + recorded in ``bed_snapshots``. 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 ``file_type`` and file metadata. + creation_date: Build date recorded on every row + (defaults to now, UTC). + + Returns: + The created snapshot rows. + """ + if isinstance(artifacts, BedSnapshotArtifact): + artifacts = [artifacts] + if creation_date is None: + creation_date = datetime.now(timezone.utc) + + # Upload everything first; only record rows once all uploads succeed. + results: list[BedSnapshotResult] = [] + for artifact in artifacts: + key = f"{SNAPSHOT_S3_PREFIX}/{os.path.basename(artifact.path)}" + self.config.upload_s3(artifact.path, s3_path=key) + results.append( + BedSnapshotResult( + file_path=key, + file_type=artifact.file_type, + creation_date=creation_date, + record_count=artifact.record_count, + file_size=artifact.file_size, + checksum=artifact.checksum, + schema_version=artifact.schema_version, + ) + ) + + with Session(self._db_engine.engine) as session: + for result in results: + session.add( + BedSnapshot( + file_path=result.file_path, + file_type=result.file_type, + creation_date=result.creation_date, + record_count=result.record_count, + file_size=result.file_size, + checksum=result.checksum, + schema_version=result.schema_version, + ) + ) + session.commit() + + _LOGGER.info(f"Recorded {len(results)} rows in bed_snapshots") + return BedSnapshotListResult(count=len(results), results=results) + + def delete(self, id: int, remove_s3: bool = True) -> None: + """ + Delete a snapshot index row. + + Args: + id: Primary key of the snapshot row. + remove_s3: Also delete the underlying S3 object. + + Returns: + None. + + Raises: + SnapshotNotFoundError: If no row with this id exists. + """ + with Session(self._db_engine.engine) as session: + row = session.scalar(select(BedSnapshot).where(BedSnapshot.id == id)) + if row is None: + raise SnapshotNotFoundError(f"Snapshot 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, + limit: int | None = 100, + offset: int = 0, + ) -> BedSnapshotListResult: + """ + List all snapshot index rows in the database, newest first. + + Args: + file_type: Optional filter on file type. + limit: Maximum number of rows to return. ``None`` returns all rows. + offset: Number of rows to skip. + + Returns: + List of snapshots and the total matching count. + """ + statement = select(BedSnapshot) + count_statement = select(func.count()).select_from(BedSnapshot) + if file_type is not None: + statement = statement.where(BedSnapshot.file_type == file_type) + count_statement = count_statement.where(BedSnapshot.file_type == file_type) + statement = statement.order_by( + BedSnapshot.creation_date.desc(), BedSnapshot.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 BedSnapshotListResult(count=total, results=results) + + def get_by_filename(self, filename: str) -> BedSnapshotResult: + """ + Resolve a snapshot by its file name (the basename of its S3 key). + + Returns the newest row whose ``file_path`` basename equals ``filename``. + Used to round-trip an export's DRS object-id back to its row. + + Args: + filename: The bare file name, e.g. + ``bedbase_metadata_2026_08_03.parquet``. + + Returns: + The matching snapshot row. + + Raises: + SnapshotNotFoundError: If no row matches. + """ + filename = os.path.basename(filename) + with Session(self._db_engine.engine) as session: + rows = session.scalars( + select(BedSnapshot) + .where(BedSnapshot.file_path.like(f"%{filename}")) + .order_by(BedSnapshot.creation_date.desc(), BedSnapshot.id.desc()) + ).all() + for row in rows: + if os.path.basename(row.file_path) == filename: + return self._to_result(row) + raise SnapshotNotFoundError(f"Snapshot '{filename}' not found.") + + def delete_by_checksum(self, checksum: str, remove_s3: bool = True) -> None: + """ + Delete snapshot index rows by their checksum. + + Deletes every ``bed_snapshots`` row whose ``checksum`` matches (a checksum + identifies one file's content) and optionally removes the underlying S3 + objects. + + Args: + checksum: SHA256 checksum of the snapshot file. + remove_s3: Also delete the underlying S3 object(s). + + Returns: + None. + + Raises: + SnapshotNotFoundError: If no row matches the checksum. + """ + with Session(self._db_engine.engine) as session: + rows = session.scalars( + select(BedSnapshot).where(BedSnapshot.checksum == checksum) + ).all() + if not rows: + raise SnapshotNotFoundError( + f"Snapshot 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) + + def get(self, id: int) -> BedSnapshotResult: + """ + Get a single snapshot index row by id. + + Args: + id: Primary key of the snapshot row. + + Returns: + The snapshot row. + + Raises: + SnapshotNotFoundError: If no row with this id exists. + """ + with Session(self._db_engine.engine) as session: + row = session.scalar(select(BedSnapshot).where(BedSnapshot.id == id)) + if row is None: + raise SnapshotNotFoundError(f"Snapshot with id '{id}' not found.") + return self._to_result(row) + + @staticmethod + def _to_result(row: BedSnapshot) -> BedSnapshotResult: + return BedSnapshotResult( + file_path=row.file_path, + file_type=row.file_type, + creation_date=row.creation_date, + record_count=row.record_count, + file_size=row.file_size, + checksum=row.checksum, + schema_version=row.schema_version, + ) diff --git a/docs/changelog.md b/docs/changelog.md index a5fe3fcb..f0c7efe4 100644 --- a/docs/changelog.md +++ b/docs/changelog.md @@ -3,44 +3,74 @@ This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html) and [Keep a Changelog](https://keepachangelog.com/en/1.0.0/) format. -### [0.14.12] - 2026-04-22 +### [0.14.13] - 2026-07-13 +### Fixed: +- Cache `get_stats()` with a TTL to avoid running uncached COUNT queries on the bed table on every request to hot API paths (stats, neighbours, list, search) +- Eliminated an N+1 query in `get_neighbours()` by fetching all neighbour metadata in a single batched query (with annotations eager-loaded) instead of one query per neighbour; stale Qdrant points are now skipped rather than raising +- Eliminated an N+1 in `BedAgentBedSet.get_ids_list()`: it was refetching each bedset by id and lazy-loading its full bedfile membership just to build the list page. Now builds results directly from the paginated query; `bed_ids` is left unpopulated on list results (use `get(identifier)` for a single bedset's member ids) +- `get_detailed_stats()` no longer reuses a `Session` after its `with` block has closed (was forcing 3 extra connection checkouts for `_stats_comments`/`_stats_geo_status`/`_get_geo_stats`); all queries now share one session/transaction +- Replaced the `bed_files_info()` call inside `get_detailed_stats()` with a targeted 3-column query, avoiding a full-table `FileInfo` Pydantic construction (with per-row try/except) for every bed record just to extract `number_of_regions`/`mean_region_width`/`file_size` for histogram binning +- `BedAgentBedFile.get_ids_list()` (backs `/bed/list`) had no `order_by()` on its paginated query, so row order across pages was undefined -- rows could be duplicated or skipped between requests. Now orders by `Bed.id`. +- `get_detailed_stats()` crashed with a pydantic `ValidationError` whenever `bed_compliance`, `data_format`, `genome_alias`, `species_name`, `assay`, or `cell_line` had NULL rows: the `GROUP BY` queries included the NULL group, producing a `None` dict key, which `FileStats`'s `dict[str, int]` fields reject. All six queries now filter out NULLs before grouping. + + +### Added: +- Added a denormalized `bedfile_count` column to `bedsets`, exposed as `BedSetMetadata.bedfile_count`. Set once at bedset creation time (membership is write-once; `add_bedfile`/`delete_bedfile` are unimplemented), so reads never need to touch `bedfile_bedset_relation` to know a bedset's size. Requires a DB migration -- see `scripts/migrations/2026_07_31_add_bedset_bedfile_count.sql` + +## [0.15.0] - 2026-08-17 +### Added: +- Alembic migration support, including `alembic.ini`, migration script templates, and configuration files in `bbconf/alembic/`, with clear instructions in the `README.md` for generating and applying migrations. +- TTL-based cache (with locking) for the `get_stats()` method in `bbconf/bbagent.py` to avoid repeated expensive COUNT queries on hot API paths. +- jsonb columns for genomic distribution data storage in bedfiles and bedsets +- missing imports and properties for snapshot support in `bbconf/bbagent.py` + +### Updated: +- Updated binning functions to handle empty input lists gracefully, preventing errors when there is no data. +- Refactored `get_detailed_stats()` to gather numeric statistics in a single optimized query, filter out null values, and avoid loading unnecessary objects, increasing efficiency and accuracy. +- Minor workflow YAML formatting fix. +- Updated database indexes, making data query faster +- Updated bedhost endpoints, making them more efficient +- Updated the pull request template to require confirmation of completed migration steps when schema changes are made. + + +## [0.14.12] - 2026-04-22 ### Changed: - Updated yacman version to 2.0.0 -### [0.14.11] - 2026-04-15 +## [0.14.11] - 2026-04-15 ### Fixed: - External id search -### [0.14.10] - 2026-04-05 +## [0.14.10] - 2026-04-05 ### Fixed: - version info bug -### [0.14.9] - 2026-02-26 +## [0.14.9] - 2026-02-26 ### Changed: - Modernized docstrings - Type annotation for python 3.10+ - Updated requirements - Updated package installation way to use pyproject.toml and hatchling -### [0.14.8] - 2026-02-17 +## [0.14.8] - 2026-02-17 ### Changed: - Updated versions of dependencies -### [0.14.7] - 2026-02-16 +## [0.14.7] - 2026-02-16 ### Changed: - Updated requirements -### [0.14.6] - 2026-02-06 +## [0.14.6] - 2026-02-06 ### Fixed: - Fixed qdrant upload exception catching -### [0.14.5] - 2026-02-05 +## [0.14.5] - 2026-02-05 ### Changed: - Updated reindexing script -### [0.14.4] - 2026-02-04 +## [0.14.4] - 2026-02-04 ### Changed: - Updated reindexing of bed files to use only verified genome digests @@ -51,16 +81,16 @@ This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm ### Fixed: - Saving of big file size (changed to bigint db column type) -### [0.14.2] - 2026-01-21 +## [0.14.2] - 2026-01-21 ### Added: - Added method that fetches available reference genomes -### [0.14.1] - 2025-12-22 +## [0.14.1] - 2025-12-22 ### Fixed: - Fixed hybrid search reindexing - Updated limits in reindexing -### [0.14.0] - 2025-12-18 +## [0.14.0] - 2025-12-18 ### Fixed: - Insertion of tokenized files @@ -73,11 +103,11 @@ This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm ### Added: - Added hybrid semantic search.(dense + sparse search) -### [0.13.0] - 2025-11-24 +## [0.13.0] - 2025-11-24 ### Added: - Conversion of bedfile to umap from predefined model -### [0.12.0] - 2025-09-11 +## [0.12.0] - 2025-09-11 ### Added: - New qdrant semantic search - Added more plots to bedbase summary page @@ -91,12 +121,12 @@ This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm ### Fixed: - Issues in bedfile update method -### [0.11.4] - 2025-06-01 +## [0.11.4] - 2025-06-01 ### Fixed: - SQL search -### [0.11.3] - 2025-05-27 +## [0.11.3] - 2025-05-27 ### Fixed: - Usage tracker - Order of comprehensive stats @@ -106,11 +136,11 @@ This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm - Concise option in stats method -### [0.11.2] - 2025-06-22 +## [0.11.2] - 2025-06-22 ### Added: - Statistics about bed files grouped by organism -### [0.11.1] - 2025-05-22 +## [0.11.1] - 2025-05-22 ### Fixed: - Bedbuncher bug diff --git a/pyproject.toml b/pyproject.toml index fba05bdb..ac341771 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "bbconf" -version = "0.14.12" +version = "0.15.0" description = "Configuration and data management tool for BEDbase" readme = "README.md" license = "BSD-2-Clause" @@ -27,7 +27,6 @@ dependencies = [ "pydantic >= 2.9.0", "botocore >= 1.34.0, < 1.36.0", "boto3 >= 1.34.54, < 1.36.0", - "pephubclient >= 0.4.5", "sqlalchemy_schemadisplay", "zarr < 3.0.0", "pyyaml >= 6.0.1", @@ -37,6 +36,8 @@ dependencies = [ "umap-learn >= 0.5.8", "qdrant_client >= 1.16.1", "setuptools < 70.0.0", + "cachetools >= 4.2.4", + "alembic >= 1.19.1", ] [project.urls] @@ -74,3 +75,6 @@ exclude = ["manual_testing.py"] [tool.ruff.lint.isort] known-first-party = ["bbconf"] + +[tool.ruff.lint.per-file-ignores] +"bbconf/alembic/env.py" = ["E402"] diff --git a/tests/config_test.yaml b/tests/config_test.yaml index ef8069eb..41c866c8 100644 --- a/tests/config_test.yaml +++ b/tests/config_test.yaml @@ -17,10 +17,6 @@ qdrant: s3: bucket: bedbase endpoint_url: "None" -phc: - namespace: bedbase - name: bedbase - tag: test access_methods: http: type: "https" diff --git a/tests/conftest.py b/tests/conftest.py index 85d0b01b..e0ebba4b 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -114,7 +114,6 @@ def example_dict(): files=files, classification=classification, upload_qdrant=False, - upload_pephub=False, upload_s3=True, local_path=DATA_PATH, overwrite=False, @@ -125,11 +124,3 @@ def example_dict(): @pytest.fixture def load_test_data(): get_bbagent().config.db_engine() - - -@pytest.fixture() -def mocked_phc(mocker): - mocker.patch( - "pephubclient.modules.sample.PEPHubSample.get", - return_value={"sample_name": BED_TEST_ID, "other_metadata": "other_metadata_1"}, - ) diff --git a/tests/test_analysis_files.py b/tests/test_analysis_files.py new file mode 100644 index 00000000..ad39674c --- /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) diff --git a/tests/test_bedfile.py b/tests/test_bedfile.py index c6687574..db5cadf8 100644 --- a/tests/test_bedfile.py +++ b/tests/test_bedfile.py @@ -8,7 +8,7 @@ from bbconf.exceptions import BedFIleExistsError, BEDFileNotFoundError from .conftest import SERVICE_UNAVAILABLE, get_bbagent -from .utils import BED_TEST_ID, ContextManagerDBTesting +from .utils import BED_TEST_ID, BEDSET_TEST_ID, ContextManagerDBTesting @pytest.mark.skipif(SERVICE_UNAVAILABLE, reason="Database is not available") @@ -52,16 +52,13 @@ def test_add_nofail(self, bbagent_obj, example_dict, mocker): bbagent_obj.bed.add(**example_dict) assert bbagent_obj.bed.exists(example_dict["identifier"]) - def test_get_all(self, bbagent_obj, mocked_phc): + def test_get_all(self, bbagent_obj): with ContextManagerDBTesting(config=bbagent_obj.config, add_data=True): return_result = bbagent_obj.bed.get(BED_TEST_ID, full=True) assert return_result is not None assert return_result.files is not None assert return_result.plots is not None - # TODO: PEPhub is disabled - # assert return_result.raw_metadata is not None - assert return_result.genome_alias == "hg38" assert return_result.stats.number_of_regions == 1 @@ -69,6 +66,16 @@ def test_get_all(self, bbagent_obj, mocked_phc): assert return_result.plots.chrombins is not None assert return_result.license_id == DEFAULT_LICENSE + def test_get_all_bedsets_bedfile_count(self, bbagent_obj): + with ContextManagerDBTesting( + config=bbagent_obj.config, add_data=True, bedset=True + ): + return_result = bbagent_obj.bed.get(BED_TEST_ID, full=True) + + assert len(return_result.bedsets) == 1 + assert return_result.bedsets[0].id == BEDSET_TEST_ID + assert return_result.bedsets[0].bedfile_count == 1 + def test_get_all_not_found(self, bbagent_obj): with ContextManagerDBTesting(config=bbagent_obj.config, add_data=True): return_result = bbagent_obj.bed.get(BED_TEST_ID, full=False) @@ -76,22 +83,11 @@ def test_get_all_not_found(self, bbagent_obj): assert return_result is not None assert return_result.files is None assert return_result.plots is None - assert return_result.raw_metadata is None assert return_result.stats is None assert return_result.genome_alias == "hg38" assert return_result.id == BED_TEST_ID - @pytest.mark.skip( - "Skipped, because PHC is disabled" - ) # TODO: should we disable PHC everywhere? - def test_get_raw_metadata(self, bbagent_obj, mocked_phc): - with ContextManagerDBTesting(config=bbagent_obj.config, add_data=True): - return_result = bbagent_obj.bed.get_raw_metadata(BED_TEST_ID) - - assert return_result is not None - assert return_result.sample_name == BED_TEST_ID - def test_get_stats(self, bbagent_obj): with ContextManagerDBTesting(config=bbagent_obj.config, add_data=True): return_result = bbagent_obj.bed.get_stats(BED_TEST_ID) diff --git a/tests/test_bedset.py b/tests/test_bedset.py index ecc8948c..1f7977fd 100644 --- a/tests/test_bedset.py +++ b/tests/test_bedset.py @@ -53,7 +53,6 @@ def test_crate_bedset_all(self, bbagent_obj, mocker): }, statistics=True, upload_s3=True, - upload_pephub=False, no_fail=True, ) with Session(bbagent_obj.config.db_engine.engine) as session: @@ -61,6 +60,7 @@ def test_crate_bedset_all(self, bbagent_obj, mocker): assert result is not None assert result.name == "test_name" assert len([k for k in result.files]) == 1 + assert result.bedfile_count == 1 def test_get_metadata_full(self, bbagent_obj): with ContextManagerDBTesting( @@ -73,6 +73,7 @@ def test_get_metadata_full(self, bbagent_obj): assert result.statistics.sd is not None assert result.statistics.mean is not None assert result.plots is not None + assert result.bedfile_count == 1 def test_get_metadata_not_full(self, bbagent_obj): with ContextManagerDBTesting( @@ -84,6 +85,7 @@ def test_get_metadata_not_full(self, bbagent_obj): assert result.md5sum == "bbad0000000000000000000000000000" assert result.statistics is None assert result.plots is None + assert result.bedfile_count == 1 def test_get_not_found(self, bbagent_obj): with ContextManagerDBTesting( @@ -128,6 +130,10 @@ def test_get_bedset_list(self, bbagent_obj): assert result.offset == 0 assert len(result.results) == 1 assert result.results[0].id == BEDSET_TEST_ID + # bed_ids is intentionally left unpopulated in list results to + # avoid lazy-loading full bedfile membership for every row + assert result.results[0].bed_ids is None + assert result.results[0].bedfile_count == 1 def test_get_bedset_list_offset(self, bbagent_obj): with ContextManagerDBTesting( diff --git a/tests/test_common.py b/tests/test_common.py index 76d6a7ec..6baf7568 100644 --- a/tests/test_common.py +++ b/tests/test_common.py @@ -20,6 +20,16 @@ def test_get_stats(bbagent_obj): assert return_result.genomes_number == 1 +@pytest.mark.skipif(SERVICE_UNAVAILABLE, reason="Database is not available") +def test_get_detailed_stats(bbagent_obj): + with ContextManagerDBTesting(config=bbagent_obj.config, add_data=True, bedset=True): + return_result = bbagent_obj.get_detailed_stats() + + assert return_result + assert return_result.number_of_regions.mean == 1 + assert return_result.mean_region_width.mean == 3 + + @pytest.mark.skipif(SERVICE_UNAVAILABLE, reason="Database is not available") def test_get_licenses(bbagent_obj): return_result = bbagent_obj.list_of_licenses diff --git a/tests/test_universes.py b/tests/test_universes.py index 366b3770..15b8a45e 100644 --- a/tests/test_universes.py +++ b/tests/test_universes.py @@ -65,6 +65,6 @@ def test_add_get_tokenized(self, bbagent_obj, mocker): assert zarr_mock.called assert f"s3://bedbase/{saved_path}" == zarr_path - def test_get_tokenized(self, bbagent_obj, mocked_phc): + def test_get_tokenized(self, bbagent_obj): # how to test it? ... diff --git a/tests/utils.py b/tests/utils.py index fd586efa..2714488a 100644 --- a/tests/utils.py +++ b/tests/utils.py @@ -138,6 +138,7 @@ def _add_bedset_data(self): bedset_standard_deviation=stats, md5sum="bbad0000000000000000000000000000", processed=False, + bedfile_count=1, ) new_bed_bedset = BedFileBedSetRelation( bedfile_id=BED_TEST_ID,