Coverage for nexusLIMS/db/migrations/env.py: 100%

1# ruff: noqa: ERA001 

2"""Alembic migration environment configuration for NexusLIMS. 

3 

4This module configures the Alembic migration environment for the NexusLIMS database. 

5It handles both online and offline migration modes and automatically configures the 

6database URL from the NexusLIMS settings. 

7 

8Key features: 

9 - Automatically reads database path from NX_DB_PATH environment variable 

10 - Configures SQLModel metadata for autogenerate support 

11 - Supports both online (live database) and offline (SQL script) migrations 

12 - Imports all SQLModel classes to ensure complete schema detection 

13 

14Usage: 

15 This file is automatically used by Alembic when running migration commands: 

16 uv run alembic upgrade head 

17 uv run alembic revision --autogenerate -m "description" 

18 

19Note: 

20 All SQLModel model classes must be imported in this file (even if not directly 

21 used) to ensure Alembic can detect them for autogenerate operations. 

22""" 

23 

24import os 

25import re 

26from pathlib import Path 

27 

28import sqlalchemy as sa 

29from alembic import context 

30from alembic.script import ScriptDirectory 

31from sqlalchemy import engine_from_config, pool 

32 

33# Import SQLModel metadata and models 

34from sqlmodel import SQLModel 

35 

36from nexusLIMS.db.models import Instrument, SessionLog, UploadLog # noqa: F401 

37 

38# Derive the migrations directory from this file's own location. 

39# Works regardless of whether the package is installed or run from source. 

40_MIGRATIONS_DIR = Path(__file__).resolve().parent 

41 

42# this is the Alembic Config object, which provides 

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

44config = context.config 

45 

46# Set sqlalchemy.url directly from the environment variable rather than going 

47# through nexusLIMS.config.settings, because Settings validation requires fields 

48# (like NX_CDCS_TOKEN, NX_DATA_PATH, etc.) that are irrelevant for database 

49# migrations. This allows 'nexuslims db' commands to work with only 

50# NX_DB_PATH set. 

51_db_path = os.getenv("NX_DB_PATH", "") 

52config.set_main_option("sqlalchemy.url", f"sqlite:///{_db_path}") 

53 

54# Set target_metadata to SQLModel metadata for autogenerate support 

55target_metadata = SQLModel.metadata 

56 

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

58# can be acquired: 

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

60# ... etc. 

61 

62 

63def _generate_revision_id(context_obj) -> str: 

64 """Generate a user-friendly sequential revision ID. 

65 

66 This function creates revision IDs in the format: NNN_description 

67 where NNN is a zero-padded sequential number. 

68 

69 Examples: 001_initial_schema, 002_add_upload_log, 003_add_constraints 

70 

71 This is much more readable than random hex values while maintaining 

72 clear ordering. 

73 """ 

74 script = ScriptDirectory.from_config(config) 

75 

76 # Find the highest existing numeric revision 

77 max_num = 0 

78 for rev in script.walk_revisions(): 

79 if rev.revision and rev.revision[0].isdigit(): 

80 try: 

81 # Extract numeric prefix (e.g., "001" from "001_description") 

82 num_part = rev.revision.split("_")[0] 

83 max_num = max(max_num, int(num_part)) 

84 except (ValueError, IndexError): # pragma: no cover 

85 # Skip if not in our format 

86 pass 

87 

88 # Generate next sequential number 

89 next_num = max_num + 1 

90 

91 # Get the message from the context (cleaned up for use in ID) 

92 message = context_obj.opts.get("message", "migration") 

93 if message: 

94 # Convert message to lowercase, replace spaces/special chars with underscores 

95 sanitized = re.sub(r"[^\w\s-]", "", message.lower()) 

96 sanitized = re.sub(r"[-\s]+", "_", sanitized).strip("_") 

97 # Limit length to keep IDs reasonable 

98 sanitized = sanitized[:50] 

99 else: 

100 sanitized = "migration" 

101 

102 return f"{next_num:03d}_{sanitized}" 

103 

104 

105def process_revision_directives(context_obj, _revision, directives): 

106 """Alembic hook to customize revision generation. 

107 

108 This is called by Alembic when creating new migrations via 

109 'nexuslims db alembic revision --autogenerate'. 

110 

111 It replaces the default random hex revision ID with a sequential 

112 numbered ID for better readability. 

113 """ 

114 if config.cmd_opts and config.cmd_opts.autogenerate: 

115 script = directives[0] 

116 if script.upgrade_ops.is_empty(): 

117 # Don't generate empty migrations 

118 directives[:] = [] 

119 return 

120 

121 # Use our custom revision ID generator 

122 script.rev_id = _generate_revision_id(context_obj) 

123 

124 

125def run_migrations_offline() -> None: 

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

127 

128 This configures the context with just a URL 

129 and not an Engine, though an Engine is acceptable 

130 here as well. By skipping the Engine creation 

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

132 

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

134 script output. 

135 

136 """ 

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

138 context.configure( 

139 url=url, 

140 target_metadata=target_metadata, 

141 literal_binds=True, 

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

143 process_revision_directives=process_revision_directives, 

144 ) 

145 

146 with context.begin_transaction(): 

147 context.run_migrations() 

148 

149 

150def run_migrations_online() -> None: 

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

152 

153 In this scenario we need to create an Engine 

154 and associate a connection with the context. 

155 

156 """ 

157 from nexusLIMS.db.migrations.utils import create_backup # noqa: PLC0415 

158 

159 connectable = engine_from_config( 

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

161 prefix="sqlalchemy.", 

162 poolclass=pool.NullPool, 

163 ) 

164 

165 with connectable.connect() as connection: 

166 context.configure( 

167 connection=connection, 

168 target_metadata=target_metadata, 

169 process_revision_directives=process_revision_directives, 

170 ) 

171 

172 with context.begin_transaction(): 

173 # Create automatic backup before running migrations 

174 # Skip backup for new database initialization (no alembic_version table yet) 

175 try: 

176 destination_rev = context.get_context().opts.get("destination_rev") 

177 if destination_rev: 

178 # Check if alembic_version table exists 

179 # (indicates existing database) 

180 inspector = sa.inspect(connection) 

181 has_alembic_version = ( 

182 "alembic_version" in inspector.get_table_names() 

183 ) 

184 

185 if has_alembic_version: 

186 create_backup(connection) 

187 # else: new database initialization, skip backup 

188 except Exception: # noqa: S110 

189 # If we can't determine if backup is needed, skip it 

190 # (e.g., for read-only operations like current/history) 

191 pass 

192 

193 context.run_migrations() 

194 

195 

196if context.is_offline_mode(): 

197 run_migrations_offline() 

198else: 

199 run_migrations_online()