import logging
from typing import Any, Dict, Optional
from dae.genotype_storage import get_genotype_storage_factory
from dae.genotype_storage.genotype_storage import GenotypeStorage
logger = logging.getLogger(__name__)
[docs]class GenotypeStorageRegistry:
"""Registry for genotype storages.
This class could accept genotype storages config from a GPF instance
configuration and instantiate and register all genotype storages defined
in this configuration. To do this, one could use
:meth:`GenotypeStorageRegistry.register_storages_configs`.
To create and register single genotype storage using its configuration
you can use :meth:`GenotypeStorageRegistry.register_storage_config`.
When you have already created an instance of genotype storage, you can use
:meth:`GenotypeStorageRegistry.register_genotype_storage` to register
it.
"""
def __init__(self) -> None:
self._genotype_storages: Dict[str, GenotypeStorage] = {}
self._default_genotype_storage: Optional[GenotypeStorage] = None
[docs] def register_storage_config(
self, storage_config: dict[str, Any]) -> GenotypeStorage:
"""Create a genotype storage using storage config and registers it."""
storage_type = storage_config["storage_type"]
storage_factory = get_genotype_storage_factory(storage_type)
genotype_storage = storage_factory(storage_config)
genotype_storage.start()
return self.register_genotype_storage(genotype_storage)
[docs] def register_genotype_storage(
self, storage: GenotypeStorage) -> GenotypeStorage:
"""Register a genotype storage instance."""
if not isinstance(storage, GenotypeStorage):
raise ValueError(
f"trying to register object of type <{type(storage)}>"
f" as genotype storage.")
storage_id = storage.storage_id
self._genotype_storages[storage_id] = storage
return storage
[docs] def register_default_storage(
self, genotype_storage: GenotypeStorage) -> None:
"""Register a genotype storage and make it the default storage."""
self.register_genotype_storage(genotype_storage)
self._default_genotype_storage = genotype_storage
[docs] def get_default_genotype_storage(self) -> GenotypeStorage:
"""Return the default genotype storage if one is defined.
Otherwise, return None.
"""
if self._default_genotype_storage is None:
raise ValueError("default genotype storage not set")
return self._default_genotype_storage
[docs] def get_genotype_storage(self, storage_id: str) -> GenotypeStorage:
"""Return genotype storage with specified storage_id.
If the method can not find storage with the specified ID, it will raise
ValueError exception.
"""
if storage_id is None:
return self.get_default_genotype_storage()
if storage_id not in self._genotype_storages:
raise ValueError(f"unknown storage id {storage_id}")
return self._genotype_storages[storage_id]
[docs] def get_all_genotype_storage_ids(self) -> list[str]:
"""Return list of all registered genotype storage IDs."""
return list(self._genotype_storages.keys())
[docs] def get_all_genotype_storages(self) -> list[GenotypeStorage]:
"""Return list of registered genotype storages."""
return list(self._genotype_storages.values())
[docs] def register_storages_configs(
self, genotype_storages_config: dict[str, Any]) -> None:
"""Create and register all genotype storages defined in config.
When defining a GPF instance, we specify a `genotype_storage` section
in the configuration. If you pass this whole configuration section
to this method, it will create and register all genotype storages
defined in that configuration section.
"""
for storage_config in genotype_storages_config["storages"]:
self.register_storage_config(storage_config)
default_storage_id = genotype_storages_config.get("default")
if default_storage_id is not None:
storage = self.get_genotype_storage(default_storage_id)
self.register_default_storage(storage)
[docs] def shutdown(self) -> None:
for storage_id, storage in self._genotype_storages.items():
logger.info("shutting down genotype storage %s", storage_id)
storage.shutdown()