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 # if set, the badge is only awarded while this feature flag is on (the flag defaults to on, so 

81 # the badge keeps being awarded until the flag is turned off) 

82 flag: str | None = None 

83 

84 

85@functools.cache 

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

87 """ 

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

89 """ 

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

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

92 

93 

94@functools.cache 

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

96 """ 

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

98 """ 

99 data = get_badge_data()["static_badges"] 

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

101 

102 

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

104 """ 

105 Check a language code is valid 

106 """ 

107 return code in get_language_dict() 

108 

109 

110@functools.cache 

111def get_postcard_front_image() -> bytes: 

112 """ 

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

114 """ 

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

116 

117 

118@functools.cache 

119def get_postcard_font() -> bytes: 

120 """ 

121 Returns the font file for postcard text rendering. 

122 """ 

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

124 

125 

126@functools.cache 

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

128 """ 

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

130 """ 

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

132 

133 

134@functools.cache 

135def get_postcard_back_left_template() -> bytes: 

136 """ 

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

138 """ 

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

140 

141 

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

143 """ 

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

145 

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

147 

148 We sync as follows: 

149 

150 1. Lock the table to be updated fully 

151 2. Defer all constraints 

152 3. Truncate the table 

153 4. Re-insert everything 

154 

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

156 """ 

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

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

159 

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

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

162 

163 timezone_areas_file = resources_folder / "timezone_areas.sql" 

164 

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

166 if not config.DEV: 

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

168 

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

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

171 

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

173 tz_sql = f.read() 

174 

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

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

177 

178 session.execute(delete(Region)) 

179 for code, name in regions: 

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

181 

182 session.execute(delete(Language)) 

183 for code, name in languages: 

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

185 

186 session.execute(delete(TimezoneArea)) 

187 session.execute(text(tz_sql))