Skip to content

Remove unreachable AssetManager slow-path helper#70996

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
takayoshi-makabe wants to merge 1 commit into from
Closed

Remove unreachable AssetManager slow-path helper #70996

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
25 changes: 0 additions & 25 deletions airflow-core/src/airflow/assets/manager.py
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -826,31 +826,6 @@ def _get_or_create_apdr(
)
return apdr

@classmethod
def _queue_dagruns_nonpartitioned_slow_path(
cls, asset_id: int, dags_to_queue: set[DagModel], event: AssetEvent, session: Session
) -> None:
def _queue_dagrun_if_needed(dag: DagModel) -> str | None:
item = AssetDagRunQueue(target_dag_id=dag.dag_id, asset_id=asset_id, created_at=event.timestamp)
# Don't error whole transaction when a single RunQueue item conflicts.
# https://docs.sqlalchemy.org/en/20/orm/session_transaction.html#using-savepoint
try:
with session.begin_nested():
existing = session.get(
AssetDagRunQueue, {"target_dag_id": dag.dag_id, "asset_id": asset_id}
)
if existing and existing.created_at >= event.timestamp:
cls.logger().debug("Skipping record %s due to newer timestamp", item)
return dag.dag_id # already queued with a newer timestamp
session.merge(item)
except exc.IntegrityError:
cls.logger().debug("Skipping record %s", item, exc_info=True)
return dag.dag_id

queued_results = (_queue_dagrun_if_needed(dag) for dag in dags_to_queue)
if queued_dag_ids := [r for r in queued_results if r is not None]:
cls.logger().debug("consuming dag ids %s", queued_dag_ids)

@classmethod
def _queue_dagruns_nonpartitioned_mysql(
cls, asset_id: int, dags_to_queue: set[DagModel], event: AssetEvent, session: Session
Expand Down