Manual transform tasks now run via deployments (all of them) & remove the dependency tasks from the list - #1431
Conversation
|
No actionable comments were generated in the recent review. 🎉 ℹ️ Recent review info⚙️ Run configurationConfiguration used: Organization UI Review profile: CHILL Plan: Pro Run ID: 📒 Files selected for processing (1)
WalkthroughManual dbt transform deployments now include ordered prerequisite chains, APIs return primary renderable tasks, a backfill command migrates existing deployments, secret names use persisted values, and organization cleanup handles additional related records. ChangesTransform deployment flow
Estimated code review effort: 4 (Complex) | ~60 minutes Sequence Diagram(s)sequenceDiagram
participant OrgTaskFunctions
participant PipelineService
participant PrefectService
participant OrgDataFlowv1
OrgTaskFunctions->>PipelineService: build prerequisite and primary task chain
OrgTaskFunctions->>PrefectService: create deployment
PrefectService-->>OrgTaskFunctions: return deployment data
OrgTaskFunctions->>OrgDataFlowv1: save ordered task mappings
Possibly related PRs
Suggested reviewers: 🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✨ Finishing Touches📝 Generate docstrings
🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
Codecov Report❌ Patch coverage is Additional details and impacted files@@ Coverage Diff @@
## main #1431 +/- ##
==========================================
+ Coverage 61.67% 61.70% +0.03%
==========================================
Files 152 152
Lines 17779 17813 +34
==========================================
+ Hits 10965 10992 +27
- Misses 6814 6821 +7 ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 3
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
ddpui/core/orchestrate/pipeline_service.py (1)
535-537: 🎯 Functional Correctness | 🟡 Minor | ⚡ Quick winDrop or wire through
payload.ddpui/api/pipeline_api.py:175-189still acceptsTaskParametersand passes them intoPipelineService.run_pipeline, butrun_pipelineignores them entirely. Remove the parameter from the API/service signature or forward it into the run; otherwise client-supplied flags/options are silently discarded.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@ddpui/core/orchestrate/pipeline_service.py` around lines 535 - 537, Update run_pipeline and its caller in pipeline_api.py so the TaskParameters payload is no longer silently ignored: either remove payload from both signatures and calls, or wire it through to the pipeline execution path where client-supplied options are consumed. Keep the API and service signatures consistent.
🧹 Nitpick comments (4)
ddpui/services/org_cleanup_service.py (1)
323-337: 🚀 Performance & Scalability | 🔵 Trivial | ⚡ Quick winOptimize
UserPreferencesdeletion by avoiding N+1 queries.The current implementation iteratively queries and deletes
UserPreferencesfor eachOrgUser, which leads to an N+1 query pattern (executing two queries per user). You can optimize this by bulk-deleting the preferences for the entire organization before iterating over the users. This approach also allows you to accurately log the intended deletion count during adry_run, which is currently skipped.♻️ Proposed refactor
def delete_orgusers(self): """ deletes all org users; first removes UserPreferences rows that FK to each OrgUser (they don't CASCADE) so the OrgUser delete doesn't violate the FK constraint. """ + n_prefs = UserPreferences.objects.filter(orguser__org=self.org).count() + if n_prefs: + logger.info("will delete %s UserPreferences row(s) for org users", n_prefs) + if not self.dry_run: + UserPreferences.objects.filter(orguser__org=self.org).delete() + for orguser in OrgUser.objects.filter(org=self.org): logger.info("will delete orguser %s", orguser.user.email) if not self.dry_run: - n_prefs = UserPreferences.objects.filter(orguser=orguser).count() - if n_prefs: - logger.info( - "deleting %s UserPreferences row(s) attached to orguser %s", - n_prefs, - orguser.user.email, - ) - UserPreferences.objects.filter(orguser=orguser).delete() orguser.delete()🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@ddpui/services/org_cleanup_service.py` around lines 323 - 337, Update the organization user-deletion flow to bulk-query and delete all UserPreferences associated with self.org before iterating through OrgUser records, eliminating the per-user count and delete queries. Preserve informative logging, including the total intended preference count during dry_run, and keep the actual deletion gated by the existing dry_run condition.ddpui/ddpdbt/dbt_service.py (1)
141-150: 📐 Maintainability & Code Quality | 🔵 TrivialDuplicate sanitized-block-name persistence logic (see consolidated comment).
Same check-then-create pattern as
ddpui/api/orgtask_api.py:190-199. Correct behavior, but duplicated — see the consolidated refactor suggestion.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@ddpui/ddpdbt/dbt_service.py` around lines 141 - 150, The sanitized block-name check-and-create logic is duplicated between the dbt service and orgtask API. Extract the shared persistence behavior into a reusable helper, then update the relevant flow around the current OrgPrefectBlockv1.objects.filter/create calls to use it while preserving the org, SECRET block type, and stored_block_name values.ddpui/tests/api_tests/test_orgtask_api.py (1)
193-202: 🎯 Functional Correctness | 🔵 Trivial | ⚡ Quick winFixture doesn't build the actual dependency chain it claims to mirror.
Each manual deployment here maps only the primary task; production chains always prepend git-pull/git-clone + dbt-clean + dbt-deps (see
create_prefect_deployment_for_dbtcore_task). This is enough for the current slug-based assertions, but it means the new cross-chain lock-aggregation logic inget_prefect_transformation_tasks(locks on shared prep tasks surfacing across sibling dataflows) is never actually exercised by these tests.♻️ Suggested fix: seed the full chain per deployment
- if task.slug in ("dbt-run", "dbt-test", "dbt-seed"): - new_dataflow = OrgDataFlowv1.objects.create( - org=org, - name=f"test-{task.slug}-deployment", - deployment_name=f"test-{task.slug}-deployment", - deployment_id=f"test-{task.slug}-deployment-id", - dataflow_type="manual", - ) - - DataflowOrgTask.objects.create( - dataflow=new_dataflow, - orgtask=org_task, - ) + if task.slug in ("dbt-run", "dbt-test", "dbt-seed"): + new_dataflow = OrgDataFlowv1.objects.create( + org=org, + name=f"test-{task.slug}-deployment", + deployment_name=f"test-{task.slug}-deployment", + deployment_id=f"test-{task.slug}-deployment-id", + dataflow_type="manual", + ) + chain_tasks = [ + OrgTask.objects.get(org=org, task__slug="git-pull"), + OrgTask.objects.get(org=org, task__slug="dbt-clean"), + OrgTask.objects.get(org=org, task__slug="dbt-deps"), + org_task, + ] + for idx, chained in enumerate(chain_tasks): + DataflowOrgTask.objects.create(dataflow=new_dataflow, orgtask=chained, seq=idx)🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@ddpui/tests/api_tests/test_orgtask_api.py` around lines 193 - 202, Expand the manual deployment fixture in the task loop to seed the complete dbt dependency chain, not only the primary task: include the production-equivalent git-pull/git-clone, dbt-clean, and dbt-deps tasks linked to each deployment before the LONG_RUNNING dbt task. Reuse the chain construction behavior from create_prefect_deployment_for_dbtcore_task so get_prefect_transformation_tasks exercises shared preparation-task lock aggregation across sibling dataflows.ddpui/api/orgtask_api.py (1)
190-199: 📐 Maintainability & Code Quality | 🔵 TrivialDuplicate sanitized-block-name persistence logic.
Correct fix (using the proxy-returned
block_namefor the exists-check/create, sinceOrgPrefectBlockv1.block_nameis unique), but the identical check-then-create block is copy-pasted inddpui/ddpdbt/dbt_service.py(update_github_pat_storage). Worth extracting into a shared helper (see consolidated note).🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@ddpui/api/orgtask_api.py` around lines 190 - 199, The sanitized block-name persistence check is duplicated between the current flow and dbt_service.update_github_pat_storage. Extract the shared OrgPrefectBlockv1 exists-check/create behavior into a reusable helper, then update both callers to use it with the proxy-returned block_name while preserving the org and SECRET values.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@ddpui/core/orgtaskfunctions.py`:
- Around line 239-244: Wrap the entire DataflowOrgTask creation loop in the
surrounding function with an atomic database transaction so all mappings are
committed together or rolled back on any failure. Update the function containing
the mapped_orgtasks loop, preserving the existing dataflow, orgtask, and seq
values and ensuring no partial rows remain if creation fails.
In `@ddpui/management/commands/backfill_manual_transform_tasks_dependencies.py`:
- Around line 216-219: Make the DataflowOrgTask delete-and-recreate sequence in
the backfill command atomic by wrapping the mapping update around the existing
dataflow update flow in a Django transaction. Ensure any failure in delete or
creation rolls back all mapping changes, while retaining the existing chain
ordering and only committing after the Prefect update and local mapping
recreation succeed.
- Around line 154-202: Move the dry_run guard in the manual dataflow update loop
so it runs before the get_or_create calls for git clone/pull, dbt clean, and dbt
deps OrgTasks. In dry-run mode, report the planned chained-task update without
invoking PipelineService methods that can write to the database; preserve the
existing non-dry-run chain construction and update behavior.
---
Outside diff comments:
In `@ddpui/core/orchestrate/pipeline_service.py`:
- Around line 535-537: Update run_pipeline and its caller in pipeline_api.py so
the TaskParameters payload is no longer silently ignored: either remove payload
from both signatures and calls, or wire it through to the pipeline execution
path where client-supplied options are consumed. Keep the API and service
signatures consistent.
---
Nitpick comments:
In `@ddpui/api/orgtask_api.py`:
- Around line 190-199: The sanitized block-name persistence check is duplicated
between the current flow and dbt_service.update_github_pat_storage. Extract the
shared OrgPrefectBlockv1 exists-check/create behavior into a reusable helper,
then update both callers to use it with the proxy-returned block_name while
preserving the org and SECRET values.
In `@ddpui/ddpdbt/dbt_service.py`:
- Around line 141-150: The sanitized block-name check-and-create logic is
duplicated between the dbt service and orgtask API. Extract the shared
persistence behavior into a reusable helper, then update the relevant flow
around the current OrgPrefectBlockv1.objects.filter/create calls to use it while
preserving the org, SECRET block type, and stored_block_name values.
In `@ddpui/services/org_cleanup_service.py`:
- Around line 323-337: Update the organization user-deletion flow to bulk-query
and delete all UserPreferences associated with self.org before iterating through
OrgUser records, eliminating the per-user count and delete queries. Preserve
informative logging, including the total intended preference count during
dry_run, and keep the actual deletion gated by the existing dry_run condition.
In `@ddpui/tests/api_tests/test_orgtask_api.py`:
- Around line 193-202: Expand the manual deployment fixture in the task loop to
seed the complete dbt dependency chain, not only the primary task: include the
production-equivalent git-pull/git-clone, dbt-clean, and dbt-deps tasks linked
to each deployment before the LONG_RUNNING dbt task. Reuse the chain
construction behavior from create_prefect_deployment_for_dbtcore_task so
get_prefect_transformation_tasks exercises shared preparation-task lock
aggregation across sibling dataflows.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: 3cc78abd-7740-4dfd-870d-5d5c08992f8d
📒 Files selected for processing (9)
ddpui/api/data_api.pyddpui/api/orgtask_api.pyddpui/core/orchestrate/pipeline_service.pyddpui/core/orgtaskfunctions.pyddpui/ddpdbt/dbt_service.pyddpui/management/commands/backfill_manual_transform_tasks_dependencies.pyddpui/services/org_cleanup_service.pyddpui/tests/api_tests/test_orgtask_api.pyddpui/utils/constants.py
| for idx, chained_orgtask in enumerate(mapped_orgtasks): | ||
| DataflowOrgTask.objects.create( | ||
| dataflow=new_dataflow, | ||
| orgtask=chained_orgtask, | ||
| seq=idx, | ||
| ) |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick win
Wrap the multi-row DataflowOrgTask creation in a transaction.
This loop now creates up to 4 rows (was a single .create() before). The Prefect deployment has already been created/pointed at the full chained config by this point; if the loop fails partway (e.g. on the 2nd or 3rd item), the dataflow ends up with an incomplete/incorrect set of mappings while Prefect already expects the full chain — run_pipeline would then compute a wrong org_tasks ordering (or fail with "no org task mapped") for that deployment. This same pattern is duplicated in the backfill command (see consolidated comment).
🛡️ Proposed fix
+from django.db import transaction
...
- for idx, chained_orgtask in enumerate(mapped_orgtasks):
- DataflowOrgTask.objects.create(
- dataflow=new_dataflow,
- orgtask=chained_orgtask,
- seq=idx,
- )
+ with transaction.atomic():
+ for idx, chained_orgtask in enumerate(mapped_orgtasks):
+ DataflowOrgTask.objects.create(
+ dataflow=new_dataflow,
+ orgtask=chained_orgtask,
+ seq=idx,
+ )📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| for idx, chained_orgtask in enumerate(mapped_orgtasks): | |
| DataflowOrgTask.objects.create( | |
| dataflow=new_dataflow, | |
| orgtask=chained_orgtask, | |
| seq=idx, | |
| ) | |
| with transaction.atomic(): | |
| for idx, chained_orgtask in enumerate(mapped_orgtasks): | |
| DataflowOrgTask.objects.create( | |
| dataflow=new_dataflow, | |
| orgtask=chained_orgtask, | |
| seq=idx, | |
| ) |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@ddpui/core/orgtaskfunctions.py` around lines 239 - 244, Wrap the entire
DataflowOrgTask creation loop in the surrounding function with an atomic
database transaction so all mappings are committed together or rolled back on
any failure. Update the function containing the mapped_orgtasks loop, preserving
the existing dataflow, orgtask, and seq values and ensuring no partial rows
remain if creation fails.
| for dataflow in manual_dataflows: | ||
| # primary = last-by-seq mapped orgtask that is a LONG_RUNNING dbt task | ||
| primary = None | ||
| for dfot in sorted( | ||
| dataflow.datafloworgtasks.all(), | ||
| key=lambda d: d.seq, | ||
| reverse=True, | ||
| ): | ||
| ot = dfot.orgtask | ||
| if ot.task.type == TaskType.DBT and ot.task.slug in LONG_RUNNING_TASKS: | ||
| primary = ot | ||
| break | ||
| if primary is None: | ||
| # dbt-cloud-job or orphan — not migratable through this path | ||
| continue | ||
|
|
||
| # build the desired chain | ||
| if is_eks: | ||
| git_orgtask = PipelineService.get_or_create_git_clone_orgtask(org) | ||
| else: | ||
| git_orgtask = PipelineService.get_or_create_git_pull_orgtask(org) | ||
|
|
||
| chain = [ | ||
| git_orgtask, | ||
| PipelineService.get_or_create_dbt_clean_orgtask(org), | ||
| PipelineService.get_or_create_dbt_deps_orgtask(org), | ||
| primary, | ||
| ] | ||
|
|
||
| new_task_configs, err = pipeline_with_orgtasks( | ||
| org, | ||
| chain, | ||
| cli_block=cli_profile_block, | ||
| dbt_project_params=dbt_project_params, | ||
| gitrepo_url=org.dbt.gitrepo_url, | ||
| ) | ||
| if err: | ||
| self.stderr.write(f" [update] {primary.task.slug}: build failed: {err}") | ||
| continue | ||
|
|
||
| new_deployment_params = {"config": {"tasks": new_task_configs, "org_slug": org.slug}} | ||
|
|
||
| if dry_run: | ||
| self.stdout.write( | ||
| f" [DRY] [update] {primary.task.slug} ({dataflow.deployment_id}): " | ||
| f"would set {len(new_task_configs)} chained tasks and rewrite mappings" | ||
| ) | ||
| result["updated"] += 1 | ||
| continue |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟠 Major | ⚡ Quick win
--dry-run still writes to the DB.
PipelineService.get_or_create_git_clone_orgtask / get_or_create_git_pull_orgtask / get_or_create_dbt_clean_orgtask / get_or_create_dbt_deps_orgtask are called unconditionally here, before the if dry_run: check further down (line ~196). These are real get_or_create writes — if the org doesn't already have these prep OrgTask rows, running with --dry-run will create them, contradicting the documented contract ("--dry-run: Print what would happen without touching Prefect or DB") and the help text. Step B (below) and the top-of-function prep block both correctly gate their writes behind the dry-run flag; this block doesn't.
🛡️ Proposed fix — check dry_run before any get_or_create call
if primary is None:
# dbt-cloud-job or orphan — not migratable through this path
continue
+ if dry_run:
+ self.stdout.write(
+ f" [DRY] [update] {primary.task.slug} ({dataflow.deployment_id}): "
+ f"would rewrite chain (git + dbt-clean + dbt-deps + primary) and mappings"
+ )
+ result["updated"] += 1
+ continue
+
# build the desired chain
if is_eks:
git_orgtask = PipelineService.get_or_create_git_clone_orgtask(org)
else:
git_orgtask = PipelineService.get_or_create_git_pull_orgtask(org)
chain = [
git_orgtask,
PipelineService.get_or_create_dbt_clean_orgtask(org),
PipelineService.get_or_create_dbt_deps_orgtask(org),
primary,
]
new_task_configs, err = pipeline_with_orgtasks(...)
if err:
self.stderr.write(f" [update] {primary.task.slug}: build failed: {err}")
continue
new_deployment_params = {"config": {"tasks": new_task_configs, "org_slug": org.slug}}
- if dry_run:
- self.stdout.write(
- f" [DRY] [update] {primary.task.slug} ({dataflow.deployment_id}): "
- f"would set {len(new_task_configs)} chained tasks and rewrite mappings"
- )
- result["updated"] += 1
- continue
-
try:
prefect_service.update_dataflow_v1(...)📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| for dataflow in manual_dataflows: | |
| # primary = last-by-seq mapped orgtask that is a LONG_RUNNING dbt task | |
| primary = None | |
| for dfot in sorted( | |
| dataflow.datafloworgtasks.all(), | |
| key=lambda d: d.seq, | |
| reverse=True, | |
| ): | |
| ot = dfot.orgtask | |
| if ot.task.type == TaskType.DBT and ot.task.slug in LONG_RUNNING_TASKS: | |
| primary = ot | |
| break | |
| if primary is None: | |
| # dbt-cloud-job or orphan — not migratable through this path | |
| continue | |
| # build the desired chain | |
| if is_eks: | |
| git_orgtask = PipelineService.get_or_create_git_clone_orgtask(org) | |
| else: | |
| git_orgtask = PipelineService.get_or_create_git_pull_orgtask(org) | |
| chain = [ | |
| git_orgtask, | |
| PipelineService.get_or_create_dbt_clean_orgtask(org), | |
| PipelineService.get_or_create_dbt_deps_orgtask(org), | |
| primary, | |
| ] | |
| new_task_configs, err = pipeline_with_orgtasks( | |
| org, | |
| chain, | |
| cli_block=cli_profile_block, | |
| dbt_project_params=dbt_project_params, | |
| gitrepo_url=org.dbt.gitrepo_url, | |
| ) | |
| if err: | |
| self.stderr.write(f" [update] {primary.task.slug}: build failed: {err}") | |
| continue | |
| new_deployment_params = {"config": {"tasks": new_task_configs, "org_slug": org.slug}} | |
| if dry_run: | |
| self.stdout.write( | |
| f" [DRY] [update] {primary.task.slug} ({dataflow.deployment_id}): " | |
| f"would set {len(new_task_configs)} chained tasks and rewrite mappings" | |
| ) | |
| result["updated"] += 1 | |
| continue | |
| for dataflow in manual_dataflows: | |
| # primary = last-by-seq mapped orgtask that is a LONG_RUNNING dbt task | |
| primary = None | |
| for dfot in sorted( | |
| dataflow.datafloworgtasks.all(), | |
| key=lambda d: d.seq, | |
| reverse=True, | |
| ): | |
| ot = dfot.orgtask | |
| if ot.task.type == TaskType.DBT and ot.task.slug in LONG_RUNNING_TASKS: | |
| primary = ot | |
| break | |
| if primary is None: | |
| # dbt-cloud-job or orphan — not migratable through this path | |
| continue | |
| if dry_run: | |
| self.stdout.write( | |
| f" [DRY] [update] {primary.task.slug} ({dataflow.deployment_id}): " | |
| f"would rewrite chain (git + dbt-clean + dbt-deps + primary) and mappings" | |
| ) | |
| result["updated"] += 1 | |
| continue | |
| # build the desired chain | |
| if is_eks: | |
| git_orgtask = PipelineService.get_or_create_git_clone_orgtask(org) | |
| else: | |
| git_orgtask = PipelineService.get_or_create_git_pull_orgtask(org) | |
| chain = [ | |
| git_orgtask, | |
| PipelineService.get_or_create_dbt_clean_orgtask(org), | |
| PipelineService.get_or_create_dbt_deps_orgtask(org), | |
| primary, | |
| ] | |
| new_task_configs, err = pipeline_with_orgtasks( | |
| org, | |
| chain, | |
| cli_block=cli_profile_block, | |
| dbt_project_params=dbt_project_params, | |
| gitrepo_url=org.dbt.gitrepo_url, | |
| ) | |
| if err: | |
| self.stderr.write(f" [update] {primary.task.slug}: build failed: {err}") | |
| continue | |
| new_deployment_params = {"config": {"tasks": new_task_configs, "org_slug": org.slug}} |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@ddpui/management/commands/backfill_manual_transform_tasks_dependencies.py`
around lines 154 - 202, Move the dry_run guard in the manual dataflow update
loop so it runs before the get_or_create calls for git clone/pull, dbt clean,
and dbt deps OrgTasks. In dry-run mode, report the planned chained-task update
without invoking PipelineService methods that can write to the database;
preserve the existing non-dry-run chain construction and update behavior.
| # wipe + recreate DataflowOrgTask mappings so seq matches chain order | ||
| DataflowOrgTask.objects.filter(dataflow=dataflow).delete() | ||
| for idx, orgtask in enumerate(chain): | ||
| DataflowOrgTask.objects.create(dataflow=dataflow, orgtask=orgtask, seq=idx) |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick win
Wipe+recreate of DataflowOrgTask mappings isn't atomic.
This happens after prefect_service.update_dataflow_v1 already succeeded. If the delete or one of the recreate calls fails partway through, the deployment on Prefect now expects the new chained config while Django's local mapping table is left incomplete — see consolidated comment (shared root cause with ddpui/core/orgtaskfunctions.py).
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@ddpui/management/commands/backfill_manual_transform_tasks_dependencies.py`
around lines 216 - 219, Make the DataflowOrgTask delete-and-recreate sequence in
the backfill command atomic by wrapping the mapping update around the existing
dataflow update flow in a Django transaction. Ensure any failure in delete or
creation rolls back all mapping changes, while retaining the existing chain
ordering and only committing after the Prefect update and local mapping
recreation succeed.
2 participants
Summary by CodeRabbit