@provide_session<EOL><INDENT>def skip(self, dag_run, execution_date, tasks, session=None):<DEDENT>
if not tasks:<EOL><INDENT>return<EOL><DEDENT>task_ids = [d.task_id for d in tasks]<EOL>now = timezone.utcnow()<EOL>if dag_run:<EOL><INDENT>session.query(TaskInstance).filter(<EOL>TaskInstance.dag_id == dag_run.dag_id,<EOL>TaskInstance.execution_date == dag_run.execution_date,<EOL>TaskInstance.task_id.in_(task_ids)<EOL>).update({TaskInstance.state: State.SKIPPED,<EOL>TaskInstance.start_date: now,<EOL>TaskInstance.end_date: now},<EOL>synchronize_session=False)<EOL>session.commit()<EOL><DEDENT>else:<EOL><INDENT>assert execution_date is not None, "<STR_LIT>"<EOL>self.log.warning("<STR_LIT>")<EOL>for task in tasks:<EOL><INDENT>ti = TaskInstance(task, execution_date=execution_date)<EOL>ti.state = State.SKIPPED<EOL>ti.start_date = now<EOL>ti.end_date = now<EOL>session.merge(ti)<EOL><DEDENT>session.commit()<EOL><DEDENT>
Sets tasks instances to skipped from the same dag run.

:param dag_run: the DagRun for which to set the tasks to skipped
:param execution_date: execution_date
:param tasks: tasks to skip (not task_ids)
:param session: db session to use