Warning

This document is for an old release of Galaxy. You can alternatively view this page in the latest release if it exists or view the top of the latest release's documentation.

Source code for galaxy.managers.genomes

from typing import (
    Any,
    List,
    TYPE_CHECKING,
)

from sqlalchemy import (
    func,
    text,
)

from galaxy import model as m
from galaxy.exceptions import (
    ReferenceDataError,
    RequestParameterInvalidException,
)
from galaxy.managers.context import ProvidesUserContext
from galaxy.structured_app import StructuredApp
from .base import raise_filter_err

if TYPE_CHECKING:
    from galaxy.managers.base import OrmFilterParsersType


[docs]class GenomesManager:
[docs] def __init__(self, app: StructuredApp): self._app = app self.genomes = app.genomes
[docs] def get_dbkeys(self, user: m.User, chrom_info: bool) -> List[List[str]]: return self.genomes.get_dbkeys(user, chrom_info)
[docs] def is_registered_dbkey(self, dbkey: str, user: m.User) -> bool: dbkeys = self.get_dbkeys(user, chrom_info=False) for _, key in dbkeys: if dbkey == key: return True return False
[docs] def get_genome( self, trans: ProvidesUserContext, id: str, num: int, chrom: str, low: int, high: int, reference: bool ) -> Any: if reference: region = self.genomes.reference(trans, dbkey=id, chrom=chrom, low=low, high=high) return {"dataset_type": "refseq", "data": region.sequence} else: return self.genomes.chroms(trans, dbkey=id, num=num, chrom=chrom, low=low)
[docs] def get_sequence(self, trans: ProvidesUserContext, id: str, chrom: str, low: int, high: int) -> Any: region = self.genomes.reference(trans, dbkey=id, chrom=chrom, low=low, high=high) return region.sequence
[docs] def get_indexes(self, id: str, index_type: str) -> Any: index_extensions = {"fasta_indexes": ".fai"} if index_type not in index_extensions: raise RequestParameterInvalidException(f"Invalid index type: {index_type}") tbl_entries = self._app.tool_data_tables.data_tables[index_type].data ext = index_extensions[index_type] index_filename = self._get_index_filename(id, tbl_entries, ext, index_type) try: with open(index_filename) as f: return f.read() except OSError: raise ReferenceDataError(f"Failed to load index file for {id}")
def _get_index_filename(self, id, tbl_entries, ext, index_type): try: paths = [x[-1] for x in tbl_entries if id in x] file_name = paths.pop() except TypeError: raise ReferenceDataError(f"Data tables not found for {index_type}") except IndexError: raise ReferenceDataError(f"Data tables not found for {index_type} for {id}") else: return f"{file_name}{ext}"
[docs]class GenomeFilterMixin: orm_filter_parsers: "OrmFilterParsersType" database_connection: str valid_ops = ("eq", "contains", "has")
[docs] def create_genome_filter(self, attr, op, val): def _create_genome_filter(model_class=None): if op not in GenomeFilterMixin.valid_ops: raise_filter_err(attr, op, val, "bad op in filter") if model_class is None: return True # Doesn't filter genome_build for collections if model_class.__name__ == "HistoryDatasetCollectionAssociation": return False # TODO: should use is_postgres(self.database_connection) in 23.2 if self.database_connection.startswith("postgres"): column = text("convert_from(metadata, 'UTF8')::json ->> 'dbkey'") else: column = func.json_extract(model_class.table.c._metadata, "$.dbkey") lower_val = val.lower() # Ignore case # dbkey can either be "hg38" or '["hg38"]', so we need to check both if op == "eq": cond = func.lower(column) == lower_val or func.lower(column) == f'["{lower_val}"]' else: cond = func.lower(column).contains(lower_val, autoescape=True) return cond return _create_genome_filter
def _add_parsers(self, database_connection: str): self.database_connection = database_connection self.orm_filter_parsers.update({"genome_build": self.create_genome_filter})