You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by GitBox <gi...@apache.org> on 2021/10/04 23:04:35 UTC
[GitHub] [airflow] ephraimbuddy opened a new pull request #18724: Support the use of `run_id` in `set_task_instance_state`(REST API)
ephraimbuddy opened a new pull request #18724:
URL: https://github.com/apache/airflow/pull/18724
We can now use run_id as well as execution_date to update states
of task instances
Depends on https://github.com/apache/airflow/pull/18642
---
**^ Add meaningful description above**
Read the **[Pull Request Guidelines](https://github.com/apache/airflow/blob/main/CONTRIBUTING.rst#pull-request-guidelines)** for more information.
In case of fundamental code change, Airflow Improvement Proposal ([AIP](https://cwiki.apache.org/confluence/display/AIRFLOW/Airflow+Improvements+Proposals)) is needed.
In case of a new dependency, check compliance with the [ASF 3rd Party License Policy](https://www.apache.org/legal/resolved.html#category-x).
In case of backwards incompatible changes please leave a note in [UPDATING.md](https://github.com/apache/airflow/blob/main/UPDATING.md).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] github-actions[bot] commented on pull request #18724: Support the use of `run_id` in `set_task_instance_state`(REST API)
Posted by GitBox <gi...@apache.org>.
github-actions[bot] commented on pull request #18724:
URL: https://github.com/apache/airflow/pull/18724#issuecomment-974549986
This pull request has been automatically marked as stale because it has not had recent activity. It will be closed in 5 days if no further activity occurs. Thank you for your contributions.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] uranusjr commented on a change in pull request #18724: Use `DagRun.run_id` instead of `execution_date` when updating state of TIs(UI & REST API)
Posted by GitBox <gi...@apache.org>.
uranusjr commented on a change in pull request #18724:
URL: https://github.com/apache/airflow/pull/18724#discussion_r783727049
##########
File path: airflow/models/dag.py
##########
@@ -1660,14 +1664,24 @@ def set_task_instance_state(
:param past: Include all past TaskInstances of the given task_id
:type past: bool
"""
- from airflow.api.common.experimental.mark_tasks import set_state
+ from airflow.api.common.mark_tasks import set_state
+
+ if not exactly_one(execution_date, dag_run_id):
+ raise ValueError("Exactly one of execution_date or dag_run_id must be provided")
+ input_execution_date = execution_date
+ if dag_run_id:
+ dag_run = (
+ session.query(DagRun).filter(DagRun.run_id == dag_run_id, DagRun.dag_id == self.dag_id).one()
+ ) # Raises an error if not found
+ resolve_execution_date = dag_run.execution_date
Review comment:
How about
```python
if execution_date is None:
dag_run = ...
resolved_execution_date = dag_run.execution_date
else:
resolved_execution_date = execution_date
```
This would simplify the later code to
```python
end_date = resolve_execution_date if not future else None
start_date = resolve_execution_date if not future else None
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ephraimbuddy commented on a change in pull request #18724: Use `DagRun.run_id` instead of `execution_date` when updating state of TIs(UI & REST API)
Posted by GitBox <gi...@apache.org>.
ephraimbuddy commented on a change in pull request #18724:
URL: https://github.com/apache/airflow/pull/18724#discussion_r775043368
##########
File path: airflow/api_connexion/endpoints/task_instance_endpoint.py
##########
@@ -309,15 +308,34 @@ def post_set_task_instances_state(*, dag_id: str, session: Session = NEW_SESSION
error_message = f"Task ID {task_id} not found"
raise NotFound(error_message)
- execution_date = data['execution_date']
- try:
- session.query(TI).filter_by(execution_date=execution_date, task_id=task_id, dag_id=dag_id).one()
- except NoResultFound:
- raise NotFound(f"Task instance not found for task {task_id} on execution_date {execution_date}")
+ execution_date = data.get('execution_date')
+ run_id = data.get('dag_run_id')
+
+ if run_id is not None:
+ ti = (
+ session.query(func.count(TI.dag_id))
+ .filter(TI.task_id == task_id, TI.dag_id == dag_id, TI.run_id == run_id)
+ .scalar()
+ )
+ if not ti:
+ raise NotFound(
+ detail=f"Task instance not found for task {task_id!r} on DAG run with ID {run_id!r}"
+ )
+ elif execution_date is not None:
+ ti = (
+ session.query(TI.run_id)
+ .filter(TI.execution_date == execution_date, TI.task_id == task_id, TI.dag_id == dag_id)
+ .first()
+ )
+ if not ti:
+ raise NotFound(
+ detail=f"Task instance not found for task {task_id!r} on execution_date {execution_date}"
+ )
+ run_id = ti.run_id
Review comment:
Yeah. I have refactored it into taskinstance so we can now do TaskInstance.find just like we do for DagRun
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ephraimbuddy commented on a change in pull request #18724: Use `DagRun.run_id` instead of `execution_date` when updating state of TIs(UI & REST API)
Posted by GitBox <gi...@apache.org>.
ephraimbuddy commented on a change in pull request #18724:
URL: https://github.com/apache/airflow/pull/18724#discussion_r774986561
##########
File path: airflow/api_connexion/endpoints/task_instance_endpoint.py
##########
@@ -309,15 +308,34 @@ def post_set_task_instances_state(*, dag_id: str, session: Session = NEW_SESSION
error_message = f"Task ID {task_id} not found"
raise NotFound(error_message)
- execution_date = data['execution_date']
- try:
- session.query(TI).filter_by(execution_date=execution_date, task_id=task_id, dag_id=dag_id).one()
- except NoResultFound:
- raise NotFound(f"Task instance not found for task {task_id} on execution_date {execution_date}")
+ execution_date = data.get('execution_date')
+ run_id = data.get('dag_run_id')
+
+ if run_id is not None:
+ ti = (
+ session.query(func.count(TI.dag_id))
+ .filter(TI.task_id == task_id, TI.dag_id == dag_id, TI.run_id == run_id)
+ .scalar()
+ )
+ if not ti:
+ raise NotFound(
+ detail=f"Task instance not found for task {task_id!r} on DAG run with ID {run_id!r}"
+ )
+ elif execution_date is not None:
+ ti = (
+ session.query(TI.run_id)
+ .filter(TI.execution_date == execution_date, TI.task_id == task_id, TI.dag_id == dag_id)
+ .first()
+ )
+ if not ti:
+ raise NotFound(
+ detail=f"Task instance not found for task {task_id!r} on execution_date {execution_date}"
+ )
+ run_id = ti.run_id
Review comment:
One thing is, we don't also need to backfill it because `dag.set_task_instance_state` also accepts execution date. So I will work on passing the execution_date instead of the backfill
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] dstandish commented on a change in pull request #18724: Use `DagRun.run_id` instead of `execution_date` when updating state of TIs(UI & REST API)
Posted by GitBox <gi...@apache.org>.
dstandish commented on a change in pull request #18724:
URL: https://github.com/apache/airflow/pull/18724#discussion_r765345335
##########
File path: airflow/api_connexion/endpoints/task_instance_endpoint.py
##########
@@ -297,15 +296,34 @@ def post_set_task_instances_state(dag_id, session):
error_message = f"Task ID {task_id} not found"
raise NotFound(error_message)
- execution_date = data['execution_date']
- try:
- session.query(TI).filter_by(execution_date=execution_date, task_id=task_id, dag_id=dag_id).one()
- except NoResultFound:
- raise NotFound(f"Task instance not found for task {task_id} on execution_date {execution_date}")
+ execution_date = data.get('execution_date')
+ run_id = data.get('dag_run_id')
+
Review comment:
should we check that either run date or exec date is passed but not both? i think i have seen that kind of
i have seen elsewhere in
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] uranusjr commented on pull request #18724: Use `DagRun.run_id` instead of `execution_date` when updating state of TIs(UI & REST API)
Posted by GitBox <gi...@apache.org>.
uranusjr commented on pull request #18724:
URL: https://github.com/apache/airflow/pull/18724#issuecomment-994485121
LGTM, but the changes in `api/common` are conflicting terribly due to that other deprecation PR 😓
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ephraimbuddy commented on a change in pull request #18724: Use `DagRun.run_id` instead of `execution_date` when updating state of TIs(UI & REST API)
Posted by GitBox <gi...@apache.org>.
ephraimbuddy commented on a change in pull request #18724:
URL: https://github.com/apache/airflow/pull/18724#discussion_r783742658
##########
File path: airflow/api_connexion/endpoints/task_instance_endpoint.py
##########
@@ -309,14 +308,28 @@ def post_set_task_instances_state(*, dag_id: str, session: Session = NEW_SESSION
error_message = f"Task ID {task_id} not found"
raise NotFound(error_message)
- execution_date = data['execution_date']
- try:
- session.query(TI).filter_by(execution_date=execution_date, task_id=task_id, dag_id=dag_id).one()
- except NoResultFound:
- raise NotFound(f"Task instance not found for task {task_id} on execution_date {execution_date}")
+ execution_date = data.get('execution_date')
+ run_id = data.get('dag_run_id')
+ if (
+ execution_date
+ and (
+ session.query(TI)
+ .filter(TI.task_id == task_id, TI.dag_id == dag_id, TI.execution_date == execution_date)
+ .one_or_none()
+ )
+ is None
+ ):
+ raise NotFound(
+ detail=f"Task instance not found for task {task_id!r} on execution_date {execution_date}"
+ )
+
+ if run_id and not TI.find(task_id=task_id, dag_id=dag_id, run_id=run_id):
+ error_message = f"Task instance not found for task {task_id!r} on DAG run with ID {run_id!r}"
+ raise NotFound(detail=error_message)
tis = dag.set_task_instance_state(
task_id=task_id,
+ dag_run_id=run_id,
execution_date=execution_date,
Review comment:
We have schema level validation. So it won't allow both values to be supplied
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] uranusjr commented on a change in pull request #18724: Support the use of `run_id` in `set_task_instance_state`(REST API)
Posted by GitBox <gi...@apache.org>.
uranusjr commented on a change in pull request #18724:
URL: https://github.com/apache/airflow/pull/18724#discussion_r754992077
##########
File path: airflow/api/common/experimental/mark_tasks.py
##########
@@ -104,6 +120,16 @@ def set_state(
if dag is None:
raise ValueError("Received tasks with no DAG")
+ if run_id is not None:
+ dr = (
+ session.query(DagRun.execution_date)
+ .filter(DagRun.run_id == run_id, DagRun.dag_id == dag.dag_id)
+ .one_or_none()
+ )
+ if dr is None:
+ raise ValueError(f"Run id {run_id} not found")
+ execution_date = dr.execution_date
+
dates = get_execution_dates(dag, execution_date, future, past)
Review comment:
The real problem is actually `get_execution_date`. In stead of getting a list of dates and then look up task instances from them (`get_all_dag_task_query`), we should instead get a list of `run_id`s and looks up task instances of those instead. This entire function should be entirely reconstructed to operate around DAG runs instead of dates, and all of `get_execution_dates`, `verify_dag_run_integrity`, `get_subdag_runs`, and `get_all_dag_task_query` should either be deprecated or removed/rewritten entirely.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ephraimbuddy commented on a change in pull request #18724: Use `DagRun.run_id` instead of `execution_date` when updating state of TIs(UI & REST API)
Posted by GitBox <gi...@apache.org>.
ephraimbuddy commented on a change in pull request #18724:
URL: https://github.com/apache/airflow/pull/18724#discussion_r764229317
##########
File path: airflow/api/common/experimental/mark_tasks.py
##########
@@ -60,6 +62,7 @@ def _create_dagruns(dag, execution_dates, state, run_type):
return dag_runs
+@deprecated(reason="Use airflow.api.common.mark_tasks.set_state instead", version="2.2.3")
Review comment:
I have updated it but left the test file. Helped to verify the re-written functions.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] uranusjr commented on a change in pull request #18724: Use `DagRun.run_id` instead of `execution_date` when updating state of TIs(UI & REST API)
Posted by GitBox <gi...@apache.org>.
uranusjr commented on a change in pull request #18724:
URL: https://github.com/apache/airflow/pull/18724#discussion_r774963161
##########
File path: airflow/api_connexion/endpoints/task_instance_endpoint.py
##########
@@ -309,15 +308,34 @@ def post_set_task_instances_state(*, dag_id: str, session: Session = NEW_SESSION
error_message = f"Task ID {task_id} not found"
raise NotFound(error_message)
- execution_date = data['execution_date']
- try:
- session.query(TI).filter_by(execution_date=execution_date, task_id=task_id, dag_id=dag_id).one()
- except NoResultFound:
- raise NotFound(f"Task instance not found for task {task_id} on execution_date {execution_date}")
+ execution_date = data.get('execution_date')
+ run_id = data.get('dag_run_id')
+
+ if run_id is not None:
+ ti = (
+ session.query(func.count(TI.dag_id))
+ .filter(TI.task_id == task_id, TI.dag_id == dag_id, TI.run_id == run_id)
+ .scalar()
+ )
+ if not ti:
+ raise NotFound(
+ detail=f"Task instance not found for task {task_id!r} on DAG run with ID {run_id!r}"
+ )
+ elif execution_date is not None:
+ ti = (
+ session.query(TI.run_id)
+ .filter(TI.execution_date == execution_date, TI.task_id == task_id, TI.dag_id == dag_id)
+ .first()
+ )
+ if not ti:
+ raise NotFound(
+ detail=f"Task instance not found for task {task_id!r} on execution_date {execution_date}"
+ )
+ run_id = ti.run_id
Review comment:
Can (should?) this backfilling run_id from execution_date be done in the schema class?
##########
File path: airflow/models/dagrun.py
##########
@@ -296,9 +297,9 @@ def next_dagruns_to_examine(
def find(
cls,
dag_id: Optional[Union[str, List[str]]] = None,
- run_id: Optional[str] = None,
- execution_date: Optional[Union[datetime, List[datetime]]] = None,
- state: Optional[Union[str, DagRunState]] = None,
+ run_id: Optional[Iterable] = None,
+ execution_date: Optional[Union[datetime, Iterable]] = None,
Review comment:
```suggestion
run_id: Optional[Iterable[str]] = None,
execution_date: Optional[Union[datetime, Iterable[datetime]]] = None,
```
##########
File path: airflow/models/dag.py
##########
@@ -1661,14 +1665,22 @@ def set_task_instance_state(
:param past: Include all past TaskInstances of the given task_id
:type past: bool
"""
- from airflow.api.common.experimental.mark_tasks import set_state
+ from airflow.api.common.mark_tasks import set_state
+
+ if not (execution_date is None) ^ (dag_run_id is None):
Review comment:
Let’s use the `exactly_one` util added recently.
##########
File path: tests/conftest.py
##########
@@ -533,7 +533,7 @@ def create_dagrun(self, **kwargs):
if "run_type" not in kwargs:
kwargs["run_type"] = DagRunType.from_run_id(kwargs["run_id"])
- if "execution_date" not in kwargs:
+ if "execution_date" not in kwargs or kwargs["execution_date"] is None:
Review comment:
Maybe
```suggestion
if kwargs.get("execution_date") is None:
```
##########
File path: airflow/models/dag.py
##########
@@ -1690,6 +1702,9 @@ def set_task_instance_state(
include_upstream=False,
)
+ if dag_run_id:
+ execution_date = dag_run.execution_date
Review comment:
Should this be merged into the `if dag_run_id` block above?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ephraimbuddy commented on a change in pull request #18724: Use `DagRun.run_id` instead of `execution_date` when updating state of TIs(UI & REST API)
Posted by GitBox <gi...@apache.org>.
ephraimbuddy commented on a change in pull request #18724:
URL: https://github.com/apache/airflow/pull/18724#discussion_r774985756
##########
File path: airflow/api_connexion/endpoints/task_instance_endpoint.py
##########
@@ -309,15 +308,34 @@ def post_set_task_instances_state(*, dag_id: str, session: Session = NEW_SESSION
error_message = f"Task ID {task_id} not found"
raise NotFound(error_message)
- execution_date = data['execution_date']
- try:
- session.query(TI).filter_by(execution_date=execution_date, task_id=task_id, dag_id=dag_id).one()
- except NoResultFound:
- raise NotFound(f"Task instance not found for task {task_id} on execution_date {execution_date}")
+ execution_date = data.get('execution_date')
+ run_id = data.get('dag_run_id')
+
+ if run_id is not None:
+ ti = (
+ session.query(func.count(TI.dag_id))
+ .filter(TI.task_id == task_id, TI.dag_id == dag_id, TI.run_id == run_id)
+ .scalar()
+ )
+ if not ti:
+ raise NotFound(
+ detail=f"Task instance not found for task {task_id!r} on DAG run with ID {run_id!r}"
+ )
+ elif execution_date is not None:
+ ti = (
+ session.query(TI.run_id)
+ .filter(TI.execution_date == execution_date, TI.task_id == task_id, TI.dag_id == dag_id)
+ .first()
+ )
+ if not ti:
+ raise NotFound(
+ detail=f"Task instance not found for task {task_id!r} on execution_date {execution_date}"
+ )
+ run_id = ti.run_id
Review comment:
Have checked on this, we don't have access to the ` dag` at the schema level. Looking up the task instance may not always give a correct result
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ephraimbuddy commented on a change in pull request #18724: Use `DagRun.run_id` instead of `execution_date` when updating state of TIs(UI & REST API)
Posted by GitBox <gi...@apache.org>.
ephraimbuddy commented on a change in pull request #18724:
URL: https://github.com/apache/airflow/pull/18724#discussion_r763385437
##########
File path: setup.py
##########
@@ -490,7 +490,7 @@ def write_version(filename: str = os.path.join(*[my_dir, "airflow", "git_version
'python-telegram-bot~=13.0',
]
trino = [
- 'trino>=0.301.0',
+ 'trino>=0.301.0,<0.307.0',
Review comment:
The `trino` latest release causes issues and I think it came up because I added `deprecated`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ashb commented on a change in pull request #18724: Support the use of `run_id` in `set_task_instance_state`(REST API)
Posted by GitBox <gi...@apache.org>.
ashb commented on a change in pull request #18724:
URL: https://github.com/apache/airflow/pull/18724#discussion_r722123772
##########
File path: airflow/api_connexion/schemas/task_instance_schema.py
##########
@@ -122,7 +122,8 @@ class SetTaskInstanceStateFormSchema(Schema):
dry_run = fields.Boolean(dump_default=True)
task_id = fields.Str(required=True)
- execution_date = fields.DateTime(required=True, validate=validate_istimezone)
+ execution_date = fields.DateTime(validate=validate_istimezone)
+ dag_run_id = fields.Str()
Review comment:
Is it possible to add a custom validation so that this is an either/or thing? (I don't think passing both should be allowed)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ashb commented on a change in pull request #18724: Use `DagRun.run_id` instead of `execution_date` when updating state of TIs(UI & REST API)
Posted by GitBox <gi...@apache.org>.
ashb commented on a change in pull request #18724:
URL: https://github.com/apache/airflow/pull/18724#discussion_r783814623
##########
File path: airflow/api_connexion/endpoints/task_instance_endpoint.py
##########
@@ -309,14 +308,28 @@ def post_set_task_instances_state(*, dag_id: str, session: Session = NEW_SESSION
error_message = f"Task ID {task_id} not found"
raise NotFound(error_message)
- execution_date = data['execution_date']
- try:
- session.query(TI).filter_by(execution_date=execution_date, task_id=task_id, dag_id=dag_id).one()
- except NoResultFound:
- raise NotFound(f"Task instance not found for task {task_id} on execution_date {execution_date}")
+ execution_date = data.get('execution_date')
+ run_id = data.get('dag_run_id')
+ if (
+ execution_date
+ and (
+ session.query(TI)
+ .filter(TI.task_id == task_id, TI.dag_id == dag_id, TI.execution_date == execution_date)
+ .one_or_none()
+ )
+ is None
+ ):
+ raise NotFound(
+ detail=f"Task instance not found for task {task_id!r} on execution_date {execution_date}"
+ )
+
+ if run_id and not TI.find(task_id=task_id, dag_id=dag_id, run_id=run_id):
Review comment:
```suggestion
if run_id and not suession.query(TI).get{'task_id': task_id, 'dag_id': dag_id, 'run_id': run_id})):
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ephraimbuddy commented on a change in pull request #18724: Support the use of `run_id` in `set_task_instance_state`(REST API)
Posted by GitBox <gi...@apache.org>.
ephraimbuddy commented on a change in pull request #18724:
URL: https://github.com/apache/airflow/pull/18724#discussion_r754398212
##########
File path: airflow/api/common/experimental/mark_tasks.py
##########
@@ -104,6 +120,16 @@ def set_state(
if dag is None:
raise ValueError("Received tasks with no DAG")
+ if run_id is not None:
+ dr = (
+ session.query(DagRun.execution_date)
+ .filter(DagRun.run_id == run_id, DagRun.dag_id == dag.dag_id)
+ .one_or_none()
+ )
+ if dr is None:
+ raise ValueError(f"Run id {run_id} not found")
+ execution_date = dr.execution_date
+
dates = get_execution_dates(dag, execution_date, future, past)
Review comment:
It looks like if we don't convert it here, we will eventually convert to `execution_date ` in `get_execution_date` when we want to get the execution dates for the DAG since we will need to get the `start_date` and `end_date` to use in checking the execution dates if future or past is supplied.
I can't really figure how to get the execution dates without converting the run_id to execution_date.
Working on it but will appreciate any clue
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ephraimbuddy commented on pull request #18724: Use `DagRun.run_id` instead of `execution_date` when updating state of TIs(UI & REST API)
Posted by GitBox <gi...@apache.org>.
ephraimbuddy commented on pull request #18724:
URL: https://github.com/apache/airflow/pull/18724#issuecomment-989867929
Fixed small typing issues
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] uranusjr commented on pull request #18724: Use `DagRun.run_id` instead of `execution_date` when updating state of TIs(UI & REST API)
Posted by GitBox <gi...@apache.org>.
uranusjr commented on pull request #18724:
URL: https://github.com/apache/airflow/pull/18724#issuecomment-1000769229
How much of the new `mark_task` module is copied from the previous location? I started to read it half way before realising it’s mostly old (and bad) code. They should be refactored, but in a subsequent PR instead of this one :(
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] uranusjr commented on a change in pull request #18724: Use `DagRun.run_id` instead of `execution_date` when updating state of TIs(UI & REST API)
Posted by GitBox <gi...@apache.org>.
uranusjr commented on a change in pull request #18724:
URL: https://github.com/apache/airflow/pull/18724#discussion_r775022989
##########
File path: airflow/api_connexion/endpoints/task_instance_endpoint.py
##########
@@ -309,15 +308,34 @@ def post_set_task_instances_state(*, dag_id: str, session: Session = NEW_SESSION
error_message = f"Task ID {task_id} not found"
raise NotFound(error_message)
- execution_date = data['execution_date']
- try:
- session.query(TI).filter_by(execution_date=execution_date, task_id=task_id, dag_id=dag_id).one()
- except NoResultFound:
- raise NotFound(f"Task instance not found for task {task_id} on execution_date {execution_date}")
+ execution_date = data.get('execution_date')
+ run_id = data.get('dag_run_id')
+
+ if run_id is not None:
+ ti = (
+ session.query(func.count(TI.dag_id))
+ .filter(TI.task_id == task_id, TI.dag_id == dag_id, TI.run_id == run_id)
+ .scalar()
+ )
+ if not ti:
+ raise NotFound(
+ detail=f"Task instance not found for task {task_id!r} on DAG run with ID {run_id!r}"
+ )
+ elif execution_date is not None:
+ ti = (
+ session.query(TI.run_id)
+ .filter(TI.execution_date == execution_date, TI.task_id == task_id, TI.dag_id == dag_id)
+ .first()
+ )
+ if not ti:
+ raise NotFound(
+ detail=f"Task instance not found for task {task_id!r} on execution_date {execution_date}"
+ )
+ run_id = ti.run_id
Review comment:
Makes sense. Do you think it’d be worthwhile to refactor this out as a function for reuse (is it useful?)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ephraimbuddy commented on a change in pull request #18724: Support the use of `run_id` in `set_task_instance_state`(REST API)
Posted by GitBox <gi...@apache.org>.
ephraimbuddy commented on a change in pull request #18724:
URL: https://github.com/apache/airflow/pull/18724#discussion_r722150445
##########
File path: airflow/api_connexion/schemas/task_instance_schema.py
##########
@@ -122,7 +122,8 @@ class SetTaskInstanceStateFormSchema(Schema):
dry_run = fields.Boolean(dump_default=True)
task_id = fields.Str(required=True)
- execution_date = fields.DateTime(required=True, validate=validate_istimezone)
+ execution_date = fields.DateTime(validate=validate_istimezone)
+ dag_run_id = fields.Str()
Review comment:
The openAPi spec would have been a better place to handle it but it doesn't support it.
We can do it here but it won't appear in the documentation
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ephraimbuddy commented on a change in pull request #18724: Support the use of `run_id` in `set_task_instance_state`(REST API)
Posted by GitBox <gi...@apache.org>.
ephraimbuddy commented on a change in pull request #18724:
URL: https://github.com/apache/airflow/pull/18724#discussion_r754246424
##########
File path: airflow/api_connexion/openapi/v1.yaml
##########
@@ -2863,6 +2863,11 @@ components:
description: The execution date.
type: string
format: datetime
+ deprecated: true
Review comment:
I have added support running with either execution_date or run_id
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ephraimbuddy commented on a change in pull request #18724: Support the use of `run_id` in `set_task_instance_state`(REST API)
Posted by GitBox <gi...@apache.org>.
ephraimbuddy commented on a change in pull request #18724:
URL: https://github.com/apache/airflow/pull/18724#discussion_r722148345
##########
File path: airflow/api_connexion/openapi/v1.yaml
##########
@@ -2863,6 +2863,11 @@ components:
description: The execution date.
type: string
format: datetime
+ deprecated: true
Review comment:
I think the execution_date in this use case is needed. I would go with closing this PR because if we still use run_id, we would still need the execution_date
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ephraimbuddy commented on pull request #18724: Use `DagRun.run_id` instead of `execution_date` when updating state of TIs(UI & REST API)
Posted by GitBox <gi...@apache.org>.
ephraimbuddy commented on pull request #18724:
URL: https://github.com/apache/airflow/pull/18724#issuecomment-994638017
> LGTM, but the changes in `api/common` are conflicting terribly due to that other deprecation PR 😓
I will wait for the other PR before proceeding...
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ashb commented on a change in pull request #18724: Use `DagRun.run_id` instead of `execution_date` when updating state of TIs(UI & REST API)
Posted by GitBox <gi...@apache.org>.
ashb commented on a change in pull request #18724:
URL: https://github.com/apache/airflow/pull/18724#discussion_r783815163
##########
File path: airflow/models/taskinstance.py
##########
@@ -742,6 +742,34 @@ def error(self, session=NEW_SESSION):
session.merge(self)
session.commit()
+ @classmethod
+ @provide_session
+ def find(
Review comment:
Ah its used one place -- but still, I don't think we need it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ephraimbuddy commented on pull request #18724: Support the use of `run_id` in `set_task_instance_state`(REST API)
Posted by GitBox <gi...@apache.org>.
ephraimbuddy commented on pull request #18724:
URL: https://github.com/apache/airflow/pull/18724#issuecomment-934151714
I took the easiest way for this PR. There are a set of functions that uses the execution date underneath this PR but I limited the PR to the REST API. Let me know if those functions need to be touched too
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] uranusjr commented on a change in pull request #18724: Use `DagRun.run_id` instead of `execution_date` when updating state of TIs(UI & REST API)
Posted by GitBox <gi...@apache.org>.
uranusjr commented on a change in pull request #18724:
URL: https://github.com/apache/airflow/pull/18724#discussion_r775022760
##########
File path: airflow/models/dag.py
##########
@@ -1690,6 +1702,9 @@ def set_task_instance_state(
include_upstream=False,
)
+ if dag_run_id:
+ execution_date = dag_run.execution_date
Review comment:
I kind of feel bad we need to rely on passing `None` (instead of not passing) for these arguments, but I guess it’s easiest. Maybe we should give these `execution_date`s different names instead of reusing the variable name? `input_execution_date` (which can be None) and `resolved_execution_date` (which is always datetime), for example.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ephraimbuddy commented on a change in pull request #18724: Use `DagRun.run_id` instead of `execution_date` when updating state of TIs(UI & REST API)
Posted by GitBox <gi...@apache.org>.
ephraimbuddy commented on a change in pull request #18724:
URL: https://github.com/apache/airflow/pull/18724#discussion_r763952998
##########
File path: setup.py
##########
@@ -490,7 +490,7 @@ def write_version(filename: str = os.path.join(*[my_dir, "airflow", "git_version
'python-telegram-bot~=13.0',
]
trino = [
- 'trino>=0.301.0',
+ 'trino>=0.301.0,<0.307.0',
Review comment:
Yes. the trino provider tests were failing but I guess if I didn't add `deprecated` lib it won't surface. I will remove it and instead make a separate PR if it still occurs again
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ashb commented on a change in pull request #18724: Support the use of `run_id` in `set_task_instance_state`(REST API)
Posted by GitBox <gi...@apache.org>.
ashb commented on a change in pull request #18724:
URL: https://github.com/apache/airflow/pull/18724#discussion_r754299710
##########
File path: airflow/api/common/experimental/mark_tasks.py
##########
@@ -104,6 +120,16 @@ def set_state(
if dag is None:
raise ValueError("Received tasks with no DAG")
+ if run_id is not None:
+ dr = (
+ session.query(DagRun.execution_date)
+ .filter(DagRun.run_id == run_id, DagRun.dag_id == dag.dag_id)
+ .one_or_none()
+ )
+ if dr is None:
+ raise ValueError(f"Run id {run_id} not found")
+ execution_date = dr.execution_date
+
dates = get_execution_dates(dag, execution_date, future, past)
Review comment:
Hmmm, if a run_id is given, we should really just use that to look up things, rather than convert it back to a date.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] uranusjr commented on a change in pull request #18724: Support the use of `run_id` in `set_task_instance_state`(REST API)
Posted by GitBox <gi...@apache.org>.
uranusjr commented on a change in pull request #18724:
URL: https://github.com/apache/airflow/pull/18724#discussion_r753907424
##########
File path: airflow/api_connexion/endpoints/task_instance_endpoint.py
##########
@@ -297,15 +297,33 @@ def post_set_task_instances_state(dag_id, session):
error_message = f"Task ID {task_id} not found"
raise NotFound(error_message)
- execution_date = data['execution_date']
- try:
- session.query(TI).filter_by(execution_date=execution_date, task_id=task_id, dag_id=dag_id).one()
- except NoResultFound:
- raise NotFound(f"Task instance not found for task {task_id} on execution_date {execution_date}")
+ execution_date = data.get('execution_date')
+ run_id = data.get('dag_run_id')
+ if not run_id and not execution_date:
+ raise BadRequest(detail="You must provide either the dag_run_id or the execution_date")
+ if run_id and execution_date:
+ raise BadRequest(detail="You cannot provide both the dag_run_id and the execution_date")
Review comment:
I think this can be better handled in the schema (`set_task_instance_state_form`) instead. There are existing examples for this.
##########
File path: airflow/api_connexion/endpoints/task_instance_endpoint.py
##########
@@ -297,15 +297,33 @@ def post_set_task_instances_state(dag_id, session):
error_message = f"Task ID {task_id} not found"
raise NotFound(error_message)
- execution_date = data['execution_date']
- try:
- session.query(TI).filter_by(execution_date=execution_date, task_id=task_id, dag_id=dag_id).one()
- except NoResultFound:
- raise NotFound(f"Task instance not found for task {task_id} on execution_date {execution_date}")
+ execution_date = data.get('execution_date')
+ run_id = data.get('dag_run_id')
+ if not run_id and not execution_date:
+ raise BadRequest(detail="You must provide either the dag_run_id or the execution_date")
+ if run_id and execution_date:
+ raise BadRequest(detail="You cannot provide both the dag_run_id and the execution_date")
+ if run_id and not execution_date:
+ try:
+ session.query(TI).join(TI.dag_run).filter(TI.task_id == task_id, TI.dag_id == dag_id).filter(
+ DR.run_id == run_id
+ ).one()
+ except NoResultFound:
+ raise NotFound(detail=f"The taskinstance with dag_run_id: {run_id} not found")
Review comment:
```suggestion
raise NotFound(detail=f"Task instance not found for task {task_id} on DAG run with ID {run_id}")
```
to better match the other message.
##########
File path: airflow/api_connexion/endpoints/task_instance_endpoint.py
##########
@@ -297,15 +297,33 @@ def post_set_task_instances_state(dag_id, session):
error_message = f"Task ID {task_id} not found"
raise NotFound(error_message)
- execution_date = data['execution_date']
- try:
- session.query(TI).filter_by(execution_date=execution_date, task_id=task_id, dag_id=dag_id).one()
- except NoResultFound:
- raise NotFound(f"Task instance not found for task {task_id} on execution_date {execution_date}")
+ execution_date = data.get('execution_date')
+ run_id = data.get('dag_run_id')
+ if not run_id and not execution_date:
+ raise BadRequest(detail="You must provide either the dag_run_id or the execution_date")
+ if run_id and execution_date:
+ raise BadRequest(detail="You cannot provide both the dag_run_id and the execution_date")
+ if run_id and not execution_date:
+ try:
+ session.query(TI).join(TI.dag_run).filter(TI.task_id == task_id, TI.dag_id == dag_id).filter(
+ DR.run_id == run_id
+ ).one()
Review comment:
Although as a related improvement, it is better to use `count()` for both of these checks IMO.
##########
File path: airflow/api_connexion/endpoints/task_instance_endpoint.py
##########
@@ -297,15 +297,33 @@ def post_set_task_instances_state(dag_id, session):
error_message = f"Task ID {task_id} not found"
raise NotFound(error_message)
- execution_date = data['execution_date']
- try:
- session.query(TI).filter_by(execution_date=execution_date, task_id=task_id, dag_id=dag_id).one()
- except NoResultFound:
- raise NotFound(f"Task instance not found for task {task_id} on execution_date {execution_date}")
+ execution_date = data.get('execution_date')
+ run_id = data.get('dag_run_id')
+ if not run_id and not execution_date:
+ raise BadRequest(detail="You must provide either the dag_run_id or the execution_date")
+ if run_id and execution_date:
+ raise BadRequest(detail="You cannot provide both the dag_run_id and the execution_date")
+ if run_id and not execution_date:
+ try:
+ session.query(TI).join(TI.dag_run).filter(TI.task_id == task_id, TI.dag_id == dag_id).filter(
+ DR.run_id == run_id
+ ).one()
Review comment:
```suggestion
session.query(TI).join(TI.dag_run).filter(
TI.task_id == task_id,
TI.dag_id == dag_id,
DR.run_id == run_id,
).one()
```
##########
File path: airflow/models/dag.py
##########
@@ -1619,10 +1621,12 @@ def set_task_instance_state(
:param task_id: Task ID of the TaskInstance
:type task_id: str
- :param execution_date: execution_date of the TaskInstance
- :type execution_date: datetime
+ :param execution_date: Optional execution_date of the TaskInstance
+ :type execution_date: Optional[datetime]
:param state: State to set the TaskInstance to
:type state: State
+ :param run_id: Optional run_id for the TaskInstance
+ :type run_id: Optional[str]
Review comment:
Same as above, we should either document or check the arguments.
##########
File path: airflow/api/common/experimental/mark_tasks.py
##########
@@ -62,8 +62,10 @@ def _create_dagruns(dag, execution_dates, state, run_type):
@provide_session
def set_state(
+ *,
tasks: Iterable[BaseOperator],
- execution_date: datetime.datetime,
+ execution_date: Optional[datetime.datetime] = None,
+ run_id: Optional[str] = None,
Review comment:
Should we check only _exactly one_ of `run_id` and `execution_date` are passed to this function? If not, we should mention in the docstring this function expects the _caller_ to check first.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] uranusjr commented on a change in pull request #18724: Use `DagRun.run_id` instead of `execution_date` when updating state of TIs(UI & REST API)
Posted by GitBox <gi...@apache.org>.
uranusjr commented on a change in pull request #18724:
URL: https://github.com/apache/airflow/pull/18724#discussion_r783729490
##########
File path: airflow/api_connexion/endpoints/task_instance_endpoint.py
##########
@@ -309,14 +308,28 @@ def post_set_task_instances_state(*, dag_id: str, session: Session = NEW_SESSION
error_message = f"Task ID {task_id} not found"
raise NotFound(error_message)
- execution_date = data['execution_date']
- try:
- session.query(TI).filter_by(execution_date=execution_date, task_id=task_id, dag_id=dag_id).one()
- except NoResultFound:
- raise NotFound(f"Task instance not found for task {task_id} on execution_date {execution_date}")
+ execution_date = data.get('execution_date')
+ run_id = data.get('dag_run_id')
+ if (
+ execution_date
+ and (
+ session.query(TI)
+ .filter(TI.task_id == task_id, TI.dag_id == dag_id, TI.execution_date == execution_date)
+ .one_or_none()
+ )
+ is None
+ ):
+ raise NotFound(
+ detail=f"Task instance not found for task {task_id!r} on execution_date {execution_date}"
+ )
+
+ if run_id and not TI.find(task_id=task_id, dag_id=dag_id, run_id=run_id):
+ error_message = f"Task instance not found for task {task_id!r} on DAG run with ID {run_id!r}"
+ raise NotFound(detail=error_message)
tis = dag.set_task_instance_state(
task_id=task_id,
+ dag_run_id=run_id,
execution_date=execution_date,
Review comment:
This combined with the `exactly_one` check in `set_task_instance_state` means the request would error (with 500) if the user passes _both_ `execution_date` and `run_id`.=
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ephraimbuddy commented on a change in pull request #18724: Use `DagRun.run_id` instead of `execution_date` when updating state of TIs(UI & REST API)
Posted by GitBox <gi...@apache.org>.
ephraimbuddy commented on a change in pull request #18724:
URL: https://github.com/apache/airflow/pull/18724#discussion_r763389068
##########
File path: airflow/api/common/experimental/mark_tasks.py
##########
@@ -104,6 +120,16 @@ def set_state(
if dag is None:
raise ValueError("Received tasks with no DAG")
+ if run_id is not None:
+ dr = (
+ session.query(DagRun.execution_date)
+ .filter(DagRun.run_id == run_id, DagRun.dag_id == dag.dag_id)
+ .one_or_none()
+ )
+ if dr is None:
+ raise ValueError(f"Run id {run_id} not found")
+ execution_date = dr.execution_date
+
dates = get_execution_dates(dag, execution_date, future, past)
Review comment:
I deprecated the functions inside the module and created another module where I used `run_id`. I had to use `deprecated` lib because I'm thinking maybe someone uses the API externally. Let me know what you think about the whole thing.
Thanks!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ephraimbuddy commented on a change in pull request #18724: Use `DagRun.run_id` instead of `execution_date` when updating state of TIs(UI & REST API)
Posted by GitBox <gi...@apache.org>.
ephraimbuddy commented on a change in pull request #18724:
URL: https://github.com/apache/airflow/pull/18724#discussion_r774978511
##########
File path: airflow/models/dag.py
##########
@@ -1690,6 +1702,9 @@ def set_task_instance_state(
include_upstream=False,
)
+ if dag_run_id:
+ execution_date = dag_run.execution_date
Review comment:
It will cause us to have both an` execution_date` and `dag_run_id` sent to `set_state` which accepts exactly one value
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] uranusjr commented on a change in pull request #18724: Use `DagRun.run_id` instead of `execution_date` when updating state of TIs(UI & REST API)
Posted by GitBox <gi...@apache.org>.
uranusjr commented on a change in pull request #18724:
URL: https://github.com/apache/airflow/pull/18724#discussion_r764611325
##########
File path: airflow/api_connexion/endpoints/task_instance_endpoint.py
##########
@@ -297,15 +296,34 @@ def post_set_task_instances_state(dag_id, session):
error_message = f"Task ID {task_id} not found"
raise NotFound(error_message)
- execution_date = data['execution_date']
- try:
- session.query(TI).filter_by(execution_date=execution_date, task_id=task_id, dag_id=dag_id).one()
- except NoResultFound:
- raise NotFound(f"Task instance not found for task {task_id} on execution_date {execution_date}")
+ execution_date = data.get('execution_date')
+ run_id = data.get('dag_run_id')
+
+ if run_id and not execution_date:
+ ti = (
+ session.query(func.count(TI.dag_id))
+ .filter(TI.task_id == task_id, TI.dag_id == dag_id, TI.run_id == run_id)
+ .scalar()
+ )
+ if not ti:
+ raise NotFound(
+ detail=f"Task instance not found for task {task_id!r} on DAG run with ID {run_id!r}"
+ )
+ if execution_date and not run_id:
+ ti = (
+ session.query(TI.run_id)
+ .filter(TI.execution_date == execution_date, TI.task_id == task_id, TI.dag_id == dag_id)
+ .first()
+ )
+ if not ti:
+ raise NotFound(
+ detail=f"Task instance not found for task {task_id!r} on execution_date {execution_date}"
+ )
+ run_id = ti.run_id
Review comment:
These two `if` conditions can be simplified a bit since we know only one of these two can be provided. Something like
```python
if run_id is not None:
...
elif execution_date is not None:
...
```
should be enough.
##########
File path: airflow/api_connexion/openapi/v1.yaml
##########
@@ -3072,10 +3072,17 @@ components:
type: string
execution_date:
- description: The execution date.
+ description: The execution date. Either set this or dag_run_id but not both.
type: string
format: datetime
Review comment:
Do we want to deprecate this?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ashb commented on a change in pull request #18724: Use `DagRun.run_id` instead of `execution_date` when updating state of TIs(UI & REST API)
Posted by GitBox <gi...@apache.org>.
ashb commented on a change in pull request #18724:
URL: https://github.com/apache/airflow/pull/18724#discussion_r754300938
##########
File path: airflow/api_connexion/endpoints/task_instance_endpoint.py
##########
@@ -297,15 +296,32 @@ def post_set_task_instances_state(dag_id, session):
error_message = f"Task ID {task_id} not found"
raise NotFound(error_message)
- execution_date = data['execution_date']
- try:
- session.query(TI).filter_by(execution_date=execution_date, task_id=task_id, dag_id=dag_id).one()
- except NoResultFound:
- raise NotFound(f"Task instance not found for task {task_id} on execution_date {execution_date}")
+ execution_date = data.get('execution_date')
+ run_id = data.get('dag_run_id')
+
+ if run_id and not execution_date:
+ ti = (
+ session.query(func.count(TI.dag_id))
+ .filter(TI.task_id == task_id, TI.dag_id == dag_id, TI.run_id == run_id)
+ .scalar()
+ )
+ if not ti:
+ raise NotFound(detail=f"Task instance not found for task {task_id} on DAG run with ID {run_id}")
+ if execution_date and not run_id:
+ ti = (
+ session.query(func.count(TI.dag_id))
+ .filter(TI.execution_date == execution_date, TI.task_id == task_id, TI.dag_id == dag_id)
+ .scalar()
+ )
+ if not ti:
+ raise NotFound(
+ detail=f"Task instance not found for task {task_id} on execution_date {execution_date}"
Review comment:
```suggestion
detail=f"Task instance not found for task {task_id!r} on execution_date {execution_date!r}"
```
##########
File path: airflow/api_connexion/schemas/task_instance_schema.py
##########
@@ -129,13 +129,22 @@ class SetTaskInstanceStateFormSchema(Schema):
dry_run = fields.Boolean(dump_default=True)
task_id = fields.Str(required=True)
- execution_date = fields.DateTime(required=True, validate=validate_istimezone)
+ execution_date = fields.DateTime(validate=validate_istimezone)
+ dag_run_id = fields.Str()
Review comment:
Oh, this should probably just be called `run_id` -- that is what the column is anyway, and what we use in most/all of the function arguments.
##########
File path: airflow/api/common/experimental/mark_tasks.py
##########
@@ -33,6 +34,7 @@
from airflow.utils.types import DagRunType
+@deprecated(reason="Use airflow.api.common.mark_tasks._create_dagruns instead", version="2.2.3")
def _create_dagruns(dag, execution_dates, state, run_type):
Review comment:
This is a private function (that's the convention in Python for a leading underscore) so we shouldn't have to deprecate this one, we can just delete it and update what ever uses it to use the new version.
##########
File path: airflow/api/common/experimental/mark_tasks.py
##########
@@ -60,6 +62,7 @@ def _create_dagruns(dag, execution_dates, state, run_type):
return dag_runs
+@deprecated(reason="Use airflow.api.common.mark_tasks.set_state instead", version="2.2.3")
Review comment:
If the behaviour of these functions are the same then we should do something like this instead for the whole file:
```
import warnings
from airflow.api.common.mark_tasks import * # noqa
warnings.warn(
"This module is deprecated. Please use `airflow.api.common.mark_tasks` instead.", DeprecationWarning, stacklevel=2
)
```
##########
File path: airflow/api_connexion/endpoints/task_instance_endpoint.py
##########
@@ -297,15 +296,32 @@ def post_set_task_instances_state(dag_id, session):
error_message = f"Task ID {task_id} not found"
raise NotFound(error_message)
- execution_date = data['execution_date']
- try:
- session.query(TI).filter_by(execution_date=execution_date, task_id=task_id, dag_id=dag_id).one()
- except NoResultFound:
- raise NotFound(f"Task instance not found for task {task_id} on execution_date {execution_date}")
+ execution_date = data.get('execution_date')
+ run_id = data.get('dag_run_id')
+
+ if run_id and not execution_date:
+ ti = (
+ session.query(func.count(TI.dag_id))
+ .filter(TI.task_id == task_id, TI.dag_id == dag_id, TI.run_id == run_id)
+ .scalar()
+ )
+ if not ti:
+ raise NotFound(detail=f"Task instance not found for task {task_id} on DAG run with ID {run_id}")
Review comment:
```suggestion
raise NotFound(detail=f"Task instance not found for task {task_id!r} on DAG run with ID {run_id!r}")
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ashb commented on a change in pull request #18724: Support the use of `run_id` in `set_task_instance_state`(REST API)
Posted by GitBox <gi...@apache.org>.
ashb commented on a change in pull request #18724:
URL: https://github.com/apache/airflow/pull/18724#discussion_r722162665
##########
File path: airflow/api_connexion/openapi/v1.yaml
##########
@@ -2863,6 +2863,11 @@ components:
description: The execution date.
type: string
format: datetime
+ deprecated: true
Review comment:
We should support clearing just the tasks for a single dag run by ID.
(We should _look_ at refactoring the chain of methods we call to do the work)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ephraimbuddy merged pull request #18724: Use `DagRun.run_id` instead of `execution_date` when updating state of TIs(UI & REST API)
Posted by GitBox <gi...@apache.org>.
ephraimbuddy merged pull request #18724:
URL: https://github.com/apache/airflow/pull/18724
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ashb commented on a change in pull request #18724: Support the use of `run_id` in `set_task_instance_state`(REST API)
Posted by GitBox <gi...@apache.org>.
ashb commented on a change in pull request #18724:
URL: https://github.com/apache/airflow/pull/18724#discussion_r722122730
##########
File path: airflow/api_connexion/openapi/v1.yaml
##########
@@ -2863,6 +2863,11 @@ components:
description: The execution date.
type: string
format: datetime
+ deprecated: true
+
+ dag_run_id:
+ description: The DagRun dag_run_id
Review comment:
This should be an either/or with execution date -- not passing both.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] uranusjr commented on a change in pull request #18724: Support the use of `run_id` in `set_task_instance_state`(REST API)
Posted by GitBox <gi...@apache.org>.
uranusjr commented on a change in pull request #18724:
URL: https://github.com/apache/airflow/pull/18724#discussion_r722245434
##########
File path: airflow/api_connexion/endpoints/task_instance_endpoint.py
##########
@@ -287,11 +287,38 @@ def post_set_task_instances_state(dag_id, session):
error_message = f"Task ID {task_id} not found"
raise NotFound(error_message)
- execution_date = data['execution_date']
+ execution_date = data.get('execution_date')
+ run_id = data.get('dag_run_id')
+ if not run_id and not execution_date:
+ raise BadRequest(detail="You must provide either the dag_run_id or the execution_date")
+ if run_id and execution_date:
+ try:
+ session.query(TI).join(TI.dag_run).filter(
+ TI.execution_date == execution_date, TI.task_id == task_id, TI.dag_id == dag_id
+ ).filter(DR.run_id == run_id).one()
+ except NoResultFound:
+ raise NotFound(
+ detail=f"DagRun with run_id: {run_id} and execution_date: {execution_date} not found"
+ )
+ if run_id and not execution_date:
+ try:
+ ti = (
+ session.query(TI)
+ .join(TI.dag_run)
+ .filter(TI.task_id == task_id, TI.dag_id == dag_id)
+ .filter(DR.run_id == run_id)
+ .one()
+ )
+ execution_date = ti.execution_date
Review comment:
I think this can be done by simply changing the field to query to `DR.execution_date`? Not entirely sure.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] uranusjr commented on a change in pull request #18724: Support the use of `run_id` in `set_task_instance_state`(REST API)
Posted by GitBox <gi...@apache.org>.
uranusjr commented on a change in pull request #18724:
URL: https://github.com/apache/airflow/pull/18724#discussion_r754992077
##########
File path: airflow/api/common/experimental/mark_tasks.py
##########
@@ -104,6 +120,16 @@ def set_state(
if dag is None:
raise ValueError("Received tasks with no DAG")
+ if run_id is not None:
+ dr = (
+ session.query(DagRun.execution_date)
+ .filter(DagRun.run_id == run_id, DagRun.dag_id == dag.dag_id)
+ .one_or_none()
+ )
+ if dr is None:
+ raise ValueError(f"Run id {run_id} not found")
+ execution_date = dr.execution_date
+
dates = get_execution_dates(dag, execution_date, future, past)
Review comment:
Instead of getting a list of dates and then look up task instances from them (`get_all_dag_task_query`), we should instead get a list of `run_id`s and looks up task instances of those instead. This entire function should be entirely reconstructed to operate around DAG runs instead of dates, and all of `get_execution_dates`, `verify_dag_run_integrity`, `get_subdag_runs`, and `get_all_dag_task_query` should either be deprecated or removed/rewritten entirely.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ephraimbuddy commented on a change in pull request #18724: Use `DagRun.run_id` instead of `execution_date` when updating state of TIs(UI & REST API)
Posted by GitBox <gi...@apache.org>.
ephraimbuddy commented on a change in pull request #18724:
URL: https://github.com/apache/airflow/pull/18724#discussion_r763939338
##########
File path: airflow/api/common/experimental/mark_tasks.py
##########
@@ -60,6 +62,7 @@ def _create_dagruns(dag, execution_dates, state, run_type):
return dag_runs
+@deprecated(reason="Use airflow.api.common.mark_tasks.set_state instead", version="2.2.3")
Review comment:
Another problem is, I didn't use execution_date at `airflow.api.common.mark_tasks`. Should I add it?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ephraimbuddy commented on a change in pull request #18724: Use `DagRun.run_id` instead of `execution_date` when updating state of TIs(UI & REST API)
Posted by GitBox <gi...@apache.org>.
ephraimbuddy commented on a change in pull request #18724:
URL: https://github.com/apache/airflow/pull/18724#discussion_r763955916
##########
File path: airflow/api_connexion/schemas/task_instance_schema.py
##########
@@ -129,13 +129,22 @@ class SetTaskInstanceStateFormSchema(Schema):
dry_run = fields.Boolean(dump_default=True)
task_id = fields.Str(required=True)
- execution_date = fields.DateTime(required=True, validate=validate_istimezone)
+ execution_date = fields.DateTime(validate=validate_istimezone)
+ dag_run_id = fields.Str()
Review comment:
Yeah. I thought of that but why I used dag_run_id is just to be consistent with already existing endpoints which uses `dag_run_id` instead of `run_id`. If it's not a problem, let me know, so I can use it.
See https://airflow.apache.org/docs/apache-airflow/stable/stable-rest-api-ref.html#operation/get_dag_run and https://airflow.apache.org/docs/apache-airflow/stable/stable-rest-api-ref.html#tag/TaskInstance etc
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ephraimbuddy commented on a change in pull request #18724: Use `DagRun.run_id` instead of `execution_date` when updating state of TIs(UI & REST API)
Posted by GitBox <gi...@apache.org>.
ephraimbuddy commented on a change in pull request #18724:
URL: https://github.com/apache/airflow/pull/18724#discussion_r764785564
##########
File path: airflow/api_connexion/openapi/v1.yaml
##########
@@ -3072,10 +3072,17 @@ components:
type: string
execution_date:
- description: The execution date.
+ description: The execution date. Either set this or dag_run_id but not both.
type: string
format: datetime
Review comment:
I think we should continue to support it
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ephraimbuddy commented on a change in pull request #18724: Use `DagRun.run_id` instead of `execution_date` when updating state of TIs(UI & REST API)
Posted by GitBox <gi...@apache.org>.
ephraimbuddy commented on a change in pull request #18724:
URL: https://github.com/apache/airflow/pull/18724#discussion_r765813039
##########
File path: tests/conftest.py
##########
@@ -533,7 +533,7 @@ def create_dagrun(self, **kwargs):
if "run_type" not in kwargs:
kwargs["run_type"] = DagRunType.from_run_id(kwargs["run_id"])
- if "execution_date" not in kwargs:
+ if "execution_date" not in kwargs or kwargs["execution_date"] is None:
Review comment:
Updated the dag_maker.create_dagrun for ease of use
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] dstandish commented on a change in pull request #18724: Use `DagRun.run_id` instead of `execution_date` when updating state of TIs(UI & REST API)
Posted by GitBox <gi...@apache.org>.
dstandish commented on a change in pull request #18724:
URL: https://github.com/apache/airflow/pull/18724#discussion_r765345335
##########
File path: airflow/api_connexion/endpoints/task_instance_endpoint.py
##########
@@ -297,15 +296,34 @@ def post_set_task_instances_state(dag_id, session):
error_message = f"Task ID {task_id} not found"
raise NotFound(error_message)
- execution_date = data['execution_date']
- try:
- session.query(TI).filter_by(execution_date=execution_date, task_id=task_id, dag_id=dag_id).one()
- except NoResultFound:
- raise NotFound(f"Task instance not found for task {task_id} on execution_date {execution_date}")
+ execution_date = data.get('execution_date')
+ run_id = data.get('dag_run_id')
+
Review comment:
should we check that either run date or exec date is passed but not both? i think i have seen that kind of check elsewhere
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] uranusjr commented on a change in pull request #18724: Use `DagRun.run_id` instead of `execution_date` when updating state of TIs(UI & REST API)
Posted by GitBox <gi...@apache.org>.
uranusjr commented on a change in pull request #18724:
URL: https://github.com/apache/airflow/pull/18724#discussion_r780875292
##########
File path: airflow/models/taskinstance.py
##########
@@ -722,6 +722,32 @@ def error(self, session=NEW_SESSION):
session.merge(self)
session.commit()
+ @classmethod
+ @provide_session
+ def find(
+ cls,
+ task_id: str,
+ dag_id: str,
+ execution_date: Optional[datetime] = None,
+ run_id: Optional[str] = None,
+ session: Session = NEW_SESSION,
+ ) -> bool:
+ """Returns a TaskInstance for the given criteria or none."""
Review comment:
```suggestion
"""Find a TaskInstance for the given criteria.
Paramters ``dag_id`` and ``task_id`` are required. Exactly one of ``run_id`` and
``execution_date`` is required to find the DAG run that owns the task instance.
"""
```
But can we _not_ support `execution_date` here? A long-term goal of AIP-39 is to remove `DagRun.execution_date`’s uniqueness requirement (so two runs can have the same logical date as long as they have different run IDs), and I’m hesitant to add new APIs that rely on this unique constraint.
##########
File path: airflow/models/taskinstance.py
##########
@@ -722,6 +722,32 @@ def error(self, session=NEW_SESSION):
session.merge(self)
session.commit()
+ @classmethod
+ @provide_session
+ def find(
+ cls,
+ task_id: str,
+ dag_id: str,
+ execution_date: Optional[datetime] = None,
+ run_id: Optional[str] = None,
+ session: Session = NEW_SESSION,
Review comment:
```suggestion
*,
session: Session = NEW_SESSION,
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ashb commented on a change in pull request #18724: Support the use of `run_id` in `set_task_instance_state`(REST API)
Posted by GitBox <gi...@apache.org>.
ashb commented on a change in pull request #18724:
URL: https://github.com/apache/airflow/pull/18724#discussion_r722122438
##########
File path: airflow/api_connexion/endpoints/task_instance_endpoint.py
##########
@@ -287,11 +287,38 @@ def post_set_task_instances_state(dag_id, session):
error_message = f"Task ID {task_id} not found"
raise NotFound(error_message)
- execution_date = data['execution_date']
+ execution_date = data.get('execution_date')
+ run_id = data.get('dag_run_id')
+ if not run_id and not execution_date:
+ raise BadRequest(detail="You must provide either the dag_run_id or the execution_date")
+ if run_id and execution_date:
+ try:
+ session.query(TI).join(TI.dag_run).filter(
+ TI.execution_date == execution_date, TI.task_id == task_id, TI.dag_id == dag_id
+ ).filter(DR.run_id == run_id).one()
+ except NoResultFound:
+ raise NotFound(
+ detail=f"DagRun with run_id: {run_id} and execution_date: {execution_date} not found"
+ )
+ if run_id and not execution_date:
+ try:
+ ti = (
+ session.query(TI)
+ .join(TI.dag_run)
+ .filter(TI.task_id == task_id, TI.dag_id == dag_id)
+ .filter(DR.run_id == run_id)
+ .one()
+ )
+ execution_date = ti.execution_date
Review comment:
Yeah, I think this should be reworked -- it seems a bit wasteful to load up the dag run, get the execution date, "throw out" the dag run we loaded to the load it again later!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ashb commented on a change in pull request #18724: Support the use of `run_id` in `set_task_instance_state`(REST API)
Posted by GitBox <gi...@apache.org>.
ashb commented on a change in pull request #18724:
URL: https://github.com/apache/airflow/pull/18724#discussion_r722123410
##########
File path: airflow/api_connexion/openapi/v1.yaml
##########
@@ -2863,6 +2863,11 @@ components:
description: The execution date.
type: string
format: datetime
+ deprecated: true
Review comment:
Given the `include|future` and `include_past` I'm not 100% sure we can deprecate this -- does it "make sense" to pass a run_id and then say `include_past`, or does that make more sense to stick with execution date?
(I honestly don't have an opinion right now)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ephraimbuddy commented on a change in pull request #18724: Support the use of `run_id` in `set_task_instance_state`(REST API)
Posted by GitBox <gi...@apache.org>.
ephraimbuddy commented on a change in pull request #18724:
URL: https://github.com/apache/airflow/pull/18724#discussion_r754243372
##########
File path: airflow/api/common/experimental/mark_tasks.py
##########
@@ -62,8 +62,10 @@ def _create_dagruns(dag, execution_dates, state, run_type):
@provide_session
def set_state(
+ *,
tasks: Iterable[BaseOperator],
- execution_date: datetime.datetime,
+ execution_date: Optional[datetime.datetime] = None,
+ run_id: Optional[str] = None,
Review comment:
Done
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] uranusjr commented on a change in pull request #18724: Use `DagRun.run_id` instead of `execution_date` when updating state of TIs(UI & REST API)
Posted by GitBox <gi...@apache.org>.
uranusjr commented on a change in pull request #18724:
URL: https://github.com/apache/airflow/pull/18724#discussion_r763688907
##########
File path: setup.py
##########
@@ -490,7 +490,7 @@ def write_version(filename: str = os.path.join(*[my_dir, "airflow", "git_version
'python-telegram-bot~=13.0',
]
trino = [
- 'trino>=0.301.0',
+ 'trino>=0.301.0,<0.307.0',
Review comment:
Could you provide some more details on this? Does it cause conflicts or something else?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ashb commented on a change in pull request #18724: Use `DagRun.run_id` instead of `execution_date` when updating state of TIs(UI & REST API)
Posted by GitBox <gi...@apache.org>.
ashb commented on a change in pull request #18724:
URL: https://github.com/apache/airflow/pull/18724#discussion_r783797517
##########
File path: airflow/api_connexion/openapi/v1.yaml
##########
@@ -3073,10 +3073,17 @@ components:
type: string
execution_date:
- description: The execution date.
+ description: The execution date. Either set this or dag_run_id but not both.
type: string
format: datetime
+ dag_run_id:
+ description: |
+ The task instance's DAG run ID. Either set this or execution_date but not both.
+
+ *New in version 2.2.3*
Review comment:
2.2.3 is already out now isn't it, So 2.2.4 or 2.3.0 (Probably 2.3, as this one doesn't feel like a bug fix so shouldn't be backported.)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ashb commented on a change in pull request #18724: Use `DagRun.run_id` instead of `execution_date` when updating state of TIs(UI & REST API)
Posted by GitBox <gi...@apache.org>.
ashb commented on a change in pull request #18724:
URL: https://github.com/apache/airflow/pull/18724#discussion_r783811603
##########
File path: airflow/models/taskinstance.py
##########
@@ -742,6 +742,34 @@ def error(self, session=NEW_SESSION):
session.merge(self)
session.commit()
+ @classmethod
+ @provide_session
+ def find(
Review comment:
This function is only used for this one test, right? If so lets not add it (I'm not sure the pattern we have for DagRun.find etc is _good_)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] uranusjr commented on a change in pull request #18724: Use `DagRun.run_id` instead of `execution_date` when updating state of TIs(UI & REST API)
Posted by GitBox <gi...@apache.org>.
uranusjr commented on a change in pull request #18724:
URL: https://github.com/apache/airflow/pull/18724#discussion_r765347284
##########
File path: airflow/models/dag.py
##########
@@ -179,6 +179,20 @@ def get_last_dagrun(dag_id, session, include_externally_triggered=False):
return query.first()
+def get_first_dagrun(dag_id, session, include_externally_triggered=False):
+ """
+ Returns the first dag run for a dag, None if there was none.
+ First dag run can be any type of run eg. scheduled or backfilled.
+ Overridden DagRuns are ignored.
+ """
+ DR = DagRun
+ query = session.query(DR).filter(DR.dag_id == dag_id)
+ if not include_externally_triggered:
+ query = query.filter(DR.external_trigger == expression.false())
+ query = query.order_by(DR.execution_date.asc())
+ return query.first()
Review comment:
Do we _need_ this as a separate function? (I don’t like `get_last_dagrun` to begin with and would prefer we not adding more things like it unless needed.)
##########
File path: airflow/models/dag.py
##########
@@ -1630,7 +1650,7 @@ def _get_task_instances(
def set_task_instance_state(
self,
task_id: str,
- execution_date: datetime,
+ dag_run_id: str,
Review comment:
I think we might need to keep compatibility here.
##########
File path: airflow/models/dagrun.py
##########
@@ -336,7 +336,10 @@ def find(
if dag_ids:
qry = qry.filter(cls.dag_id.in_(dag_ids))
if run_id:
- qry = qry.filter(cls.run_id == run_id)
+ if isinstance(run_id, list):
+ qry = qry.filter(cls.run_id.in_(run_id))
+ else:
+ qry = qry.filter(cls.run_id == run_id)
Review comment:
```python
if is_container(run_id):
...
elif run_id is not None:
...
```
and we can relax the type of `run_id` to `Iterable`. (And yes the same applies to `execution_date` but that’s existing code.)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org