|
11 | 11 | from sqlalchemy import ( |
12 | 12 | Column, DateTime, Float, Integer, MetaData, String, Table, Text, and_, case, |
13 | 13 | create_engine, delete, inspect, insert, literal, null, select, text, update, |
14 | | - or_, |
| 14 | + or_, event as sqlalchemy_event, |
15 | 15 | ) |
16 | 16 |
|
17 | 17 | from ..transform import ( |
|
29 | 29 | from .normative import catalog as core_catalog |
30 | 30 | from .profiles import SqlProfile, preflight |
31 | 31 | from .wide import physical_name, require_verified_catalog |
32 | | -from .workflow import TransformationError |
| 32 | +from .workflow import TransformationError, workflow_catalog |
33 | 33 |
|
34 | 34 |
|
35 | 35 | APPLY_CONTRACT = "openstatspec-in-place-transformation-v0.2" |
@@ -278,15 +278,21 @@ def _replace_value_labels( |
278 | 278 | ) |
279 | 279 | ).scalar_one_or_none() |
280 | 280 | if old_set is not None: |
281 | | - connection.execute(delete(core.value_label).where( |
282 | | - core.value_label.c.value_label_set_id == old_set |
283 | | - )) |
284 | 281 | connection.execute(delete(core.variable_value_label_set).where( |
285 | 282 | core.variable_value_label_set.c.variable_id == variable_id |
286 | 283 | )) |
287 | | - connection.execute(delete(core.value_label_set).where( |
288 | | - core.value_label_set.c.value_label_set_id == old_set |
289 | | - )) |
| 284 | + still_used = connection.execute( |
| 285 | + select(core.variable_value_label_set.c.variable_id).where( |
| 286 | + core.variable_value_label_set.c.value_label_set_id == old_set |
| 287 | + ) |
| 288 | + ).first() |
| 289 | + if still_used is None: |
| 290 | + connection.execute(delete(core.value_label).where( |
| 291 | + core.value_label.c.value_label_set_id == old_set |
| 292 | + )) |
| 293 | + connection.execute(delete(core.value_label_set).where( |
| 294 | + core.value_label_set.c.value_label_set_id == old_set |
| 295 | + )) |
290 | 296 | label_set_id = str(uuid4()) |
291 | 297 | connection.execute(insert(core.value_label_set).values( |
292 | 298 | value_label_set_id=label_set_id, |
@@ -476,6 +482,26 @@ def _apply_plan_on_connection( |
476 | 482 | table_name = _physical_table_name(dataset) |
477 | 483 | plan = submission.plan |
478 | 484 | bound = bind_transformation_plan(plan, schema) |
| 485 | + deleted_names = { |
| 486 | + operation.variable.casefold() for operation in plan.operations |
| 487 | + if isinstance(operation, DeleteVariableOperation) |
| 488 | + } |
| 489 | + original_deletions = { |
| 490 | + str(row["variable_id"]): str(row["source_name"]) for row in variables |
| 491 | + if str(row["source_name"]).casefold() in deleted_names |
| 492 | + } |
| 493 | + if original_deletions and inspect(connection).has_table("derived_variable_lineage"): |
| 494 | + lineage = workflow_catalog(MetaData()).derived_variable_lineage |
| 495 | + blocked_id = connection.execute( |
| 496 | + select(lineage.c.core_variable_id).where( |
| 497 | + lineage.c.core_variable_id.in_(original_deletions) |
| 498 | + ).limit(1) |
| 499 | + ).scalar_one_or_none() |
| 500 | + if blocked_id is not None: |
| 501 | + raise TransformationError( |
| 502 | + "variable_has_dependents", |
| 503 | + f"Variable {original_deletions[blocked_id]!r} is referenced by published lineage.", |
| 504 | + ) |
479 | 505 | output_used_physical = {"__case_ordinal"} |
480 | 506 | output_variables = [ |
481 | 507 | { |
@@ -972,6 +998,17 @@ def _run_in_place_submission( |
972 | 998 | sqlite_version = (*sqlite_version_parts, 0, 0)[:3] |
973 | 999 | allow_delete_variable = sqlite_version >= (3, 35, 0) |
974 | 1000 | engine = create_engine(database_url) |
| 1001 | + if profile.name == "sqlite": |
| 1002 | + @sqlalchemy_event.listens_for(engine, "connect") |
| 1003 | + def enable_sqlite_foreign_keys(dbapi_connection, _connection_record): |
| 1004 | + cursor = dbapi_connection.cursor() |
| 1005 | + try: |
| 1006 | + cursor.execute("PRAGMA foreign_keys = ON") |
| 1007 | + if cursor.execute("PRAGMA foreign_keys").fetchone() != (1,): |
| 1008 | + raise RuntimeError("SQLite foreign-key enforcement could not be enabled.") |
| 1009 | + finally: |
| 1010 | + cursor.close() |
| 1011 | + |
975 | 1012 | journal: dict[str, Any] = {} |
976 | 1013 | branch: str | None = None |
977 | 1014 | head: str | None = None |
|
0 commit comments