bbconf API

BedBaseAgent is the entry point. Its bed, bedset, objects, snapshot, and analysis_files properties return the agents documented below.

`BedBaseAgent`

class BedBaseAgent(config: Path | str, init_ml: bool = True)

Initialize connection to the pep_db database. You can use the basic connection parameters or libpq connection string.

Parameters:

config (Path | str): Path to the configuration file.
init_ml (bool): Initialize ML models for search (default: True). (default: True)

Attributes

config

Properties

`bed`

@property
def bed -> BedAgentBedFile

`bedset`

@property
def bedset -> BedAgentBedSet

`objects`

@property
def objects -> BBObjects

`snapshot`

@property
def snapshot -> BedAgentSnapshot

`analysis_files`

@property
def analysis_files -> BedAgentAnalysisFile

`list_of_licenses`

@property
def list_of_licenses -> list[str]

Get list of licenses from the database.

Returns:

list[str]: List of licenses.

Methods

`get_stats`

def get_stats() -> 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:

StatsReturn: Statistics.

`get_detailed_stats`

def get_detailed_stats(concise: bool = False) -> FileStats

Get comprehensive statistics for all bed files.

Parameters:

concise (bool): If True, return only top 20 items for each category. (default: False)

Returns:

FileStats: FileStats object containing detailed statistics.

`get_detailed_usage`

def get_detailed_usage() -> UsageStats

Get detailed usage statistics for the bedbase platform. This method will only return top 20 items for each category.

Returns:

UsageStats: UsageStats object containing detailed usage statistics.

`get_list_genomes`

def get_list_genomes() -> list[str]

Get list of genomes from the database.

Returns:

list[str]: List of genomes.

`get_list_assays`

def get_list_assays() -> list[str]

Get list of assays from the database.

Returns:

list[str]: List of assays.

`add_usage`

def add_usage(stats: UsageModel) -> None

`bed_files_info`

def bed_files_info() -> AllFilesInfo

Get information about all bed files in bedbase.

Returns:

AllFilesInfo: AllFilesInfo containing total count and list of file info objects.

`get_reference_genomes`

def get_reference_genomes() -> dict[str, str]

Get mapping of genome aliases to reference genome names.

Returns:

dict[str, str]: Dict mapping genome_alias to reference_genome_name.

`BedAgentBedFile`

class BedAgentBedFile(config: BedBaseConfig, bbagent_obj=None)

Class that represents a BED file in the Database.

Provides methods to add, delete, get BED files and metadata from the database.

Initialize BedAgentBedFile.

Parameters:

config (BedBaseConfig): Config object with database and qdrant engine and credentials.
bbagent_obj: BedBaseAgent object (Parent object). (default: None)

Attributes

config
bb_agent

Methods

`get`

def get(identifier: str, full: bool = False) -> BedMetadataAll

Get file metadata by identifier.

Parameters:

identifier (str): Bed file identifier.
full (bool): If True, return full metadata, including statistics and files. (default: False)

Returns:

BedMetadataAll: BED file metadata.

`get_stats`

def get_stats(identifier: str) -> BedStatsModel

Get file statistics by identifier.

Parameters:

identifier (str): Bed file identifier.

Returns:

BedStatsModel: Project statistics as BedStats object.

`get_plots`

def get_plots(identifier: str) -> BedPlots

Get file plots by identifier.

Parameters:

identifier (str): Bed file identifier.

Returns:

BedPlots: Project plots.

`get_neighbours`

def get_neighbours(identifier: str, limit: int = 10, offset: int = 0) -> BedListSearchResult

Get nearest neighbours of bed file from qdrant.

Parameters:

identifier (str): Bed file identifier.
limit (int): Number of results to return. (default: 10)
offset (int): Offset to start from. (default: 0)

Returns:

BedListSearchResult: List of nearest neighbours.

`get_files`

def get_files(identifier: str) -> BedFiles

Get file files by identifier.

Parameters:

identifier (str): Bed file identifier.

Returns:

BedFiles: Project files.

`get_classification`

def get_classification(identifier: str) -> BedClassification

Get file classification by identifier.

Parameters:

identifier (str): Bed file identifier.

Returns:

BedClassification: Project classification.

`get_objects`

def get_objects(identifier: str) -> dict[str, FileModel]

Get all object related to bedfile.

Parameters:

identifier (str): Bed file identifier.

Returns:

dict[str, FileModel]: Project objects dict.

`get_embedding`

def get_embedding(identifier: str) -> BedEmbeddingResult

Get bed file embedding of bed file from qdrant.

Parameters:

identifier (str): Bed file identifier.

Returns:

BedEmbeddingResult: Bed file embedding.

`get_ids_list`

def get_ids_list(limit: int = 100, offset: int = 0, genome: str = None, bed_compliance: str = None) -> BedListResult

Get list of bed file identifiers.

Parameters:

limit (int): Number of results to return. (default: 100)
offset (int): Offset to start from. (default: 0)
genome (str): Filter by genome. (default: None)
bed_compliance (str): Filter by bed type. e.g. 'bed6+4'. (default: None)

Returns:

BedListResult: List of bed file identifiers.

`get_reference_validation`

def get_reference_validation(identifier: str) -> RefGenValidReturnModel

Get results of reference genome validation for the bed file.

Parameters:

identifier (str): Bed file identifier.

Returns:

RefGenValidReturnModel: Reference genome validation results.

`add`

def add(identifier: str, stats: dict, metadata: dict = None, plots: dict = None, files: dict = None, classification: dict = None, ref_validation: dict[str, BaseModel] | None = None, license_id: str = DEFAULT_LICENSE, upload_qdrant: bool = False, upload_s3: bool = False, local_path: str = None, overwrite: bool = False, nofail: bool = False, processed: bool = True) -> None

Add bed file to the database.

Parameters:

identifier (str): Bed file identifier.
stats (dict): Bed file results {statistics, plots, files, metadata}.
metadata (dict): Bed file metadata. (default: None)
plots (dict): Bed file plots. (default: None)
files (dict): Bed file files. (default: None)
classification (dict): Bed file classification. (default: None)
ref_validation (dict[str, BaseModel] | None): Reference validation data. RefGenValidModel. (default: None)
license_id (str): Bed file license id (default: 'DUO:0000042'). Full list of licenses: https://raw.githubusercontent.com/EBISPOT/DUO/master/duo.csv (default: DEFAULT_LICENSE)
upload_qdrant (bool): Add bed file to qdrant indexes. (default: False)
upload_s3 (bool): Upload files to s3. (default: False)
local_path (str): Local path to the output files. (default: None)
overwrite (bool): Overwrite bed file if it already exists. (default: False)
nofail (bool): Do not raise an error for error in s3/qdrant or record exists and not overwrite. (default: False)
processed (bool): True if bedfile was processed and statistics and plots were calculated. (default: True)

Returns:

None: None.

`update`

def update(identifier: str, stats: dict | None = None, metadata: dict | None = None, plots: dict | None = None, files: dict | None = None, classification: dict | None = None, ref_validation: dict[str, BaseModel] | None = None, license_id: str = DEFAULT_LICENSE, upload_qdrant: bool = False, upload_s3: bool = True, local_path: str = None, overwrite: bool = False, nofail: bool = False, processed: bool = False) -> None

Update bed file to the database.

Parameters:

identifier (str): Bed file identifier.
stats (dict | None): Bed file results {statistics, plots, files, metadata}. (default: None)
metadata (dict | None): Bed file metadata. (default: None)
plots (dict | None): Bed file plots. (default: None)
files (dict | None): Bed file files. (default: None)
classification (dict | None): Bed file classification. (default: None)
ref_validation (dict[str, BaseModel] | None): Reference validation data. RefGenValidModel. (default: None)
license_id (str): Bed file license id (default: 'DUO:0000042'). (default: DEFAULT_LICENSE)
upload_qdrant (bool): Add bed file to qdrant indexes. (default: False)
upload_s3 (bool): Upload files to s3. (default: True)
local_path (str): Local path to the output files. (default: None)
overwrite (bool): Overwrite bed file if it already exists. (default: False)
nofail (bool): Do not raise an error for error in s3/qdrant or record exists and not overwrite. (default: False)
processed (bool): True if bedfile was processed and statistics and plots were calculated. (default: False)

Returns:

None: None.

`delete`

def delete(identifier: str) -> None

Delete bed file from the database.

Parameters:

identifier (str): Bed file identifier.

Returns:

None: None.

`upload_file_qdrant`

def upload_file_qdrant(bed_id: str, bed_file: str | GRegionSet, payload: dict = None) -> None

Convert bed file to vector and add it to qdrant database.

Warning: only hg38 genome can be added to qdrant!

Parameters:

bed_id (str): Bed file id.
bed_file (str | GRegionSet): Path to the bed file, or RegionSet object.
payload (dict): Additional metadata to store alongside vectors. (default: None)

Returns:

None: None.

`text_to_bed_search`

def text_to_bed_search(query: str, limit: int = 10, offset: int = 0, with_metadata: bool = True) -> BedListSearchResult

Search for bed files by text query in qdrant database. This is bivec_search.

Parameters:

query (str): Text query.
limit (int): Number of results to return. (default: 10)
offset (int): Offset to start from. (default: 0)
with_metadata (bool): If True, will return metadata for each result. (default: True)

Returns:

BedListSearchResult: List of bed file metadata.

`bed_to_bed_search`

def bed_to_bed_search(region_set: GRegionSet, limit: int = 10, offset: int = 0) -> BedListSearchResult

Search for bed files by using region set in qdrant database.

Parameters:

region_set (GRegionSet): RegionSet object to search for (bed file).
limit (int): Number of results to return. (default: 10)
offset (int): Offset to start from. (default: 0)

Returns:

BedListSearchResult: BedListSetResults.

`sql_search`

def sql_search(query: str, genome: str = None, assay: str = None, limit: int = 10, offset: int = 0) -> BedListSearchResult

Search for bed files by using sql exact search. This search will search files by id, name, and description.

Parameters:

query (str): Text query.
genome (str): Genome alias to filter results. (default: None)
assay (str): Filter by assay type. (default: None)
limit (int): Number of results to return. (default: 10)
offset (int): Offset to start from. (default: 0)

Returns:

BedListSearchResult: List of bed file metadata.

`reindex_qdrant`

def reindex_qdrant(batch: int = 100, purge: bool = False) -> None

Re-upload all files to qdrant. Warning: only hg38 genome can be added to qdrant!

If you want to fully reindex/reupload to qdrant, first delete collection and create new one. Upload all files to qdrant.

Parameters:

batch (int): Number of files to upload in one batch. (default: 100)

`delete_qdrant_point`

def delete_qdrant_point(identifier: str) -> None

Delete bed file from qdrant.

Parameters:

identifier (str): Bed file identifier.

Returns:

None: None.

`exists`

def exists(identifier: str) -> bool

Check if bed file exists in the database.

Parameters:

identifier (str): Bed file identifier.

Returns:

bool: True if bed file exists, False otherwise.

`exists_universe`

def exists_universe(identifier: str) -> bool

Check if universe exists in the database.

Parameters:

identifier (str): Universe identifier.

Returns:

bool: True if universe exists, False otherwise.

`add_universe`

def add_universe(bedfile_id: str, bedset_id: str = None, construct_method: str = None) -> str

Add universe to the database.

Parameters:

bedfile_id (str): Bed file identifier.
bedset_id (str): Bedset identifier. (default: None)
construct_method (str): Method used to construct the universe. (default: None)

Returns:

str: Universe identifier.

`delete_universe`

def delete_universe(identifier: str) -> None

Delete universe from the database.

Parameters:

identifier (str): Universe identifier.

Returns:

None: None.

`add_tokenized`

def add_tokenized(bed_id: str, universe_id: str, token_vector: list, overwrite: bool = False) -> str

Add tokenized bed file to the database.

Parameters:

bed_id (str): Bed file identifier.
universe_id (str): Universe identifier.
token_vector (list): List of tokens.
overwrite (bool): Overwrite tokenized file if it already exists. (default: False)

Returns:

str: Token path.

`get_tokenized`

def get_tokenized(bed_id: str, universe_id: str) -> TokenizedBedResponse

Get zarr file from the database.

Parameters:

bed_id (str): Bed file identifier.
universe_id (str): Universe identifier.

Returns:

TokenizedBedResponse: Zarr path.

`delete_tokenized`

def delete_tokenized(bed_id: str, universe_id: str) -> None

Delete tokenized bed file from the database.

Parameters:

bed_id (str): Bed file identifier.
universe_id (str): Universe identifier.

Returns:

None: None.

`exist_tokenized`

def exist_tokenized(bed_id: str, universe_id: str) -> bool

Check if tokenized bed file exists in the database.

Parameters:

bed_id (str): Bed file identifier.
universe_id (str): Universe identifier.

Returns:

bool: True if tokenized bed file exists, False otherwise.

`get_tokenized_link`

def get_tokenized_link(bed_id: str, universe_id: str) -> TokenizedPathResponse

Get tokenized link to tokenized file.

Parameters:

bed_id (str): Bed file identifier.
universe_id (str): Universe identifier.

Returns:

TokenizedPathResponse: Token link.

Raises:

TokenizeFileNotExistError: If the tokenized file does not exist.

`get_missing_plots`

def get_missing_plots(plot_name: str, limit: int = 1000, offset: int = 0) -> list[str]

Get list of bed files that are missing plot.

Parameters:

plot_name (str): Plot name.
limit (int): Number of results to return. (default: 1000)
offset (int): Offset to start from. (default: 0)

Returns:

list[str]: List of bed file identifiers.

`get_missing_stats`

def get_missing_stats(limit: int = 1000, offset: int = 0) -> list[str]

Get list of bed files that are missing statistics.

Parameters:

limit (int): Number of results to return. (default: 1000)
offset (int): Offset to start from. (default: 0)

Returns:

list[str]: List of bed file identifiers.

`get_missing_files`

def get_missing_files(limit: int = 1000, offset: int = 0) -> list[str]

Get list of bed files that are missing files (bigBed files).

Parameters:

limit (int): Number of results to return. (default: 1000)
offset (int): Offset to start from. (default: 0)

Returns:

list[str]: List of bed file identifiers.

`get_unprocessed`

def get_unprocessed(limit: int = 1000, offset: int = 0, genome: str | list | None = None) -> BedListResult

Get bed files that are not processed.

Parameters:

limit (int): Number of results to return. (default: 1000)
offset (int): Offset to start from. (default: 0)
genome (str | list | None): Genome alias or list of genome aliases to filter by. e.g. "hg38" or ["hg38", "mm10"]. By default None, which means no filtering by genome. (default: None)

Returns:

BedListResult: List of bed file identifiers.

`reindex_hybrid_search`

def reindex_hybrid_search(batch: int = 1000, purge: bool = False) -> None

Reindex all bed files for semantic database.

Parameters:

batch (int): Number of files to upload in one batch. (default: 1000)
purge (bool): Resets indexed in database for all files to False. (default: False)

Returns:

None: None.

`hybrid_search`

def hybrid_search(query: str = 'liver', genome_alias: str = '', assay: str = '', limit: int = 100, offset: int = 0, with_metadata: bool = True) -> BedListSearchResult

Run semantic search for bed files using qdrant. This is not bivec search, but usual qdrant search with sparse and dense embeddings.

Parameters:

query (str): Text query to search for. (default: 'liver')
genome_alias (str): Genome alias to filter results. (default: '')
assay (str): Filter by assay type. (default: '')
limit (int): Number of results to return. (default: 100)
offset (int): Offset to start from. (default: 0)
with_metadata (bool): If True, metadata will be returned in the results. Default is True. (default: True)

Returns:

BedListSearchResult: List of bed file metadata.

`search_external_file`

def search_external_file(source: str, accession: str) -> BedListSearchResult

Search for bed files by external source and accession number. e.g. source='geo', accession='GSE12345'.

Parameters:

source (str): External source, e.g. 'geo' or 'encode'.
accession (str): Accession number, e.g. 'GSE12345' or 'ENCSR12345'.

Returns:

BedListSearchResult: List of bed file metadata.

`BedAgentBedSet`

class BedAgentBedSet(config: BedBaseConfig)

Class that represents Bedset in Database.

This class has method to add, delete, get files and metadata from the database.

Initialize BedAgentBedSet.

Parameters:

config (BedBaseConfig): Config object.

Attributes

config

Methods

`get`

def get(identifier: str, full: bool = False) -> BedSetMetadata

Get file metadata by identifier.

Parameters:

identifier (str): Bed file identifier.
full (bool): Return full record with stats, plots, files and metadata. (default: False)

Returns:

BedSetMetadata: BED file metadata.

`get_plots`

def get_plots(identifier: str) -> BedSetPlots

Get plots for bedset by identifier.

Parameters:

identifier (str): Bedset identifier.

Returns:

BedSetPlots: Bedset plots.

`get_objects`

def get_objects(identifier: str) -> dict[str, FileModel]

Get objects for bedset by identifier.

Parameters:

identifier (str): Bedset identifier.

Returns:

dict[str, FileModel]: Bedset objects.

`get_statistics`

def get_statistics(identifier: str) -> BedSetStats

Get statistics for bedset by identifier.

Parameters:

identifier (str): Bedset identifier.

Returns:

BedSetStats: Bedset statistics.

`get_bedset_pep`

def get_bedset_pep(identifier: str) -> dict

Create pep file for a bedset.

Parameters:

identifier (str): Bedset identifier.

Returns:

dict: Pep dict.

`get_track_hub_file`

def get_track_hub_file(identifier: str) -> str

Get track hub file for bedset.

Parameters:

identifier (str): Bedset identifier.

Returns:

str: Track hub file.

`create`

def create(identifier: str, name: str, bedid_list: list[str], description: str | None = None, statistics: bool = False, annotation: dict | None = None, plots: dict | None = None, upload_s3: bool = False, local_path: str = '', no_fail: bool = False, overwrite: bool = False, processed: bool = True) -> None

Create bedset in the database.

Parameters:

identifier (str): Bedset identifier.
name (str): Bedset name.
description (str | None): Bedset description. (default: None)
bedid_list (list[str]): List of bed file identifiers.
statistics (bool): Calculate statistics for bedset. (default: False)
annotation (dict | None): Bedset annotation (author, source). (default: None)
plots (dict | None): Dictionary with plots. (default: None)
upload_s3 (bool): Upload bedset to s3. (default: False)
local_path (str): Local path to the output files. (default: '')
no_fail (bool): Do not raise an error if bedset already exists. (default: False)
overwrite (bool): Overwrite the record in the database. (default: False)
processed (bool): Flag to indicate that bedset is processed. [Default: True]. (default: True)

Returns:

None: None.

`get_ids_list`

def get_ids_list(query: str | None = None, limit: int = 10, offset: int = 0) -> BedSetListResult

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.

Parameters:

query (str | None): Search query. (default: None)
limit (int): Limit of results. (default: 10)
offset (int): Offset of results. (default: 0)

Returns:

BedSetListResult: List of bedsets.

`get_bedset_bedfiles`

def get_bedset_bedfiles(identifier: str) -> BedSetBedFiles

Get list of bedfiles in bedset.

Parameters:

identifier (str): Bedset identifier.

Returns:

BedSetBedFiles: List of bedfiles.

`delete`

def delete(identifier: str) -> None

Delete bed file from the database.

Parameters:

identifier (str): Bedset identifier.

Returns:

None: None.

`exists`

def exists(identifier: str) -> bool

Check if bedset exists in the database.

Parameters:

identifier (str): Bedset identifier.

Returns:

bool: True if bedset exists, False otherwise.

`get_unprocessed`

def get_unprocessed(limit: int = 100, offset: int = 0) -> BedSetListResult

Get unprocessed bedset from the database.

Parameters:

limit (int): Limit of results. (default: 100)
offset (int): Offset of results. (default: 0)

Returns:

BedSetListResult: Bedset metadata.

`add_bedfile`

def add_bedfile(identifier: str, bedfile: str) -> None

`delete_bedfile`

def delete_bedfile(identifier: str, bedfile: str) -> None

`BBObjects`

class BBObjects(config: BedBaseConfig)

Initialize BBObjects.

Parameters:

config (BedBaseConfig): Config object.

Attributes

config
bed
bedset

Methods

`get_thumbnail_uri`

def get_thumbnail_uri(record_type: Literal['bed', 'bedset'], record_id: str, result_id: str, access_id: str = 'http') -> str

Create URL to access a bed- or bedset-associated thumbnail.

Parameters:

record_type (Literal['bed', 'bedset']): Table name ["bed", "bedset"].
record_id (str): Record identifier.
result_id (str): Column name (result name).
access_id (str): Access id (e.g. http, s3, etc.). (default: 'http')

Returns:

str: String with thumbnail.

`get_object_uri`

def get_object_uri(record_type: Literal['bed', 'bedset'], record_id: str, result_id: str, access_id: str) -> str

Create URL to access a bed- or bedset-associated file.

Parameters:

record_type (Literal['bed', 'bedset']): Table name ["bed", "bedset"].
record_id (str): Record identifier.
result_id (str): Column name (result name).
access_id (str): Access id (e.g. http, s3, etc.).

Returns:

str: URI string for the object.

`get_drs_metadata`

def get_drs_metadata(record_type: Literal['bed', 'bedset'], record_id: str, result_id: str, base_uri: str) -> DRSModel

Get DRS metadata for a bed- or bedset-associated file.

Parameters:

record_type (Literal['bed', 'bedset']): Bed or bedset.
record_id (str): Record identifier.
result_id (str): Name of the result file to get metadata for.
base_uri (str): Base uri to use for the self_uri field (server hostname of DRS broker).

Returns:

DRSModel: DRS metadata.

`construct_drs_metadata`

def construct_drs_metadata(base_uri: str, object_id: str, record_metadata: FileModel, created_time: datetime.datetime | None = None, modified_time: datetime.datetime | None = None) -> DRSModel

Construct DRS metadata object.

Parameters:

base_uri (str): Base uri to use for the self_uri field (server hostname of DRS broker).
object_id (str): Record identifier.
record_metadata (FileModel): Metadata of the record.
created_time (datetime.datetime | None): Time of creation. (default: None)
modified_time (datetime.datetime | None): Time of last modification. (default: None)

Returns:

DRSModel: DRS metadata.

`BedAgentSnapshot`

class BedAgentSnapshot(config: BedBaseConfig)

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.

Initialize BedAgentSnapshot.

Parameters:

config (BedBaseConfig): Config object.

Attributes

config

Methods

`add`

def add(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.

Parameters:

artifacts (BedSnapshotArtifact | list[BedSnapshotArtifact]): One artifact or a list of them. Each carries the local path to upload plus its file_type and file metadata.
creation_date (datetime | None): Build date recorded on every row (defaults to now, UTC). (default: None)

Returns:

BedSnapshotListResult: The created snapshot rows.

`delete`

def delete(id: int, remove_s3: bool = True) -> None

Delete a snapshot index row.

Parameters:

id (int): Primary key of the snapshot row.
remove_s3 (bool): Also delete the underlying S3 object. (default: True)

Returns:

None: None.

Raises:

SnapshotNotFoundError: If no row with this id exists.

`list`

def list(file_type: str | None = None, limit: int | None = 100, offset: int = 0) -> BedSnapshotListResult

List all snapshot index rows in the database, newest first.

Parameters:

file_type (str | None): Optional filter on file type. (default: None)
limit (int | None): Maximum number of rows to return. None returns all rows. (default: 100)
offset (int): Number of rows to skip. (default: 0)

Returns:

BedSnapshotListResult: List of snapshots and the total matching count.

`get_by_filename`

def get_by_filename(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.

Parameters:

filename (str): The bare file name, e.g. bedbase_metadata_2026_08_03.parquet.

Returns:

BedSnapshotResult: The matching snapshot row.

Raises:

SnapshotNotFoundError: If no row matches.

`delete_by_checksum`

def delete_by_checksum(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.

Parameters:

checksum (str): SHA256 checksum of the snapshot file.
remove_s3 (bool): Also delete the underlying S3 object(s). (default: True)

Returns:

None: None.

Raises:

SnapshotNotFoundError: If no row matches the checksum.

`get`

def get(id: int) -> BedSnapshotResult

Get a single snapshot index row by id.

Parameters:

id (int): Primary key of the snapshot row.

Returns:

BedSnapshotResult: The snapshot row.

Raises:

SnapshotNotFoundError: If no row with this id exists.

`BedAgentAnalysisFile`

class BedAgentAnalysisFile(config: BedBaseConfig)

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.

Initialize BedAgentAnalysisFile.

Parameters:

config (BedBaseConfig): Config object.

Attributes

config

Methods

`add`

def add(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.

Parameters:

artifacts (AnalysisFileArtifact | list[AnalysisFileArtifact]): One artifact or a list of them. Each carries the local path to upload plus its name and file metadata.
creation_date (datetime | None): Upload date recorded on every row (defaults to now, UTC). (default: None)

Returns:

AnalysisFileListResult: The created analysis-file rows.

`delete`

def delete(id: int, remove_s3: bool = True) -> None

Delete an analysis-file index row.

Parameters:

id (int): Primary key of the analysis-file row.
remove_s3 (bool): Also delete the underlying S3 object. (default: True)

Returns:

None: None.

Raises:

AnalysisFileNotFoundError: If no row with this id exists.

`list`

def list(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.

Parameters:

file_type (str | None): Optional filter on file type. (default: None)
genome (str | None): Optional filter on genome/assembly. (default: None)
tag (str | None): Optional filter; keep only rows whose tags contain this tag. (default: None)
limit (int | None): Maximum number of rows to return. None returns all rows. (default: 100)
offset (int): Number of rows to skip. (default: 0)

Returns:

AnalysisFileListResult: List of analysis files and the total matching count.

`get`

def get(id: int) -> AnalysisFileResult

Get a single analysis-file index row by id.

Parameters:

id (int): Primary key of the analysis-file row.

Returns:

AnalysisFileResult: The analysis-file row.

Raises:

AnalysisFileNotFoundError: If no row with this id exists.

`get_by_name`

def get_by_name(name: str, genome: str | None = None) -> AnalysisFileResult

Resolve an analysis file by its logical name (newest matching row).

Parameters:

name (str): Logical name/key, e.g. openSignalMatrix.
genome (str | None): Optional genome/assembly to disambiguate, e.g. hg38. (default: None)

Returns:

AnalysisFileResult: The newest matching analysis-file row.

Raises:

AnalysisFileNotFoundError: If no row matches.

`get_by_filename`

def get_by_filename(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.

Parameters:

filename (str): The bare file name, e.g. openSignalMatrix_hg38.txt.gz.

Returns:

AnalysisFileResult: The matching analysis-file row.

Raises:

AnalysisFileNotFoundError: If no row matches.

`delete_by_checksum`

def delete_by_checksum(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.

Parameters:

checksum (str): SHA256 checksum of the analysis file.
remove_s3 (bool): Also delete the underlying S3 object(s). (default: True)

Returns:

None: None.

Raises:

AnalysisFileNotFoundError: If no row matches the checksum.