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

1"""Remove unused fields and rename schema_name to display_name. 

2 

3Removes computer_name, computer_ip, computer_mount (deprecated Session Logger App 

4fields) and calendar_name (unused field). Also renames schema_name to display_name 

5for clarity. 

6 

7Revision ID: v2_5_0b 

8Revises: v2_5_0a 

9Create Date: 2026-02-08 14:30:00.000000 

10 

11""" 

12 

13from typing import Sequence, Union 

14 

15import sqlalchemy as sa 

16from alembic import op 

17 

18# revision identifiers, used by Alembic. 

19revision: str = "v2_5_0b" 

20down_revision: Union[str, Sequence[str], None] = "v2_5_0a" 

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

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

23 

24 

25def upgrade() -> None: 

26 """Remove unused fields and rename schema_name to display_name.""" 

27 # Define the table structure for SQL mode support 

28 # This allows batch operations to work in offline/SQL mode without reflection 

29 instruments_table = sa.Table( 

30 "instruments", 

31 sa.MetaData(), 

32 sa.Column("instrument_pid", sa.String(length=100), primary_key=True), 

33 sa.Column("api_url", sa.String(), unique=True, nullable=False), 

34 sa.Column("calendar_name", sa.String(), nullable=True), 

35 sa.Column("calendar_url", sa.String(), nullable=False), 

36 sa.Column("location", sa.String(length=100), nullable=False), 

37 sa.Column("schema_name", sa.String(), nullable=False), 

38 sa.Column("property_tag", sa.String(length=20), nullable=False), 

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

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

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

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

43 sa.Column("computer_ip", sa.String(length=15), nullable=True), 

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

45 ) 

46 

47 with op.batch_alter_table( 

48 "instruments", schema=None, copy_from=instruments_table 

49 ) as batch_op: 

50 # Drop unused computer fields 

51 batch_op.drop_column("computer_mount") 

52 batch_op.drop_column("computer_ip") 

53 batch_op.drop_column("computer_name") 

54 # Drop unused calendar_name field 

55 batch_op.drop_column("calendar_name") 

56 # Rename schema_name to display_name 

57 batch_op.alter_column("schema_name", new_column_name="display_name") 

58 

59 

60def downgrade() -> None: 

61 """Restore removed fields and revert display_name to schema_name.""" 

62 # Define the table structure for SQL mode support (current state after upgrade) 

63 instruments_table = sa.Table( 

64 "instruments", 

65 sa.MetaData(), 

66 sa.Column("instrument_pid", sa.String(length=100), primary_key=True), 

67 sa.Column("api_url", sa.String(), unique=True, nullable=False), 

68 sa.Column("calendar_url", sa.String(), nullable=False), 

69 sa.Column("location", sa.String(length=100), nullable=False), 

70 sa.Column("display_name", sa.String(), nullable=False), 

71 sa.Column("property_tag", sa.String(length=20), nullable=False), 

72 sa.Column("filestore_path", sa.String(), nullable=False), 

73 sa.Column("harvester", sa.String(), nullable=False), 

74 sa.Column("timezone", sa.String(), nullable=False), 

75 ) 

76 

77 with op.batch_alter_table( 

78 "instruments", schema=None, copy_from=instruments_table 

79 ) as batch_op: 

80 # Restore computer fields (all optional, no unique constraints for simplicity) 

81 batch_op.add_column(sa.Column("computer_name", sa.String(), nullable=True)) 

82 batch_op.add_column( 

83 sa.Column("computer_ip", sa.String(length=15), nullable=True) 

84 ) 

85 batch_op.add_column(sa.Column("computer_mount", sa.String(), nullable=True)) 

86 # Restore calendar_name field (nullable since we have no data to populate it) 

87 batch_op.add_column(sa.Column("calendar_name", sa.String(), nullable=True)) 

88 # Revert display_name to schema_name 

89 batch_op.alter_column("display_name", new_column_name="schema_name")