from collections import Counter
from typing import List, TypeVar
from q2_sdk.core.exceptions import ConfigurationError
from .db_config import DbConfig
T = TypeVar("T", bound=DbConfig)
[docs]
class DbConfigList:
"""Holds instances of DbConfig with properties for filtering"""
def __init__(self, db_configs: List[T]):
"""
:param db_configs: DbConfig instances which belong to this object
"""
all_keys = Counter([x.name for x in db_configs])
duplicates = [x for x, y in all_keys.items() if y > 1]
if duplicates:
error_msg = "Duplicate DbConfig key definitions: %s" % duplicates
raise ConfigurationError(error_msg)
self.db_configs = db_configs
def __getitem__(self, key):
for item in self.db_configs:
if item.name == key:
return item.default
raise KeyError(key)
@property
def required(self):
"""self.db_configs with required=True"""
return [x for x in self.db_configs if x.required]
@property
def optional(self):
"""self.db_configs with required=False"""
return [x for x in self.db_configs if not x.required]
@staticmethod
def from_dict(required_configs: dict, optional_configs: dict):
required = [DbConfig(x, y) for x, y in required_configs.items()]
optional = [DbConfig(x, y, required=False) for x, y in optional_configs.items()]
return DbConfigList(required + optional)