Coverage for tld/base.py: 86%

59 statements  

« prev     ^ index     » next       coverage.py v7.2.7, created at 2024-05-26 22:29 +0000

1import logging 

2from codecs import open as codecs_open 

3from typing import Dict, ItemsView, Optional, Union 

4from urllib.request import urlopen 

5 

6from .exceptions import TldImproperlyConfigured, TldIOError 

7from .helpers import project_dir 

8 

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) 

16 

17LOGGER = logging.getLogger(__name__) 

18 

19 

20class Registry(type): 

21 

22 REGISTRY: Dict[str, "BaseTLDSourceParser"] = {} 

23 

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 

31 

32 @property 

33 def _uid(cls) -> str: 

34 return getattr(cls, "uid", cls.__name__) 

35 

36 @classmethod 

37 def reset(mcs) -> None: 

38 mcs.REGISTRY = {} 

39 

40 @classmethod 

41 def get( 

42 mcs, key: str, default: "BaseTLDSourceParser" = None 

43 ) -> Union["BaseTLDSourceParser", None]: 

44 return mcs.REGISTRY.get(key, default) 

45 

46 @classmethod 

47 def items(mcs) -> ItemsView[str, "BaseTLDSourceParser"]: 

48 return mcs.REGISTRY.items() 

49 

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) 

57 

58 

59class BaseTLDSourceParser(metaclass=Registry): 

60 """Base TLD source parser.""" 

61 

62 uid: Optional[str] = None 

63 source_url: str 

64 local_path: str 

65 include_private: bool = True 

66 

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 ) 

74 

75 @classmethod 

76 def get_tld_names(cls, fail_silently: bool = False, retry_count: int = 0): 

77 """Get tld names. 

78 

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 ) 

87 

88 @classmethod 

89 def update_tld_names(cls, fail_silently: bool = False) -> bool: 

90 """Update the local copy of the TLD file. 

91 

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) 

112 

113 return True