airflow/models/dag.py
def create_dagrun(
        self,
        state: DagRunState,
        execution_date: Optional[datetime] = None,
        run_id: Optional[str] = None,
        start_date: Optional[datetime] = None,
        external_trigger: Optional[bool] = False,
        conf: Optional[dict] = None,
        run_type: Optional[DagRunType] = None,
        session=None,
        dag_hash: Optional[str] = None,
        creating_job_id: Optional[int] = None,
        data_interval: Optional[Tuple[datetime, datetime]] = None,
    ):
        """
        Creates a dag run from this dag including the tasks associated with this dag.
        Returns the dag run.

        :param run_id: defines the run id for this dag run
        :type run_id: str
        :param run_type: type of DagRun
        :type run_type: airflow.utils.types.DagRunType
        :param execution_date: the execution date of this dag run
        :type execution_date: datetime.datetime
        :param state: the state of the dag run
        :type state: airflow.utils.state.DagRunState
        :param start_date: the date this dag run should be evaluated
        :type start_date: datetime
        :param external_trigger: whether this dag run is externally triggered
        :type external_trigger: bool
        :param conf: Dict containing configuration/parameters to pass to the DAG
        :type conf: dict
        :param creating_job_id: id of the job creating this DagRun
        :type creating_job_id: int
        :param session: database session
        :type session: sqlalchemy.orm.session.Session
        :param dag_hash: Hash of Serialized DAG
        :type dag_hash: str
        :param data_interval: Data interval of the DagRun
        :type data_interval: tuple[datetime, datetime] | None
        """
        if run_id:  # Infer run_type from run_id if needed.
            if not isinstance(run_id, str):
                raise ValueError(f"`run_id` expected to be a str is {type(run_id)}")
            if not run_type:
                run_type = DagRunType.from_run_id(run_id)
        elif run_type and execution_date is not None:  # Generate run_id from run_type and execution_date.
            if not isinstance(run_type, DagRunType):
                raise ValueError(f"`run_type` expected to be a DagRunType is {type(run_type)}")
            run_id = DagRun.generate_run_id(run_type, execution_date)
        else:
            raise AirflowException(
                "Creating DagRun needs either `run_id` or both `run_type` and `execution_date`"
            )

        logical_date = timezone.coerce_datetime(execution_date)
        if data_interval is None and logical_date is not None:
            warnings.warn(
                "Calling `DAG.create_dagrun()` without an explicit data interval is deprecated",
                DeprecationWarning,
                stacklevel=3,
            )
            if run_type == DagRunType.MANUAL:
                data_interval = self.timetable.infer_manual_data_interval(run_after=logical_date)
            else:
                data_interval = self.infer_automated_data_interval(logical_date)

        # create a copy of params before validating
        copied_params = copy.deepcopy(self.params)
        copied_params.update(conf or {})
        copied_params.validate()

        run = DagRun(
            dag_id=self.dag_id,
            run_id=run_id,
            execution_date=logical_date,
            start_date=start_date,
            external_trigger=external_trigger,
            conf=conf,
            state=state,
            run_type=run_type,
            dag_hash=dag_hash,
            creating_job_id=creating_job_id,
            data_interval=data_interval,
        )
        session.add(run)
        session.flush()

        run.dag = self

        # create the associated task instances
        # state is None at the moment of creation
        run.verify_integrity(session=session)

        return run

        Creates a dag run from this dag including the tasks associated with this dag.
        Returns the dag run.

        :param run_id: defines the run id for this dag run
        :type run_id: str
        :param run_type: type of DagRun
        :type run_type: airflow.utils.types.DagRunType
        :param execution_date: the execution date of this dag run
        :type execution_date: datetime.datetime
        :param state: the state of the dag run
        :type state: airflow.utils.state.DagRunState
        :param start_date: the date this dag run should be evaluated
        :type start_date: datetime
        :param external_trigger: whether this dag run is externally triggered
        :type external_trigger: bool
        :param conf: Dict containing configuration/parameters to pass to the DAG
        :type conf: dict
        :param creating_job_id: id of the job creating this DagRun
        :type creating_job_id: int
        :param session: database session
        :type session: sqlalchemy.orm.session.Session
        :param dag_hash: Hash of Serialized DAG
        :type dag_hash: str
        :param data_interval: Data interval of the DagRun
        :type data_interval: tuple[datetime, datetime] | None
        
Creates a dag run from this dag including the tasks associated with this dag.
Returns the dag run.
Creates a dag run from this dag including the tasks associated with this dag . Returns the dag run .
def create_dagrun(
        self,
        state: DagRunState,
        execution_date: Optional[datetime] = None,
        run_id: Optional[str] = None,
        start_date: Optional[datetime] = None,
        external_trigger: Optional[bool] = False,
        conf: Optional[dict] = None,
        run_type: Optional[DagRunType] = None,
        session=None,
        dag_hash: Optional[str] = None,
        creating_job_id: Optional[int] = None,
        data_interval: Optional[Tuple[datetime, datetime]] = None,
    ):
        if run_id:  
            if not isinstance(run_id, str):
                raise ValueError(f"`run_id` expected to be a str is {type(run_id)}")
            if not run_type:
                run_type = DagRunType.from_run_id(run_id)
        elif run_type and execution_date is not None:  
            if not isinstance(run_type, DagRunType):
                raise ValueError(f"`run_type` expected to be a DagRunType is {type(run_type)}")
            run_id = DagRun.generate_run_id(run_type, execution_date)
        else:
            raise AirflowException(
                "Creating DagRun needs either `run_id` or both `run_type` and `execution_date`"
            )
        logical_date = timezone.coerce_datetime(execution_date)
        if data_interval is None and logical_date is not None:
            warnings.warn(
                "Calling `DAG.create_dagrun()` without an explicit data interval is deprecated",
                DeprecationWarning,
                stacklevel=3,
            )
            if run_type == DagRunType.MANUAL:
                data_interval = self.timetable.infer_manual_data_interval(run_after=logical_date)
            else:
                data_interval = self.infer_automated_data_interval(logical_date)
        copied_params = copy.deepcopy(self.params)
        copied_params.update(conf or {})
        copied_params.validate()
        run = DagRun(
            dag_id=self.dag_id,
            run_id=run_id,
            execution_date=logical_date,
            start_date=start_date,
            external_trigger=external_trigger,
            conf=conf,
            state=state,
            run_type=run_type,
            dag_hash=dag_hash,
            creating_job_id=creating_job_id,
            data_interval=data_interval,
        )
        session.add(run)
        session.flush()
        run.dag = self
        run.verify_integrity(session=session)
        return run
def create_dagrun ( self , state : DagRunState , execution_date : Optional [ datetime ] = None , run_id : Optional [ str ] = None , start_date : Optional [ datetime ] = None , external_trigger : Optional [ bool ] = False , conf : Optional [ dict ] = None , run_type : Optional [ DagRunType ] = None , session = None , dag_hash : Optional [ str ] = None , creating_job_id : Optional [ int ] = None , data_interval : Optional [ Tuple [ datetime , datetime ] ] = None , ) : if run_id : if not isinstance ( run_id , str ) : raise ValueError ( f"`run_id` expected to be a str is {type(run_id)}" ) if not run_type : run_type = DagRunType . from_run_id ( run_id ) elif run_type and execution_date is not None : if not isinstance ( run_type , DagRunType ) : raise ValueError ( f"`run_type` expected to be a DagRunType is {type(run_type)}" ) run_id = DagRun . generate_run_id ( run_type , execution_date ) else : raise AirflowException ( "Creating DagRun needs either `run_id` or both `run_type` and `execution_date`" ) logical_date = timezone . coerce_datetime ( execution_date ) if data_interval is None and logical_date is not None : warnings . warn ( "Calling `DAG.create_dagrun()` without an explicit data interval is deprecated" , DeprecationWarning , stacklevel = 3 , ) if run_type == DagRunType . MANUAL : data_interval = self . timetable . infer_manual_data_interval ( run_after = logical_date ) else : data_interval = self . infer_automated_data_interval ( logical_date ) copied_params = copy . deepcopy ( self . params ) copied_params . update ( conf or { } ) copied_params . validate ( ) run = DagRun ( dag_id = self . dag_id , run_id = run_id , execution_date = logical_date , start_date = start_date , external_trigger = external_trigger , conf = conf , state = state , run_type = run_type , dag_hash = dag_hash , creating_job_id = creating_job_id , data_interval = data_interval , ) session . add ( run ) session . flush ( ) run . dag = self run . verify_integrity ( session = session ) return run
Creates a dag run from this dag including the tasks associated with this dag.
Creates a dag run from this dag including the tasks associated with this dag .
"""
        Creates a dag run from this dag including the tasks associated with this dag.
        Returns the dag run.

        :param run_id: defines the run id for this dag run
        :type run_id: str
        :param run_type: type of DagRun
        :type run_type: airflow.utils.types.DagRunType
        :param execution_date: the execution date of this dag run
        :type execution_date: datetime.datetime
        :param state: the state of the dag run
        :type state: airflow.utils.state.DagRunState
        :param start_date: the date this dag run should be evaluated
        :type start_date: datetime
        :param external_trigger: whether this dag run is externally triggered
        :type external_trigger: bool
        :param conf: Dict containing configuration/parameters to pass to the DAG
        :type conf: dict
        :param creating_job_id: id of the job creating this DagRun
        :type creating_job_id: int
        :param session: database session
        :type session: sqlalchemy.orm.session.Session
        :param dag_hash: Hash of Serialized DAG
        :type dag_hash: str
        :param data_interval: Data interval of the DagRun
        :type data_interval: tuple[datetime, datetime] | None
        """ # Infer run_type from run_id if needed. # Generate run_id from run_type and execution_date. # create a copy of params before validating # create the associated task instances # state is None at the moment of creation
self
state
DagRunState
the state of the dag run
the state of the dag run
execution_date
Optional[datetime]
the execution date of this dag run
the execution date of this dag run
run_id
Optional[str]
defines the run id for this dag run
defines the run id for this dag run
start_date
Optional[datetime]
the date this dag run should be evaluated
the date this dag run should be evaluated
external_trigger
Optional[bool]
whether this dag run is externally triggered
whether this dag run is externally triggered
conf
Optional[dict]
Dict containing configuration/parameters to pass to the DAG
Dict containing configuration / parameters to pass to the DAG
run_type
Optional[DagRunType]
type of DagRun
type of DagRun
session
None
dag_hash
Optional[str]
Hash of Serialized DAG
Hash of Serialized DAG
creating_job_id
Optional[int]
id of the job creating this DagRun
id of the job creating this DagRun
data_interval
Optional[Tuple[datetime, datetime]]
Data interval of the DagRun
Data interval of the DagRun