« prev ^ index » next coverage.py v7.10.6, created at 2025-12-05 10:48 +0000
1"""Add eplace version with checksum
3Revision ID: b75d279ba985
4Revises: fd966d4ad99c
5Create Date: 2022-04-25 13:52:07.514013
7"""
9import sqlalchemy as sa 1a
10from alembic import op 1a
12import prefect 1a
13from prefect.blocks.core import Block 1a
15# revision identifiers, used by Alembic.
16revision = "b75d279ba985" 1a
17down_revision = "fd966d4ad99c" 1a
18branch_labels = None 1a
19depends_on = None 1a
22def upgrade(): 1a
23 # ### commands auto generated by Alembic - please adjust! ###
24 op.create_table( 1a
25 "block_type",
26 sa.Column(
27 "id",
28 prefect.server.utilities.database.UUID(),
29 server_default=sa.text(
30 "(\n (\n lower(hex(randomblob(4))) \n || '-' \n "
31 " || lower(hex(randomblob(2))) \n || '-4' \n ||"
32 " substr(lower(hex(randomblob(2))),2) \n || '-' \n ||"
33 " substr('89ab',abs(random()) % 4 + 1, 1) \n ||"
34 " substr(lower(hex(randomblob(2))),2) \n || '-' \n ||"
35 " lower(hex(randomblob(6)))\n )\n )"
36 ),
37 nullable=False,
38 ),
39 sa.Column(
40 "created",
41 prefect.server.utilities.database.Timestamp(timezone=True),
42 server_default=sa.text("(strftime('%Y-%m-%d %H:%M:%f000', 'now'))"),
43 nullable=False,
44 ),
45 sa.Column(
46 "updated",
47 prefect.server.utilities.database.Timestamp(timezone=True),
48 server_default=sa.text("(strftime('%Y-%m-%d %H:%M:%f000', 'now'))"),
49 nullable=False,
50 ),
51 sa.Column("name", sa.String(), nullable=False),
52 sa.Column("logo_url", sa.String(), nullable=True),
53 sa.Column("documentation_url", sa.String(), nullable=True),
54 sa.PrimaryKeyConstraint("id", name=op.f("pk_block_type")),
55 )
56 with op.batch_alter_table("block_type", schema=None) as batch_op: 1a
57 batch_op.create_index( 1a
58 batch_op.f("ix_block_type__updated"), ["updated"], unique=False
59 )
60 batch_op.create_index(batch_op.f("uq_block_type__name"), ["name"], unique=True) 1a
62 with op.batch_alter_table("block_document", schema=None) as batch_op: 1a
63 batch_op.add_column( 1a
64 sa.Column(
65 "block_type_id", prefect.server.utilities.database.UUID(), nullable=True
66 )
67 )
68 batch_op.drop_index("uq_block__schema_id_name") 1a
69 batch_op.create_index( 1a
70 "uq_block__type_id_name", ["block_type_id", "name"], unique=True
71 )
72 batch_op.create_foreign_key( 1a
73 batch_op.f("fk_block_document__block_type_id__block_type"),
74 "block_type",
75 ["block_type_id"],
76 ["id"],
77 ondelete="cascade",
78 )
79 batch_op.drop_constraint("fk_block__block_schema_id__block_schema") 1a
81 with op.batch_alter_table( 1a
82 "block_schema",
83 schema=None,
84 ) as batch_op:
85 batch_op.add_column( 1a
86 sa.Column(
87 "block_type_id", prefect.server.utilities.database.UUID(), nullable=True
88 )
89 )
90 batch_op.create_foreign_key( 1a
91 batch_op.f("fk_block_schema__block_type_id__block_type"),
92 "block_type",
93 ["block_type_id"],
94 ["id"],
95 ondelete="cascade",
96 )
97 batch_op.drop_column("version") 1a
98 batch_op.add_column(sa.Column("checksum", sa.String(), nullable=True)) 1a
99 batch_op.drop_index("uq_block_schema__name_version") 1a
100 batch_op.create_index( 1a
101 batch_op.f("ix_block_schema__checksum"), ["checksum"], unique=False
102 )
103 batch_op.create_index("uq_block_schema__checksum", ["checksum"], unique=True) 1a
105 # Add checksums and block types for existing block schemas
106 connection = op.get_bind() 1a
107 meta_data = sa.MetaData() 1a
108 meta_data.reflect(connection) 1a
109 BLOCK_SCHEMA = meta_data.tables["block_schema"] 1a
110 BLOCK_TYPE = meta_data.tables["block_type"] 1a
111 BLOCK_DOCUMENT = meta_data.tables["block_document"] 1a
112 results = connection.execute( 1a
113 sa.select(BLOCK_SCHEMA.c.id, BLOCK_SCHEMA.c.name, BLOCK_SCHEMA.c.fields)
114 )
115 for id, name, fields in results: 115 ↛ 116line 115 didn't jump to line 116 because the loop on line 115 never started1a
116 schema_checksum = Block._calculate_schema_checksum(fields)
117 # Add checksum
118 connection.execute(
119 sa.update(BLOCK_SCHEMA)
120 .where(BLOCK_SCHEMA.c.id == id)
121 .values(checksum=schema_checksum)
122 )
123 # Create corresponding block type
124 block_type_result = connection.execute(
125 sa.select(BLOCK_TYPE.c.id).where(BLOCK_TYPE.c.name == name)
126 ).first()
127 if block_type_result is None:
128 # Create block type if it doesn't already exist
129 connection.execute(sa.insert(BLOCK_TYPE).values(name=name))
130 block_type_result = connection.execute(
131 sa.select(BLOCK_TYPE.c.id).where(BLOCK_TYPE.c.name == name)
132 ).first()
133 new_block_type_id = block_type_result[0]
134 connection.execute(
135 sa.update(BLOCK_SCHEMA)
136 .where(BLOCK_SCHEMA.c.id == id)
137 .values(block_type_id=new_block_type_id)
138 )
139 # Associate new block type will all block documents for this block schema
140 block_document_results = connection.execute(
141 sa.select(BLOCK_DOCUMENT.c.id).where(BLOCK_DOCUMENT.c.block_schema_id == id)
142 ).all()
143 for (block_document_id,) in block_document_results:
144 connection.execute(
145 sa.update(BLOCK_DOCUMENT)
146 .where(BLOCK_DOCUMENT.c.id == block_document_id)
147 .values(block_type_id=new_block_type_id)
148 )
150 with op.batch_alter_table( 1a
151 "block_schema",
152 schema=None,
153 ) as batch_op:
154 batch_op.drop_column("name") 1a
155 batch_op.alter_column("checksum", existing_type=sa.VARCHAR(), nullable=False) 1a
156 batch_op.alter_column( 1a
157 "block_type_id", existing_type=sa.VARCHAR(), nullable=False
158 )
160 with op.batch_alter_table( 1a
161 "block_document",
162 schema=None,
163 ) as batch_op:
164 batch_op.alter_column( 1a
165 "block_type_id", existing_type=sa.VARCHAR(), nullable=False
166 )
168 batch_op.create_foreign_key( 1a
169 batch_op.f("fk_block__block_schema_id__block_schema"),
170 "block_schema",
171 ["block_schema_id"],
172 ["id"],
173 ondelete="cascade",
174 )
175 # ### end Alembic commands ###
178def downgrade(): 1a
179 # ### commands auto generated by Alembic - please adjust! ###
180 with op.batch_alter_table("block_schema", schema=None) as batch_op:
181 batch_op.drop_index("uq_block_schema__checksum")
182 batch_op.drop_index(batch_op.f("ix_block_schema__checksum"))
183 batch_op.create_index(
184 "uq_block_schema__name_version", ["name", "version"], unique=False
185 )
186 batch_op.drop_column("checksum")
187 batch_op.add_column(sa.Column("version", sa.VARCHAR(), nullable=True))
188 batch_op.add_column(sa.Column("name", sa.VARCHAR(), nullable=True))
189 batch_op.drop_constraint(
190 batch_op.f("fk_block_schema__block_type_id__block_type"), type_="foreignkey"
191 )
192 batch_op.drop_column("block_type_id")
194 with op.batch_alter_table("block_document", schema=None) as batch_op:
195 batch_op.drop_constraint(
196 batch_op.f("fk_block_document__block_type_id__block_type"),
197 type_="foreignkey",
198 )
199 batch_op.drop_index("uq_block__type_id_name")
200 batch_op.create_index(
201 "uq_block__schema_id_name", ["block_schema_id", "name"], unique=False
202 )
203 batch_op.drop_column("block_type_id")
205 with op.batch_alter_table("block_type", schema=None) as batch_op:
206 batch_op.drop_index(batch_op.f("ix_block_type__updated"))
207 batch_op.drop_index(batch_op.f("uq_block_type__name"))
209 op.drop_table("block_type")
211 # ### end Alembic commands ###