Coverage for app / backend / src / couchers / resources.py: 92%

1import functools 

2import json 

3import logging 

4from dataclasses import dataclass 

5from pathlib import Path 

6from typing import Any, cast 

7 

8from sqlalchemy import select 

9from sqlalchemy.orm import Session 

10from sqlalchemy.sql import delete, text 

11 

12from couchers.config import config 

13from couchers.db import session_scope 

14from couchers.models import Language, Region, TimezoneArea 

15 

16logger = logging.getLogger(__name__) 

17 

18resources_folder = Path(__file__).parent / ".." / ".." / "resources" 

19 

20 

21@functools.cache 

22def get_terms_of_service() -> str: 

23 """ 

24 Get the latest terms of service 

25 """ 

26 with open(resources_folder / "terms_of_service.md", "r") as f: 

27 return f.read() 

28 

29 

30@functools.cache 

31def get_icon(name: str) -> str: 

32 """ 

33 Get an icon SVG by name 

34 """ 

35 return (resources_folder / "icons" / name).read_text() 

36 

37 

38@functools.cache 

39def get_region_dict() -> dict[str, str]: 

40 """ 

41 Get a list of allowed regions as a dictionary of {alpha3: name}. 

42 """ 

43 with session_scope() as session: 

44 return {region.code: region.name for region in session.execute(select(Region)).scalars().all()} 

45 

46 

47def region_is_allowed(code: str) -> bool: 

48 """ 

49 Check a region code is valid 

50 """ 

51 return code in get_region_dict() 

52 

53 

54@functools.cache 

55def get_language_dict() -> dict[str, str]: 

56 """ 

57 Get a list of allowed languages as a dictionary of {code: name}. 

58 """ 

59 with session_scope() as session: 

60 return {language.code: language.name for language in session.execute(select(Language)).scalars().all()} 

61 

62 

63@functools.cache 

64def get_badge_data() -> dict[str, Any]: 

65 """ 

66 Get a list of profile badges in form {id: Badge} 

67 """ 

68 with open(resources_folder / "badges.json", "r") as f: 

69 data = json.load(f) 

70 return cast(dict[str, Any], data) 

71 

72 

73@dataclass(frozen=True, slots=True, kw_only=True) 

74class Badge: 

75 """Defines a profile badge that can be awarded to users.""" 

76 

77 id: str 

78 color: str 

79 admin_editable: bool 

80 

81 

82@functools.cache 

83def get_badge_dict() -> dict[str, Badge]: 

84 """ 

85 Get a list of profile badges in form {id: Badge} 

86 """ 

87 badges = [Badge(**b) for b in get_badge_data()["badges"]] 

88 return {badge.id: badge for badge in badges} 

89 

90 

91@functools.cache 

92def get_static_badge_dict() -> dict[str, list[int]]: 

93 """ 

94 Get a list of static badges in form {id: list(user_ids)} 

95 """ 

96 data = get_badge_data()["static_badges"] 

97 return cast(dict[str, list[int]], data) 

98 

99 

100def language_is_allowed(code: str) -> bool: 

101 """ 

102 Check a language code is valid 

103 """ 

104 return code in get_language_dict() 

105 

106 

107@functools.cache 

108def get_postcard_front_image() -> bytes: 

109 """ 

110 Returns the front image of the postcard as PNG bytes. 

111 """ 

112 return (resources_folder / "postcard-front.png").read_bytes() 

113 

114 

115@functools.cache 

116def get_postcard_font() -> bytes: 

117 """ 

118 Returns the font file for postcard text rendering. 

119 """ 

120 return (resources_folder / "hack-bold.ttf").read_bytes() 

121 

122 

123@functools.cache 

124def get_postcard_metadata() -> dict[str, Any]: 

125 """ 

126 Returns the postcard metadata (coordinates, sizes, etc.) from postcard-metadata.json. 

127 """ 

128 return cast(dict[str, Any], json.loads((resources_folder / "postcard-metadata.json").read_text())) 

129 

130 

131@functools.cache 

132def get_postcard_back_left_template() -> bytes: 

133 """ 

134 Returns the back left side template image for the postcard as PNG bytes. 

135 """ 

136 return (resources_folder / "postcard-back-left.png").read_bytes() 

137 

138 

139def copy_resources_to_database(session: Session) -> None: 

140 """ 

141 Syncs the source-of-truth data from files into the database. Call this at the end of a migration. 

142 

143 Foreign key constraints that refer to resource tables need to be set to DEFERRABLE. 

144 

145 We sync as follows: 

146 

147 1. Lock the table to be updated fully 

148 2. Defer all constraints 

149 3. Truncate the table 

150 4. Re-insert everything 

151 

152 Truncating and recreating guarantees the data is fully in sync. 

153 """ 

154 with open(resources_folder / "regions.json", "r") as f: 

155 regions = [(region["alpha3"], region["name"]) for region in json.load(f)] 

156 

157 with open(resources_folder / "languages.json", "r") as f: 

158 languages = [(language["code"], language["name"]) for language in json.load(f)] 

159 

160 timezone_areas_file = resources_folder / "timezone_areas.sql" 

161 

162 if not timezone_areas_file.exists(): 162 ↛ 163line 162 didn't jump to line 163 because the condition on line 162 was never true

163 if not config["DEV"]: 

164 raise Exception("Missing timezone_areas.sql and not running in dev") 

165 

166 timezone_areas_file = resources_folder / "timezone_areas.sql-fake" 

167 logger.info("Using fake timezone areas") 

168 

169 with open(timezone_areas_file, "r") as f: 

170 tz_sql = f.read() 

171 

172 # set all constraints marked as DEFERRABLE to be checked at the end of this transaction, not immediately 

173 session.execute(text("SET CONSTRAINTS ALL DEFERRED")) 

174 

175 session.execute(delete(Region)) 

176 for code, name in regions: 

177 session.add(Region(code=code, name=name)) 

178 

179 session.execute(delete(Language)) 

180 for code, name in languages: 

181 session.add(Language(code=code, name=name)) 

182 

183 session.execute(delete(TimezoneArea)) 

184 session.execute(text(tz_sql))