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

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 

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

108 """ 

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

110 

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

112 

113 We sync as follows: 

114 

115 1. Lock the table to be updated fully 

116 2. Defer all constraints 

117 3. Truncate the table 

118 4. Re-insert everything 

119 

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

121 """ 

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

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

124 

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

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

127 

128 timezone_areas_file = resources_folder / "timezone_areas.sql" 

129 

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

131 if not config["DEV"]: 

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

133 

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

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

136 

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

138 tz_sql = f.read() 

139 

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

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

142 

143 session.execute(delete(Region)) 

144 for code, name in regions: 

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

146 

147 session.execute(delete(Language)) 

148 for code, name in languages: 

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

150 

151 session.execute(delete(TimezoneArea)) 

152 session.execute(text(tz_sql))