Coverage for nexusLIMS/db/migrations/versions/v2_5_0a_add_external_user_identifiers.py: 100%

1"""Add external_user_identifiers table. 

2 

3Adds table for mapping NexusLIMS usernames to external system user IDs using 

4a star topology design with nexuslims_username as the canonical identifier. 

5 

6Supports integration with: 

7- NEMO (lab management system) 

8- LabArchives ELN (electronic lab notebook) 

9- LabArchives Scheduler (reservation system) 

10- CDCS (Configurable Data Curation System) 

11 

12Revision ID: v2_5_0a 

13Revises: v2_4_0b 

14Create Date: 2026-02-08 14:00:00.000000 

15 

16""" 

17 

18from typing import Sequence, Union 

19 

20import sqlalchemy as sa 

21from alembic import op 

22 

23# revision identifiers, used by Alembic. 

24revision: str = "v2_5_0a" 

25down_revision: Union[str, Sequence[str], None] = "v2_4_0b" 

26branch_labels: Union[str, Sequence[str], None] = None 

27depends_on: Union[str, Sequence[str], None] = None 

28 

29 

30def upgrade() -> None: 

31 """Create external_user_identifiers table.""" 

32 # Create table with CHECK constraint and UNIQUE constraints 

33 # NOTE: CHECK constraint values are hardcoded to preserve this migration's 

34 # historical accuracy. If ExternalSystem enum changes in the future, a new 

35 # migration should be created to update the constraint. 

36 op.create_table( 

37 "external_user_identifiers", 

38 sa.Column("id", sa.Integer(), nullable=False), 

39 sa.Column("nexuslims_username", sa.String(), nullable=False), 

40 sa.Column("external_system", sa.String(), nullable=False), 

41 sa.Column("external_id", sa.String(), nullable=False), 

42 sa.Column("email", sa.String(), nullable=True), 

43 sa.Column("created_at", sa.String(), nullable=False), 

44 sa.Column("last_verified_at", sa.String(), nullable=True), 

45 sa.Column("notes", sa.String(), nullable=True), 

46 sa.CheckConstraint( 

47 "external_system IN ('nemo', 'labarchives_eln', " 

48 "'labarchives_scheduler', 'cdcs')", 

49 name="valid_external_system", 

50 ), 

51 sa.PrimaryKeyConstraint("id"), 

52 sa.UniqueConstraint( 

53 "nexuslims_username", 

54 "external_system", 

55 name="nexuslims_username_system_UNIQUE", 

56 ), 

57 sa.UniqueConstraint( 

58 "external_system", "external_id", name="system_external_id_UNIQUE" 

59 ), 

60 ) 

61 

62 # Create indexes 

63 op.create_index( 

64 "idx_external_lookup", 

65 "external_user_identifiers", 

66 ["external_system", "external_id"], 

67 unique=False, 

68 ) 

69 op.create_index( 

70 "idx_nexuslims_username", 

71 "external_user_identifiers", 

72 ["nexuslims_username"], 

73 unique=False, 

74 ) 

75 

76 

77def downgrade() -> None: 

78 """Drop external_user_identifiers table.""" 

79 # Drop indexes 

80 op.drop_index("idx_nexuslims_username", table_name="external_user_identifiers") 

81 op.drop_index("idx_external_lookup", table_name="external_user_identifiers") 

82 

83 # Drop table 

84 op.drop_table("external_user_identifiers")