Coverage for an_website / __init__.py: 79.167%

1# This program is free software: you can redistribute it and/or modify 

2# it under the terms of the GNU Affero General Public License as 

3# published by the Free Software Foundation, either version 3 of the 

4# License, or (at your option) any later version. 

5# 

6# This program is distributed in the hope that it will be useful, 

7# but WITHOUT ANY WARRANTY; without even the implied warranty of 

8# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the 

9# GNU Affero General Public License for more details. 

10# 

11# You should have received a copy of the GNU Affero General Public License 

12# along with this program. If not, see <https://www.gnu.org/licenses/>. 

13 

14"""The website of the AN.""" 

15 

16 

17import multiprocessing 

18import os 

19import sys 

20import time 

21from asyncio import Event 

22from collections.abc import Mapping 

23from importlib.metadata import Distribution 

24from importlib.resources import files 

25from importlib.resources.abc import Traversable 

26from pathlib import Path 

27from typing import Final, TypedDict 

28 

29try: 

30 import orjson as json 

31 

32 if "spam" not in json.loads('{"spam":"eggs"}'): 

33 from marshal import dumps, loads 

34 

35 _loads = json.loads 

36 json.loads = lambda *args, **kwargs: loads( # nosec: B302 

37 dumps(_loads(*args, **kwargs)) 

38 ) 

39except ModuleNotFoundError: 

40 from . import fake_orjson as json # type: ignore[no-redef] # noqa: F811 

41 

42 sys.modules["orjson"] = json 

43 

44try: 

45 from pytest_is_running import is_running as pytest_is_running 

46except ModuleNotFoundError: 

47 

48 def pytest_is_running() -> bool: # noqa: D103 

49 # pylint: disable=missing-function-docstring 

50 return "pytest" in sys.modules 

51 

52 

53class MediaType(TypedDict, total=False): 

54 # pylint: disable=missing-class-docstring 

55 charset: str 

56 compressible: bool 

57 extensions: list[str] 

58 source: str 

59 

60 

61class UptimeTimer: 

62 """UptimeTimer class used for timing the uptime.""" 

63 

64 __slots__ = ("_start_time",) 

65 

66 def __init__(self) -> None: 

67 self.reset() 

68 

69 def get(self) -> float: 

70 """Get the time since start in seconds.""" 

71 return self.get_ns() / 1_000_000_000 

72 

73 def get_ns(self) -> int: 

74 """Get the time since start in nanoseconds.""" 

75 return time.monotonic_ns() - self._start_time 

76 

77 def reset(self) -> None: 

78 """Reset the timer.""" 

79 self._start_time = time.monotonic_ns() 

80 

81 

82UPTIME: Final = UptimeTimer() 

83 

84EPOCH: Final[int] = 1651075200 

85EPOCH_MS: Final[int] = EPOCH * 1000 

86 

87DIR: Final[Traversable] = files(__name__) 

88 

89MEDIA_TYPES: Final[Mapping[str, MediaType]] = json.loads( 

90 (DIR / "vendored" / "mime-db.json").read_bytes() 

91) 

92 

93NAME = "an-website" # pylint: disable=invalid-name 

94 

95 

96def get_version() -> str: 

97 """Get the version of the package.""" 

98 if isinstance(DIR, Path): 

99 # pylint: disable-next=import-outside-toplevel 

100 from get_version import get_version as gv 

101 

102 return gv(__file__, vcs="git") 

103 

104 return Distribution.from_name(NAME).version 

105 

106 

107VERSION: Final[str] = get_version() 

108 

109GH_ORG_URL: Final[str] = "https://github.com/asozialesnetzwerk" 

110GH_REPO_URL: Final[str] = f"{GH_ORG_URL}/{NAME}" 

111GH_PAGES_URL: Final[str] = f"https://github.asozial.org/{NAME}" 

112 

113CACHE_DIR: Final[Path] = Path("~/.cache/", NAME).expanduser().absolute() 

114STATIC_DIR: Final[Traversable] = DIR / "static" 

115TEMPLATES_DIR: Final[Traversable] = DIR / "templates" 

116 

117ORJSON_OPTIONS: Final[int] = ( 

118 json.OPT_SERIALIZE_NUMPY | json.OPT_NAIVE_UTC | json.OPT_UTC_Z 

119) 

120 

121CONTAINERIZED: Final[bool] = "container" in os.environ or os.path.exists( 

122 "/.dockerenv" 

123) 

124 

125 

126def traversable_to_file(traversable: Traversable) -> Path: 

127 """Convert a traversable to a path to a file directly in the fs.""" 

128 if isinstance(traversable, Path): 

129 return traversable 

130 folder = CACHE_DIR / "temp" 

131 folder.mkdir(parents=True, exist_ok=True) 

132 file = folder / traversable.name 

133 file.write_bytes(traversable.read_bytes()) 

134 return file 

135 

136 

137CA_BUNDLE_PATH: Final[str] = traversable_to_file( 

138 DIR / "ca-bundle.crt" 

139).as_posix() 

140 

141 

142if pytest_is_running(): 

143 NAME += "-test" 

144elif sys.flags.dev_mode: 

145 NAME += "-dev" 

146 

147EVENT_SHUTDOWN: Final = multiprocessing.Event() 

148 

149EVENT_ELASTICSEARCH: Final = Event() 

150EVENT_REDIS: Final = Event() 

151 

152__all__ = ( 

153 "CONTAINERIZED", 

154 "DIR", 

155 "EPOCH", 

156 "EPOCH_MS", 

157 "EVENT_ELASTICSEARCH", 

158 "EVENT_REDIS", 

159 "EVENT_SHUTDOWN", 

160 "GH_ORG_URL", 

161 "GH_PAGES_URL", 

162 "GH_REPO_URL", 

163 "MEDIA_TYPES", 

164 "NAME", 

165 "ORJSON_OPTIONS", 

166 "STATIC_DIR", 

167 "TEMPLATES_DIR", 

168 "UPTIME", 

169 "VERSION", 

170 "pytest_is_running", 

171) 

172 

173__version__ = VERSION