Coverage for /usr/local/lib/python3.12/site-packages/prefect/server/database/_migrations/versions/sqlite/2022_07_11_113314_638cbcc2a158_add_infrastructure_block_id_to_.py: 74%

23 statements  

« prev     ^ index     » next       coverage.py v7.10.6, created at 2025-12-05 13:38 +0000

1"""Add infrastructure block id to deployments 

2 

3Revision ID: 638cbcc2a158 

4Revises: 061c7e518b40 

5Create Date: 2022-07-11 11:33:14.404779 

6 

7""" 

8 

9import sqlalchemy as sa 1a

10from alembic import op 1a

11 

12import prefect 1a

13 

14# revision identifiers, used by Alembic. 

15revision = "638cbcc2a158" 1a

16down_revision = "061c7e518b40" 1a

17branch_labels = None 1a

18depends_on = None 1a

19 

20 

21def upgrade(): 1a

22 op.execute("PRAGMA foreign_keys=OFF") 1a

23 

24 with op.batch_alter_table("deployment", schema=None) as batch_op: 1a

25 batch_op.add_column( 1a

26 sa.Column( 

27 "infrastructure_document_id", 

28 prefect.server.utilities.database.UUID(), 

29 nullable=True, 

30 ) 

31 ) 

32 batch_op.create_foreign_key( 1a

33 batch_op.f("fk_deployment__infrastructure_document_id__block_document"), 

34 "block_document", 

35 ["infrastructure_document_id"], 

36 ["id"], 

37 ondelete="CASCADE", 

38 ) 

39 

40 op.execute("PRAGMA foreign_keys=ON") 1a

41 

42 with op.batch_alter_table("flow_run", schema=None) as batch_op: 1a

43 batch_op.add_column( 1a

44 sa.Column( 

45 "infrastructure_document_id", 

46 prefect.server.utilities.database.UUID(), 

47 nullable=True, 

48 ) 

49 ) 

50 batch_op.create_foreign_key( 1a

51 batch_op.f("fk_flow_run__infrastructure_document_id__block_document"), 

52 "block_document", 

53 ["infrastructure_document_id"], 

54 ["id"], 

55 ondelete="CASCADE", 

56 ) 

57 

58 

59def downgrade(): 1a

60 with op.batch_alter_table("deployment", schema=None) as batch_op: 

61 batch_op.drop_constraint( 

62 batch_op.f("fk_deployment__infrastructure_document_id__block_document"), 

63 type_="foreignkey", 

64 ) 

65 batch_op.drop_column("infrastructure_document_id") 

66 

67 with op.batch_alter_table("flow_run", schema=None) as batch_op: 

68 batch_op.drop_constraint( 

69 batch_op.f("fk_flow_run__infrastructure_document_id__block_document"), 

70 type_="foreignkey", 

71 ) 

72 batch_op.drop_column("infrastructure_document_id")