Coverage for app / backend / src / couchers / migrations / env.py: 44%

1import re 

2from logging.config import fileConfig 

3from pathlib import Path 

4from typing import Any 

5 

6from alembic import context 

7from alembic.config import Config 

8from sqlalchemy import engine_from_config, pool 

9from sqlalchemy.schema import MetaData 

10 

11from couchers import models 

12from couchers.config import config as couchers_config 

13 

14# this is the Alembic Config object, which provides 

15# access to the values within the .ini file in use. 

16config: Config = context.config 

17 

18config.set_main_option("sqlalchemy.url", couchers_config["DATABASE_CONNECTION_STRING"]) 

19 

20 

21# Interpret the config file for Python logging. 

22# This line sets up loggers basically. 

23if config.get_main_option("dont_mess_up_logging", "dont_care") == "dont_care": 23 ↛ 24line 23 didn't jump to line 24 because the condition on line 23 was never true

24 if not config.config_file_name: 

25 raise RuntimeError(config.config_file_name) 

26 fileConfig(config.config_file_name) 

27 

28# add your model's MetaData object here 

29# for 'autogenerate' support 

30# from myapp import mymodel 

31# target_metadata = mymodel.Base.metadata 

32target_metadata: MetaData = models.Base.metadata 

33 

34# other values from the config, defined by the needs of env.py, 

35# can be acquired: 

36# my_important_option = config.get_main_option("my_important_option") 

37# ... etc. 

38 

39 

40exclude_tables = config.get_section("alembic:exclude", {}).get("tables", "").split(",") 

41 

42VERSIONS_DIR = Path(__file__).parent / "versions" 

43 

44 

45def _next_ordinal() -> str: 

46 """Compute the next ordinal migration number (e.g. "0134") by scanning existing filenames.""" 

47 max_ordinal = 0 

48 for path in VERSIONS_DIR.glob("*.py"): 

49 match = re.match(r"^(\d+)_", path.name) 

50 if match: 

51 max_ordinal = max(max_ordinal, int(match.group(1))) 

52 return f"{max_ordinal + 1:04d}" 

53 

54 

55def process_revision_directives(context: Any, revision: Any, directives: Any) -> None: 

56 """Set the revision ID to the next ordinal number instead of a random hex string.""" 

57 if directives: 

58 directives[0].rev_id = _next_ordinal() 

59 

60 

61def include_name(name: str | None, type_: str, parent_names: Any) -> bool: 

62 if type_ == "schema": 

63 return name in [None, "logging"] 

64 if type_ == "table": 

65 return name not in exclude_tables 

66 return True 

67 

68 

69def include_object(obj: Any, name: str | None, type_: str, reflected: bool, compare_to: Any) -> bool: 

70 """ 

71 Filter objects during autogenerate comparison. 

72 

73 Unlike include_name (which only filters database reflection), this hook 

74 filters BOTH database objects AND model metadata objects, which is needed 

75 to properly ignore GeoAlchemy2's auto-generated spatial indexes. 

76 """ 

77 # Filter out GeoAlchemy2 auto-generated spatial indexes (from both DB and metadata) 

78 # These indexes are named like: idx_<table>_<column> and use GIST 

79 if type_ == "index" and name is not None and name.startswith("idx_") and name.endswith("_geom"): 

80 return False 

81 return True 

82 

83 

84def run_migrations_offline() -> None: 

85 """Run migrations in 'offline' mode. 

86 

87 This configures the context with just a URL 

88 and not an Engine, though an Engine is acceptable 

89 here as well. By skipping the Engine creation 

90 we don't even need a DBAPI to be available. 

91 

92 Calls to context.execute() here emit the given string to the 

93 script output. 

94 

95 """ 

96 url = config.get_main_option("sqlalchemy.url") 

97 context.configure( 

98 url=url, 

99 target_metadata=target_metadata, 

100 literal_binds=True, 

101 dialect_opts={"paramstyle": "named"}, 

102 include_schemas=True, 

103 include_name=include_name, 

104 include_object=include_object, 

105 compare_type=True, 

106 process_revision_directives=process_revision_directives, 

107 ) 

108 

109 with context.begin_transaction(): 

110 context.run_migrations() 

111 

112 

113def run_migrations_online() -> None: 

114 """Run migrations in 'online' mode. 

115 

116 In this scenario we need to create an Engine 

117 and associate a connection with the context. 

118 

119 """ 

120 connectable = engine_from_config( 

121 config.get_section(config.config_ini_section, {}), 

122 prefix="sqlalchemy.", 

123 poolclass=pool.NullPool, 

124 ) 

125 

126 with connectable.connect() as connection: 

127 context.configure( 

128 connection=connection, 

129 target_metadata=target_metadata, 

130 include_schemas=True, 

131 include_name=include_name, 

132 include_object=include_object, 

133 process_revision_directives=process_revision_directives, 

134 ) 

135 

136 with context.begin_transaction(): 

137 context.run_migrations() 

138 

139 

140if context.is_offline_mode(): 140 ↛ 141line 140 didn't jump to line 141 because the condition on line 140 was never true

141 run_migrations_offline() 

142else: 

143 run_migrations_online()