add explicit fetcher property model_name

This commit is contained in:
Diogo Sousa
2023-07-29 01:04:25 +01:00
parent 64ced15536
commit 6b19cbc69e
3 changed files with 18 additions and 10 deletions

View File

@@ -24,6 +24,11 @@ class Fetcher(
):
"""Abstract class for the fetcher."""
@property
@abstractmethod
def model_name(self) -> str:
raise NotImplementedError
@staticmethod
def transform_query(params: Dict[str, Any]) -> ProviderQueryParamsType:
"""Transform the params to the provider-specific query."""
@@ -54,28 +59,24 @@ class Fetcher(
return data
@property
@abstractmethod
def query_params_type(self):
"""Get the type of the query."""
# pylint: disable=E1101
return self.__orig_bases__[0].__args__[0]
@property
@abstractmethod
def data_type(self):
"""Get the type of the data."""
# pylint: disable=E1101
return self.__orig_bases__[0].__args__[1]
@property
@abstractmethod
def provider_query_params_type(self):
"""Get the type of the provider query."""
# pylint: disable=E1101
return self.__orig_bases__[0].__args__[2]
@property
@abstractmethod
def provider_data_type(self):
"""Get the type of the provider data."""
# pylint: disable=E1101

View File

@@ -18,13 +18,21 @@ class Provider:
"""Initialize the provider."""
self.name = name
self.description = description
self.required_credentials = required_credentials
self.formatted_credentials = []
if required_credentials is not None:
for rq in required_credentials:
self.formatted_credentials.append(f"{self.name.lower()}_{rq}")
self.fetcher_list = fetcher_list
self.fetcher_dict: Dict[str, Fetcher] = {}
self.fetcher_dict = {
str(fetcher.__name__).lower(): fetcher for fetcher in fetcher_list
}
for fetcher in fetcher_list:
try:
model_name = fetcher().model_name
except NotImplementedError as e:
raise NotImplementedError(
f"'model_name' not explicitly declared for '{fetcher.__name__}'."
) from e
self.fetcher_dict[model_name] = fetcher

View File

@@ -27,12 +27,11 @@ class QueryExecutor:
def get_fetcher(self, provider: Provider, model_name: str):
"""Get a fetcher from a provider."""
fetcher_name = (provider.name + model_name + "fetcher").lower()
if fetcher_name not in provider.fetcher_dict:
if model_name not in provider.fetcher_dict:
raise ProviderError(
f"Fetcher not found for model '{model_name}' in provider '{provider.name}'."
)
return provider.fetcher_dict[fetcher_name]
return provider.fetcher_dict[model_name]
def match_credentials(
self, provider: Provider, credentials: Optional[Dict[str, str]]