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