Repository: apache/airflow
File: airflow-core/src/airflow/models/asset.py (Python)

Code before:
    *,
    name: str | None = None,
    uri: str | None = None,
    session: Session,
) -> AssetModel | None:
    if name is None and uri is None:
        raise TypeError("either name or uri is required")
    stmt = select(AssetModel).where(AssetModel.active.has())
    if name is not None:
        stmt = stmt.where(AssetModel.name == name)
    if uri is not None:
        stmt = stmt.where(AssetModel.uri == uri)
    return session.scalar(stmt)


def remove_references_to_deleted_dags(session: Session):
    from airflow.models.dag import DagModel

    models_to_check: list[Any] = [
        DagScheduleAssetReference,
        DagScheduleAssetNameReference,
        DagScheduleAssetUriReference,
        DagScheduleAssetAliasReference,
        TaskOutletAssetReference,
    ]

    if stale_dag_ids := session.scalars(select(DagModel.dag_id).where(DagModel.is_stale)).all():
        for model in models_to_check:
            session.execute(
                delete(model)
                .where(model.dag_id.in_(stale_dag_ids))
                .execution_options(synchronize_session="fetch")
            )


alias_association_table = Table(
    "asset_alias_asset",
    Base.metadata,
    Column("alias_id", ForeignKey("asset_alias.id", ondelete="CASCADE"), primary_key=True),
    Column("asset_id", ForeignKey("asset.id", ondelete="CASCADE"), primary_key=True),
    Index("idx_asset_alias_asset_alias_id", "alias_id"),
    Index("idx_asset_alias_asset_asset_id", "asset_id"),
)

asset_alias_asset_event_association_table = Table(
    "asset_alias_asset_event",
    Base.metadata,
    Column("alias_id", ForeignKey("asset_alias.id", ondelete="CASCADE"), primary_key=True),
    Column("event_id", ForeignKey("asset_event.id", ondelete="CASCADE"), primary_key=True),
    Index("idx_asset_alias_asset_event_alias_id", "alias_id"),
    Index("idx_asset_alias_asset_event_event_id", "event_id"),

Review comment (bug):
```suggestion

# The queries need to be done in separate steps, because in case of multiple dag processors on
# MySQL there could be a deadlock caused by acquiring both exclusive lock for deletion and
# shared lock for query in reverse sequence
```

Code after:
    *,
    name: str | None = None,
    uri: str | None = None,
    session: Session,
) -> AssetModel | None:
    if name is None and uri is None:
        raise TypeError("either name or uri is required")
    stmt = select(AssetModel).where(AssetModel.active.has())
    if name is not None:
        stmt = stmt.where(AssetModel.name == name)
    if uri is not None:
        stmt = stmt.where(AssetModel.uri == uri)
    return session.scalar(stmt)


def remove_references_to_deleted_dags(session: Session):
    from airflow.models.dag import DagModel

    models_to_check: list[Any] = [
        DagScheduleAssetReference,
        DagScheduleAssetNameReference,
        DagScheduleAssetUriReference,
        DagScheduleAssetAliasReference,
        TaskOutletAssetReference,
    ]

    # The queries need to be done in separate steps, because in the case of multiple
    # dag processors on MySQL, there could be a deadlock caused by acquiring both an
    # exclusive lock for deletion and shared lock for query in reverse sequence
    if stale_dag_ids := session.scalars(select(DagModel.dag_id).where(DagModel.is_stale)).all():
        for model in models_to_check:
            session.execute(
                delete(model)
                .where(model.dag_id.in_(stale_dag_ids))
                .execution_options(synchronize_session="fetch")
            )


alias_association_table = Table(
    "asset_alias_asset",
    Base.metadata,
    Column("alias_id", ForeignKey("asset_alias.id", ondelete="CASCADE"), primary_key=True),
    Column("asset_id", ForeignKey("asset.id", ondelete="CASCADE"), primary_key=True),
    Index("idx_asset_alias_asset_alias_id", "alias_id"),
    Index("idx_asset_alias_asset_asset_id", "asset_id"),
)

asset_alias_asset_event_association_table = Table(
    "asset_alias_asset_event",
    Base.metadata,
    Column("alias_id", ForeignKey("asset_alias.id", ondelete="CASCADE"), primary_key=True),