Coverage for src/couchers/resources.py: 93%

1import functools 

2import json 

3import logging 

4from pathlib import Path 

5 

6from sqlalchemy.sql import delete, text 

7 

8from couchers.config import config 

9from couchers.db import session_scope 

10from couchers.models import Language, Region, TimezoneArea 

11from couchers.sql import couchers_select as select 

12from proto import resources_pb2 

13 

14logger = logging.getLogger(__name__) 

15 

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

17 

18 

19@functools.lru_cache 

20def get_terms_of_service(): 

21 """ 

22 Get the latest terms of service 

23 """ 

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

25 return f.read() 

26 

27 

28@functools.lru_cache 

29def get_community_guidelines(): 

30 """ 

31 Get the latest Community Guidelines 

32 """ 

33 with open(resources_folder / "community_guidelines.json", "r") as f: 

34 community_guidelines = json.load(f) 

35 ret = [] 

36 for cg in community_guidelines: 

37 with open(resources_folder / "icons" / cg["icon"], "r") as f: 

38 ret.append( 

39 resources_pb2.CommunityGuideline( 

40 title=cg["title"], 

41 guideline=cg["guideline"], 

42 icon_svg=f.read(), 

43 ) 

44 ) 

45 return ret 

46 

47 

48@functools.lru_cache 

49def get_region_dict(): 

50 """ 

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

52 """ 

53 with session_scope() as session: 

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

55 

56 

57def region_is_allowed(code): 

58 """ 

59 Check a region code is valid 

60 """ 

61 return code in get_region_dict() 

62 

63 

64@functools.lru_cache 

65def get_language_dict(): 

66 """ 

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

68 """ 

69 with session_scope() as session: 

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

71 

72 

73def language_is_allowed(code): 

74 """ 

75 Check a language code is valid 

76 """ 

77 return code in get_language_dict() 

78 

79 

80def copy_resources_to_database(session): 

81 """ 

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

83 

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

85 

86 We sync as follows: 

87 

88 1. Lock the table to be updated fully 

89 2. Defer all constraints 

90 3. Truncate the table 

91 4. Re-insert everything 

92 

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

94 """ 

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

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

97 

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

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

100 

101 timezone_areas_file = resources_folder / "timezone_areas.sql" 

102 

103 if not timezone_areas_file.exists(): 

104 if not config["DEV"]: 

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

106 

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

108 logger.info(f"Using fake timezone areas") 

109 

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

111 tz_sql = f.read() 

112 

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

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

115 

116 session.execute(delete(Region)) 

117 for code, name in regions: 

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

119 

120 session.execute(delete(Language)) 

121 for code, name in languages: 

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

123 

124 session.execute(delete(TimezoneArea)) 

125 session.execute(text(tz_sql))