Coverage for tld/base.py: 86%
59 statements
« prev ^ index » next coverage.py v7.4.4, created at 2024-05-27 05:40 +0000
« prev ^ index » next coverage.py v7.4.4, created at 2024-05-27 05:40 +0000
1import logging
2from codecs import open as codecs_open
3from typing import Dict, ItemsView, Optional, Union
4from urllib.request import urlopen
6from .exceptions import TldImproperlyConfigured, TldIOError
7from .helpers import project_dir
9__author__ = "Artur Barseghyan"
10__copyright__ = "2013-2023 Artur Barseghyan"
11__license__ = "MPL-1.1 OR GPL-2.0-only OR LGPL-2.1-or-later"
12__all__ = (
13 "BaseTLDSourceParser",
14 "Registry",
15)
17LOGGER = logging.getLogger(__name__)
20class Registry(type):
22 REGISTRY: Dict[str, "BaseTLDSourceParser"] = {}
24 def __new__(mcs, name, bases, attrs):
25 new_cls = type.__new__(mcs, name, bases, attrs)
26 # Here the name of the class is used as key but it could be any class
27 # parameter.
28 if getattr(new_cls, "_uid", None):
29 mcs.REGISTRY[new_cls._uid] = new_cls
30 return new_cls
32 @property
33 def _uid(cls) -> str:
34 return getattr(cls, "uid", cls.__name__)
36 @classmethod
37 def reset(mcs) -> None:
38 mcs.REGISTRY = {}
40 @classmethod
41 def get(
42 mcs, key: str, default: "BaseTLDSourceParser" = None
43 ) -> Union["BaseTLDSourceParser", None]:
44 return mcs.REGISTRY.get(key, default)
46 @classmethod
47 def items(mcs) -> ItemsView[str, "BaseTLDSourceParser"]:
48 return mcs.REGISTRY.items()
50 # @classmethod
51 # def get_registry(mcs) -> Dict[str, Type]:
52 # return dict(mcs.REGISTRY)
53 #
54 # @classmethod
55 # def pop(mcs, uid) -> None:
56 # mcs.REGISTRY.pop(uid)
59class BaseTLDSourceParser(metaclass=Registry):
60 """Base TLD source parser."""
62 uid: Optional[str] = None
63 source_url: str
64 local_path: str
65 include_private: bool = True
67 @classmethod
68 def validate(cls):
69 """Constructor."""
70 if not cls.uid:
71 raise TldImproperlyConfigured(
72 "The `uid` property of the TLD source parser shall be defined."
73 )
75 @classmethod
76 def get_tld_names(cls, fail_silently: bool = False, retry_count: int = 0):
77 """Get tld names.
79 :param fail_silently:
80 :param retry_count:
81 :return:
82 """
83 cls.validate()
84 raise NotImplementedError(
85 "Your TLD source parser shall implement `get_tld_names` method."
86 )
88 @classmethod
89 def update_tld_names(cls, fail_silently: bool = False) -> bool:
90 """Update the local copy of the TLD file.
92 :param fail_silently:
93 :return:
94 """
95 try:
96 remote_file = urlopen(cls.source_url)
97 local_file_abs_path = project_dir(cls.local_path)
98 local_file = codecs_open(local_file_abs_path, "wb", encoding="utf8")
99 local_file.write(remote_file.read().decode("utf8"))
100 local_file.close()
101 remote_file.close()
102 LOGGER.info(
103 f"Fetched '{cls.source_url}' as '{local_file_abs_path}'"
104 )
105 except Exception as err:
106 LOGGER.error(
107 f"Failed fetching '{cls.source_url}'. Reason: {str(err)}"
108 )
109 if fail_silently:
110 return False
111 raise TldIOError(err)
113 return True