source=getattr(task, "external_dag_id"),
                    target=task.dag_id,
                    dependency_type="sensor",
                    dependency_id=task.task_id,
                )
            )
        for obj in task.outlets or []:
            if isinstance(obj, Dataset):
                deps.append(
                    DagDependency(
                        source=task.dag_id,
                        target="dataset",
                        dependency_type="dataset",
                        dependency_id=obj.uri,
                    )
                )
        return deps
    @staticmethod
    def detect_dag_dependencies(dag: DAG | None) -> Iterable[DagDependency]:
        """Detects dependencies set directly on the DAG object."""
        if not dag:
            return
        for x in dag.dataset_triggers:
            yield DagDependency(
                source="dataset",
                target=dag.dag_id,
                dependency_type="dataset",
                dependency_id=x.uri,
            )
class SerializedBaseOperator(BaseOperator, BaseSerialization):