"roots",
            "leaves",
            
            "upstream_list",
            "downstream_list",
            
            "global_operator_extra_link_dict",
            "operator_extra_link_dict",
        )
    )
    def get_dag(self) -> DAG | None:
        raise NotImplementedError()
    @property
    def task_type(self) -> str:
        raise NotImplementedError()
    @property
    def operator_name(self) -> str:
        raise NotImplementedError()
    @property
    def inherits_from_empty_operator(self) -> bool:
        raise NotImplementedError()
    @property
    def dag_id(self) -> str:
        """Returns dag id if it has one or an adhoc + owner."""
        dag = self.get_dag()
        if dag:
            return dag.dag_id
        return f"adhoc_{self.owner}"
    @property
    def node_id(self) -> str: