diff --git a/updates_notifier/__init__.py b/updates_notifier/__init__.py index e69de29bb2d1d6434b8b29ae775ad8c2e48c5391..04edc16bbf36c9e318d84e4cbd83e886b2cc1485 100644 --- a/updates_notifier/__init__.py +++ b/updates_notifier/__init__.py @@ -0,0 +1 @@ +from .entry import Entry diff --git a/updates_notifier/fetchers/__init__.py b/updates_notifier/fetchers/__init__.py index a6b34eb1964de533fa1e10e292b7e9f2c9ef8a63..50ac6f0dc9be63a3f768bb9ed5fe0ddd552e5ca5 100644 --- a/updates_notifier/fetchers/__init__.py +++ b/updates_notifier/fetchers/__init__.py @@ -1 +1,2 @@ from .project_fetcher import ProjectFetcher +from .utils import create_fetchers_from_dict diff --git a/updates_notifier/fetchers/utils.py b/updates_notifier/fetchers/utils.py new file mode 100644 index 0000000000000000000000000000000000000000..80c9e37c857bd4be569350318ee5c79e7eec5e2d --- /dev/null +++ b/updates_notifier/fetchers/utils.py @@ -0,0 +1,99 @@ +from typing import Any, Dict, List, Type + +from .atom_rss_fetcher import AtomRSSFetcher +from .gitea_fetcher import GiteaFetcher +from .github_fetcher import GithubFetcher +from .gitlab_fetcher import GitLabFetcher +from .project_fetcher import ProjectFetcher + +_TYPE_ATOM_RSS = "atom_rss" +_TYPE_GITEA = "gitea" +_TYPE_GITHUB = "github" +_TYPE_GITLAB = "gitlab" + + +class FactoryError(RuntimeError): + pass + + +class FactoryParamError(FactoryError): + def __init__(self, param_name: str, expected_type: Type, given_type: Type): + self.message = ( + f"Invalid type for parameter {param_name}, wanted {expected_type.__name__}" + f", but got {given_type.__name__}" + ) + self.repr = ( + f"<{type(self).__name__} param_name={param_name}" + f"expected_type={expected_type.__name__} given_type={given_type.__name__}>" + ) + + def __str__(self) -> str: + return self.message + + def __repr__(self) -> str: + return self.repr + + +def _assert_param_type_is(param_name: str, value: Any, expected_type: Type) -> None: + if not isinstance(value, expected_type): + raise FactoryParamError(param_name, expected_type, type(value).__name__) + + +def create_fetcher( + fetcher_type: str, feed_id: int, name: str, **kwargs +) -> ProjectFetcher: + if not isinstance(fetcher_type, str): + raise FactoryError( + f"Fetcher type must be a str, got {type(fetcher_type).__name__}" + ) + + _assert_param_type_is("feed_id", feed_id, int) + _assert_param_type_is("name", name, str) + + if fetcher_type == _TYPE_ATOM_RSS: + url = kwargs.get("url") + _assert_param_type_is("url", url, str) + + return AtomRSSFetcher(feed_id, name, url) + + if fetcher_type == _TYPE_GITEA: + gitea_host = kwargs.get("host") + _assert_param_type_is("host", gitea_host, str) + + repository = kwargs.get("repository") + _assert_param_type_is("repository", repository, str) + + return GiteaFetcher(feed_id, name, gitea_host, repository) + + if fetcher_type == _TYPE_GITHUB: + repository = kwargs.get("repository") + _assert_param_type_is("repository", repository, str) + + return GithubFetcher(feed_id, name, repository) + + if fetcher_type == _TYPE_GITLAB: + gitlab_host = kwargs.get("host") + _assert_param_type_is("host", gitlab_host, str) + + repository = kwargs.get("repository") + _assert_param_type_is("repository", repository, str) + + return GitLabFetcher(feed_id, name, gitlab_host, repository) + + raise FactoryError(f"Invalid fetcher type '{fetcher_type}'") + + +def create_fetchers_from_dict(feeds_params: List[Any]) -> List[ProjectFetcher]: + assert isinstance(feeds_params, list) + + fetchers: List[ProjectFetcher] = [] + + for idx, params in enumerate(feeds_params): + try: + fetchers.append(create_fetcher(**params)) + except FactoryError as err: + raise RuntimeError( + "Could not load fetcher at index %s: %s", idx, str(err) + ) from err + + return fetchers