Source code for dae.genotype_storage.genotype_storage

from __future__ import annotations

import abc
import logging
from typing import Any, cast

from dae.genomic_resources.gene_models import GeneModels
from dae.genomic_resources.reference_genome import ReferenceGenome

logger = logging.getLogger(__file__)

[docs]class GenotypeStorage(abc.ABC): """Base class for genotype storages.""" def __init__(self, storage_config: dict[str, Any]): self.storage_config = \ self.validate_and_normalize_config(storage_config) self.storage_id = self.storage_config["id"] self.storage_type = cast(str, self.storage_config["storage_type"]) self._read_only = cast( bool, self.storage_config.get("read_only", False))
[docs] @classmethod def validate_and_normalize_config(cls, config: dict) -> dict: """Normalize and validate the genotype storage configuration. When validation passes returns the normalized and validated annotator configuration dict. When validation fails, raises ValueError. All genotype storage configurations are required to have: * "storage_type" - which storage type this configuration is used for; * "id" - the ID of the genotype storage instance that will be created. """ if config.get("id") is None: raise ValueError( f"genotype storage without ID; 'id' is required: {config}") if config.get("storage_type") is None: raise ValueError( f"genotype storage without type; 'storage_type' is required: " f"{config}") if config["storage_type"] not in cls.get_storage_types(): raise ValueError( f"storage configuration for <{config['storage_type']}> passed " f"to genotype storage class type <{cls.get_storage_types()}>") return config
[docs] def is_read_only(self) -> bool: return self._read_only
@property def read_only(self) -> bool: return self._read_only
[docs] @classmethod @abc.abstractmethod def get_storage_types(cls) -> set[str]: """Return the genotype storage type."""
[docs] @abc.abstractmethod def start(self) -> GenotypeStorage: """Allocate all resources needed for the genotype storage to work."""
[docs] @abc.abstractmethod def shutdown(self) -> GenotypeStorage: """Frees all resources used by the genotype storage to work."""
[docs] @abc.abstractmethod def build_backend( self, study_config: dict, genome: ReferenceGenome, gene_models: GeneModels) -> Any: """Construct a query backend for this genotype storage."""