You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "sunank200 (via GitHub)" <gi...@apache.org> on 2025/02/10 11:25:24 UTC
[PR] AIP-83 amendment: Add logic for generating run_id when logical date is None. [airflow]
sunank200 opened a new pull request, #46616:
URL: https://github.com/apache/airflow/pull/46616
<!--
Licensed to the Apache Software Foundation (ASF) under one
or more contributor license agreements. See the NOTICE file
distributed with this work for additional information
regarding copyright ownership. The ASF licenses this file
to you under the Apache License, Version 2.0 (the
"License"); you may not use this file except in compliance
with the License. You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing,
software distributed under the License is distributed on an
"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
KIND, either express or implied. See the License for the
specific language governing permissions and limitations
under the License.
-->
This PR started with changes on [#46398](https://github.com/apache/airflow/pull/46398) As logic on 46398 has changed, I have created a different PR and closed the other keeping the commit from @prabhusneha
For the AIP 83 amendment, when a null logical date, use run_after + random string to generate run_id.
Closes: [#46199](https://github.com/apache/airflow/issues/46199)
<!--
Thank you for contributing! Please make sure that your code changes
are covered with tests. And in case of new features or big changes
remember to adjust the documentation.
Feel free to ping committers for the review!
In case of an existing issue, reference it using one of the following:
closes: #ISSUE
related: #ISSUE
How to write a good git commit message:
http://chris.beams.io/posts/git-commit/
-->
<!-- Please keep an empty line above the dashes. -->
---
**^ Add meaningful description above**
Read the **[Pull Request Guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#pull-request-guidelines)** for more information.
In case of fundamental code changes, an Airflow Improvement Proposal ([AIP](https://cwiki.apache.org/confluence/display/AIRFLOW/Airflow+Improvement+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 a newsfragment file, named `{pr_number}.significant.rst` or `{issue_number}.significant.rst`, in [newsfragments](https://github.com/apache/airflow/tree/main/newsfragments).
--
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
Re: [PR] AIP-83 amendment: Add logic for generating run_id when logical date is None. [airflow]
Posted by "sunank200 (via GitHub)" <gi...@apache.org>.
sunank200 commented on code in PR #46616:
URL: https://github.com/apache/airflow/pull/46616#discussion_r1950433584
##########
kubernetes_tests/test_base.py:
##########
@@ -287,12 +292,12 @@ def start_job_in_kubernetes(self, dag_id, host):
result_json = self.start_dag(dag_id=dag_id, host=host)
dag_runs = result_json["dag_runs"]
assert len(dag_runs) > 0
- logical_date = None
+ run_after = None
dag_run_id = None
for dag_run in dag_runs:
if dag_run["dag_id"] == dag_id:
- logical_date = dag_run["logical_date"]
+ run_after = dag_run["run_after"]
dag_run_id = dag_run["dag_run_id"]
break
- assert logical_date is not None, f"No logical_date can be found for the dag with {dag_id}"
- return dag_run_id, logical_date
+ assert run_after is not None, f"No run_after can be found for the dag with {dag_id}"
+ return dag_run_id, run_after
Review Comment:
I changed it. I also added run_after for this check.
--
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
Re: [PR] AIP-83 amendment: Add logic for generating run_id when logical date is None. [airflow]
Posted by "sunank200 (via GitHub)" <gi...@apache.org>.
sunank200 commented on code in PR #46616:
URL: https://github.com/apache/airflow/pull/46616#discussion_r1948976443
##########
airflow/jobs/scheduler_job_runner.py:
##########
@@ -1286,7 +1286,7 @@ def _create_dag_runs(self, dag_models: Collection[DagModel], session: Session) -
dag.create_dagrun(
run_id=dag.timetable.generate_run_id(
run_type=DagRunType.SCHEDULED,
- logical_date=dag_model.next_dagrun,
+ run_after=dag_model.next_dagrun or dag_model.next_dagrun_create_after,
data_interval=data_interval,
),
Review Comment:
changed 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
Re: [PR] AIP-83 amendment: Add logic for generating run_id when logical date is None. [airflow]
Posted by "sunank200 (via GitHub)" <gi...@apache.org>.
sunank200 commented on code in PR #46616:
URL: https://github.com/apache/airflow/pull/46616#discussion_r1950197336
##########
airflow/utils/types.py:
##########
@@ -42,8 +40,13 @@ class DagRunType(str, enum.Enum):
def __str__(self) -> str:
return self.value
- def generate_run_id(self, logical_date: datetime) -> str:
- return f"{self}__{logical_date.isoformat()}"
+ def generate_run_id(self, *, run_after: str) -> str:
+ """
+ Generate a random string for DagRun.
Review Comment:
changed 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
Re: [PR] AIP-83 amendment: Add logic for generating run_id when logical date is None. [airflow]
Posted by "dstandish (via GitHub)" <gi...@apache.org>.
dstandish commented on code in PR #46616:
URL: https://github.com/apache/airflow/pull/46616#discussion_r1951041464
##########
airflow/api/common/trigger_dag.py:
##########
@@ -42,6 +42,7 @@ def _trigger_dag(
dag_bag: DagBag,
*,
triggered_by: DagRunTriggeredByType,
+ run_after: datetime,
Review Comment:
but basically, any public function or public endpoint, my thought is, probably we should not expose this to user as an option / param
--
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
Re: [PR] AIP-83 amendment: Add logic for generating run_id when logical date is None. [airflow]
Posted by "sunank200 (via GitHub)" <gi...@apache.org>.
sunank200 commented on code in PR #46616:
URL: https://github.com/apache/airflow/pull/46616#discussion_r1948971362
##########
airflow/api_fastapi/core_api/datamodels/dag_run.py:
##########
@@ -99,13 +100,15 @@ def check_data_intervals(cls, values):
@model_validator(mode="after")
def validate_dag_run_id(self):
if not self.dag_run_id:
- self.dag_run_id = DagRun.generate_run_id(DagRunType.MANUAL, self.logical_date)
+ self.dag_run_id = DagRun.generate_run_id(
+ run_type=DagRunType.MANUAL, logical_date=self.logical_date, run_after=self.run_after
+ )
return self
# Mypy issue https://github.com/python/mypy/issues/1362
@computed_field # type: ignore[misc]
@property
- def logical_date(self) -> datetime:
+ def run_after(self) -> datetime:
return timezone.utcnow()
Review Comment:
Changed 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
Re: [PR] AIP-83 amendment: Add logic for generating run_id when logical date is None. [airflow]
Posted by "sunank200 (via GitHub)" <gi...@apache.org>.
sunank200 commented on code in PR #46616:
URL: https://github.com/apache/airflow/pull/46616#discussion_r1951176880
##########
airflow/api/common/trigger_dag.py:
##########
@@ -42,6 +42,7 @@ def _trigger_dag(
dag_bag: DagBag,
*,
triggered_by: DagRunTriggeredByType,
+ run_after: datetime,
Review Comment:
Created an issue here to fix this: [#46650](https://github.com/apache/airflow/issues/46650)
--
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
Re: [PR] AIP-83 amendment: Add logic for generating run_id when logical date is None. [airflow]
Posted by "sunank200 (via GitHub)" <gi...@apache.org>.
sunank200 commented on code in PR #46616:
URL: https://github.com/apache/airflow/pull/46616#discussion_r1948978201
##########
airflow/api_fastapi/core_api/routes/public/dag_run.py:
##########
@@ -365,22 +367,26 @@ def trigger_dag_run(
end=pendulum.instance(body.data_interval_end),
)
else:
- data_interval = dag.timetable.infer_manual_data_interval(run_after=logical_date)
+ if body.logical_date:
+ data_interval = dag.timetable.infer_manual_data_interval(run_after=logical_date or run_after)
+ run_after = data_interval.end
+ else:
+ data_interval = None
if body.dag_run_id:
run_id = body.dag_run_id
else:
run_id = dag.timetable.generate_run_id(
run_type=DagRunType.MANUAL,
- logical_date=logical_date,
data_interval=data_interval,
+ run_after=logical_date or run_after,
)
Review Comment:
changed 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
Re: [PR] AIP-83 amendment: Add logic for generating run_id when logical date is None. [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #46616:
URL: https://github.com/apache/airflow/pull/46616#discussion_r1948950841
##########
airflow/api_fastapi/core_api/routes/public/dag_run.py:
##########
@@ -365,22 +367,26 @@ def trigger_dag_run(
end=pendulum.instance(body.data_interval_end),
)
else:
- data_interval = dag.timetable.infer_manual_data_interval(run_after=logical_date)
+ if body.logical_date:
+ data_interval = dag.timetable.infer_manual_data_interval(run_after=logical_date or run_after)
+ run_after = data_interval.end
+ else:
+ data_interval = None
if body.dag_run_id:
run_id = body.dag_run_id
else:
run_id = dag.timetable.generate_run_id(
run_type=DagRunType.MANUAL,
- logical_date=logical_date,
data_interval=data_interval,
+ run_after=logical_date or run_after,
)
Review Comment:
Same, should use the DagRun interface.
--
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
Re: [PR] AIP-83 amendment: Add logic for generating run_id when logical date is None. [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #46616:
URL: https://github.com/apache/airflow/pull/46616#discussion_r1948950357
##########
airflow/jobs/scheduler_job_runner.py:
##########
@@ -1286,7 +1286,7 @@ def _create_dag_runs(self, dag_models: Collection[DagModel], session: Session) -
dag.create_dagrun(
run_id=dag.timetable.generate_run_id(
run_type=DagRunType.SCHEDULED,
- logical_date=dag_model.next_dagrun,
+ run_after=dag_model.next_dagrun or dag_model.next_dagrun_create_after,
data_interval=data_interval,
),
Review Comment:
This should use DagRun.generate_run_id instead so the run ID has the random string if needed.
--
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
Re: [PR] AIP-83 amendment: Add logic for generating run_id when logical date is None. [airflow]
Posted by "sunank200 (via GitHub)" <gi...@apache.org>.
sunank200 commented on PR #46616:
URL: https://github.com/apache/airflow/pull/46616#issuecomment-2650508014
> Two minor issues in test set up, otherwise this should be ready.
@uranusjr these should be fixed 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
Re: [PR] AIP-83 amendment: Add logic for generating run_id when logical date is None. [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #46616:
URL: https://github.com/apache/airflow/pull/46616#discussion_r1950266124
##########
airflow/models/dagrun.py:
##########
@@ -621,10 +622,20 @@ def find_duplicate(cls, dag_id: str, run_id: str, *, session: Session = NEW_SESS
return session.scalars(select(cls).where(cls.dag_id == dag_id, cls.run_id == run_id)).one_or_none()
@staticmethod
- def generate_run_id(run_type: DagRunType, logical_date: datetime) -> str:
- """Generate Run ID based on Run Type and logical Date."""
+ def generate_run_id(
+ *, run_type: DagRunType, logical_date: datetime | None = None, run_after: datetime
+ ) -> str:
+ """
+ Generate Run ID based on Run Type, run_after and logical Date.
+
+ :param run_type: type of DagRun
+ :param logical_date: the logical date
+ :param run_after: the date before which dag run won't start.
+ """
# _Ensure_ run_type is a DagRunType, not just a string from user code
- return DagRunType(run_type).generate_run_id(logical_date)
+ if logical_date:
+ return DagRunType(run_type).generate_run_id(run_after=logical_date.isoformat())
Review Comment:
I get what @Lee-W is trying to say. The argument on DagRunType.generate_run_id is a str so yeah it’s probably better for it to be named `suffix`. On the timetable it’s a datetime (and does not include the random string) so `run_after` is more appropriate.
--
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
Re: [PR] AIP-83 amendment: Add logic for generating run_id when logical date is None. [airflow]
Posted by "sunank200 (via GitHub)" <gi...@apache.org>.
sunank200 commented on code in PR #46616:
URL: https://github.com/apache/airflow/pull/46616#discussion_r1950198938
##########
airflow/api/common/trigger_dag.py:
##########
@@ -131,6 +137,7 @@ def trigger_dag(
:param dag_id: DAG ID
:param triggered_by: the entity which triggers the dag_run
+ :param run_after: the datetime before which dag won't run.
Review Comment:
It is changed to `the datetime before which dag cannot run.`
--
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
Re: [PR] AIP-83 amendment: Add logic for generating run_id when logical date is None. [airflow]
Posted by "sunank200 (via GitHub)" <gi...@apache.org>.
sunank200 commented on code in PR #46616:
URL: https://github.com/apache/airflow/pull/46616#discussion_r1950316726
##########
airflow/models/dagrun.py:
##########
@@ -621,10 +622,20 @@ def find_duplicate(cls, dag_id: str, run_id: str, *, session: Session = NEW_SESS
return session.scalars(select(cls).where(cls.dag_id == dag_id, cls.run_id == run_id)).one_or_none()
@staticmethod
- def generate_run_id(run_type: DagRunType, logical_date: datetime) -> str:
- """Generate Run ID based on Run Type and logical Date."""
+ def generate_run_id(
+ *, run_type: DagRunType, logical_date: datetime | None = None, run_after: datetime
+ ) -> str:
+ """
+ Generate Run ID based on Run Type, run_after and logical Date.
+
+ :param run_type: type of DagRun
+ :param logical_date: the logical date
+ :param run_after: the date before which dag run won't start.
+ """
# _Ensure_ run_type is a DagRunType, not just a string from user code
- return DagRunType(run_type).generate_run_id(logical_date)
+ if logical_date:
+ return DagRunType(run_type).generate_run_id(suffix_id=run_after.isoformat())
+ return DagRunType(run_type).generate_run_id(suffix_id=f"{run_after.isoformat()}{get_random_string()}")
Review Comment:
Changed 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
Re: [PR] AIP-83 amendment: Add logic for generating run_id when logical date is None. [airflow]
Posted by "sunank200 (via GitHub)" <gi...@apache.org>.
sunank200 commented on code in PR #46616:
URL: https://github.com/apache/airflow/pull/46616#discussion_r1948981791
##########
airflow/models/backfill.py:
##########
@@ -298,8 +298,8 @@ def _create_backfill_dag_run(
dr = dag.create_dagrun(
run_id=dag.timetable.generate_run_id(
run_type=DagRunType.BACKFILL_JOB,
- logical_date=info.logical_date,
data_interval=info.data_interval,
+ run_after=info.logical_date or info.run_after,
),
Review Comment:
Changed 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
Re: [PR] AIP-83 amendment: Add logic for generating run_id when logical date is None. [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #46616:
URL: https://github.com/apache/airflow/pull/46616#discussion_r1950409180
##########
kubernetes_tests/test_base.py:
##########
@@ -264,8 +266,11 @@ def start_dag(self, dag_id, host):
assert result.status_code == 200, f"Could not enable DAG: {result_json}"
post_string = f"http://{host}/api/v1/dags/{dag_id}/dagRuns"
print(f"Calling [start_dag]#2 {post_string}")
+
# Trigger a new dagrun
- result = self.session.post(post_string, json={})
+ result = self.session.post(
+ post_string, json={"logical_date": timezone.coerce_datetime(timezone.utcnow())}
Review Comment:
I think this can simply use `datetime.datetime.now(datetime.utc)` like in Docker tests. Don’t need to import Airflow here.
--
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
Re: [PR] AIP-83 amendment: Add logic for generating run_id when logical date is None. [airflow]
Posted by "sunank200 (via GitHub)" <gi...@apache.org>.
sunank200 commented on code in PR #46616:
URL: https://github.com/apache/airflow/pull/46616#discussion_r1950196697
##########
airflow/models/dagrun.py:
##########
@@ -621,10 +622,20 @@ def find_duplicate(cls, dag_id: str, run_id: str, *, session: Session = NEW_SESS
return session.scalars(select(cls).where(cls.dag_id == dag_id, cls.run_id == run_id)).one_or_none()
@staticmethod
- def generate_run_id(run_type: DagRunType, logical_date: datetime) -> str:
- """Generate Run ID based on Run Type and logical Date."""
+ def generate_run_id(
+ *, run_type: DagRunType, logical_date: datetime | None = None, run_after: datetime
+ ) -> str:
+ """
+ Generate Run ID based on Run Type, run_after and logical Date.
+
+ :param run_type: type of DagRun
+ :param logical_date: the logical date
+ :param run_after: the date before which dag run won't start.
+ """
# _Ensure_ run_type is a DagRunType, not just a string from user code
- return DagRunType(run_type).generate_run_id(logical_date)
+ if logical_date:
+ return DagRunType(run_type).generate_run_id(run_after=logical_date.isoformat())
Review Comment:
As per discussion with @uranusjr When we have logical_date, we should generate run_id based on logical_date.
But when there is no run_after we should pass `run_after + random string`.
In generate_run_id of DagRunType, I have just added one date field and I have called it `run_after` for 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
Re: [PR] AIP-83 amendment: Add logic for generating run_id when logical date is None. [airflow]
Posted by "sunank200 (via GitHub)" <gi...@apache.org>.
sunank200 commented on code in PR #46616:
URL: https://github.com/apache/airflow/pull/46616#discussion_r1950434038
##########
kubernetes_tests/test_base.py:
##########
@@ -264,8 +266,11 @@ def start_dag(self, dag_id, host):
assert result.status_code == 200, f"Could not enable DAG: {result_json}"
post_string = f"http://{host}/api/v1/dags/{dag_id}/dagRuns"
print(f"Calling [start_dag]#2 {post_string}")
+
# Trigger a new dagrun
- result = self.session.post(post_string, json={})
+ result = self.session.post(
+ post_string, json={"logical_date": timezone.coerce_datetime(timezone.utcnow())}
Review Comment:
I have changed 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
Re: [PR] AIP-83 amendment: Add logic for generating run_id when logical date is None. [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #46616:
URL: https://github.com/apache/airflow/pull/46616#discussion_r1948951325
##########
airflow/models/backfill.py:
##########
@@ -298,8 +298,8 @@ def _create_backfill_dag_run(
dr = dag.create_dagrun(
run_id=dag.timetable.generate_run_id(
run_type=DagRunType.BACKFILL_JOB,
- logical_date=info.logical_date,
data_interval=info.data_interval,
+ run_after=info.logical_date or info.run_after,
),
Review Comment:
Ditto
--
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
Re: [PR] AIP-83 amendment: Add logic for generating run_id when logical date is None. [airflow]
Posted by "Lee-W (via GitHub)" <gi...@apache.org>.
Lee-W commented on code in PR #46616:
URL: https://github.com/apache/airflow/pull/46616#discussion_r1950172124
##########
airflow/api/common/trigger_dag.py:
##########
@@ -131,6 +137,7 @@ def trigger_dag(
:param dag_id: DAG ID
:param triggered_by: the entity which triggers the dag_run
+ :param run_after: the datetime before which dag won't run.
Review Comment:
Please unifty the docstring for the same `run_after`
##########
airflow/utils/types.py:
##########
@@ -42,8 +40,13 @@ class DagRunType(str, enum.Enum):
def __str__(self) -> str:
return self.value
- def generate_run_id(self, logical_date: datetime) -> str:
- return f"{self}__{logical_date.isoformat()}"
+ def generate_run_id(self, *, run_after: str) -> str:
+ """
+ Generate a random string for DagRun.
Review Comment:
We might need to update the docstring here
##########
airflow/models/dagrun.py:
##########
@@ -621,10 +622,20 @@ def find_duplicate(cls, dag_id: str, run_id: str, *, session: Session = NEW_SESS
return session.scalars(select(cls).where(cls.dag_id == dag_id, cls.run_id == run_id)).one_or_none()
@staticmethod
- def generate_run_id(run_type: DagRunType, logical_date: datetime) -> str:
- """Generate Run ID based on Run Type and logical Date."""
+ def generate_run_id(
+ *, run_type: DagRunType, logical_date: datetime | None = None, run_after: datetime
+ ) -> str:
+ """
+ Generate Run ID based on Run Type, run_after and logical Date.
+
+ :param run_type: type of DagRun
+ :param logical_date: the logical date
+ :param run_after: the date before which dag run won't start.
+ """
# _Ensure_ run_type is a DagRunType, not just a string from user code
- return DagRunType(run_type).generate_run_id(logical_date)
+ if logical_date:
+ return DagRunType(run_type).generate_run_id(run_after=logical_date.isoformat())
Review Comment:
Is it correct to pass logical_date as run_after? Wouldn't it be confusing? when `logical_dates` exists, we pass it as `run_after` while when `run_after` exists, we pass `run_after + random string`
--
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
Re: [PR] AIP-83 amendment: Add logic for generating run_id when logical date is None. [airflow]
Posted by "sunank200 (via GitHub)" <gi...@apache.org>.
sunank200 commented on code in PR #46616:
URL: https://github.com/apache/airflow/pull/46616#discussion_r1950582991
##########
kubernetes_tests/test_base.py:
##########
@@ -220,7 +220,7 @@ def ensure_dag_expected_state(self, host, logical_date, dag_id, expected_final_s
print(f"Received: {result}")
state = None
for dag_run in result_json["dag_runs"]:
- if dag_run["logical_date"] == logical_date:
+ if dag_run["run_after"] == logical_date:
Review Comment:
Right. I missed it while changing it back. 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
Re: [PR] AIP-83 amendment: Add logic for generating run_id when logical date is None. [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #46616:
URL: https://github.com/apache/airflow/pull/46616#discussion_r1950579397
##########
kubernetes_tests/test_base.py:
##########
@@ -220,7 +220,7 @@ def ensure_dag_expected_state(self, host, logical_date, dag_id, expected_final_s
print(f"Received: {result}")
state = None
for dag_run in result_json["dag_runs"]:
- if dag_run["logical_date"] == logical_date:
+ if dag_run["run_after"] == logical_date:
Review Comment:
I don’t we should remove the `logical_date` key.
--
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
Re: [PR] AIP-83 amendment: Add logic for generating run_id when logical date is None. [airflow]
Posted by "sunank200 (via GitHub)" <gi...@apache.org>.
sunank200 commented on code in PR #46616:
URL: https://github.com/apache/airflow/pull/46616#discussion_r1950202997
##########
airflow/models/dagrun.py:
##########
@@ -621,10 +622,20 @@ def find_duplicate(cls, dag_id: str, run_id: str, *, session: Session = NEW_SESS
return session.scalars(select(cls).where(cls.dag_id == dag_id, cls.run_id == run_id)).one_or_none()
@staticmethod
- def generate_run_id(run_type: DagRunType, logical_date: datetime) -> str:
- """Generate Run ID based on Run Type and logical Date."""
+ def generate_run_id(
+ *, run_type: DagRunType, logical_date: datetime | None = None, run_after: datetime
+ ) -> str:
+ """
+ Generate Run ID based on Run Type, run_after and logical Date.
+
+ :param run_type: type of DagRun
+ :param logical_date: the logical date
+ :param run_after: the date before which dag run won't start.
+ """
# _Ensure_ run_type is a DagRunType, not just a string from user code
- return DagRunType(run_type).generate_run_id(logical_date)
+ if logical_date:
+ return DagRunType(run_type).generate_run_id(run_after=logical_date.isoformat())
Review Comment:
I can rename this argument in another PR with a better name to unblock other PRs. WDYT?
--
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
Re: [PR] AIP-83 amendment: Add logic for generating run_id when logical date is None. [airflow]
Posted by "sunank200 (via GitHub)" <gi...@apache.org>.
sunank200 commented on code in PR #46616:
URL: https://github.com/apache/airflow/pull/46616#discussion_r1950288801
##########
airflow/models/dagrun.py:
##########
@@ -621,10 +622,20 @@ def find_duplicate(cls, dag_id: str, run_id: str, *, session: Session = NEW_SESS
return session.scalars(select(cls).where(cls.dag_id == dag_id, cls.run_id == run_id)).one_or_none()
@staticmethod
- def generate_run_id(run_type: DagRunType, logical_date: datetime) -> str:
- """Generate Run ID based on Run Type and logical Date."""
+ def generate_run_id(
+ *, run_type: DagRunType, logical_date: datetime | None = None, run_after: datetime
+ ) -> str:
+ """
+ Generate Run ID based on Run Type, run_after and logical Date.
+
+ :param run_type: type of DagRun
+ :param logical_date: the logical date
+ :param run_after: the date before which dag run won't start.
+ """
# _Ensure_ run_type is a DagRunType, not just a string from user code
- return DagRunType(run_type).generate_run_id(logical_date)
+ if logical_date:
+ return DagRunType(run_type).generate_run_id(run_after=logical_date.isoformat())
Review Comment:
Changed 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
Re: [PR] AIP-83 amendment: Add logic for generating run_id when logical date is None. [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #46616:
URL: https://github.com/apache/airflow/pull/46616#discussion_r1950290367
##########
airflow/models/dagrun.py:
##########
@@ -621,10 +622,20 @@ def find_duplicate(cls, dag_id: str, run_id: str, *, session: Session = NEW_SESS
return session.scalars(select(cls).where(cls.dag_id == dag_id, cls.run_id == run_id)).one_or_none()
@staticmethod
- def generate_run_id(run_type: DagRunType, logical_date: datetime) -> str:
- """Generate Run ID based on Run Type and logical Date."""
+ def generate_run_id(
+ *, run_type: DagRunType, logical_date: datetime | None = None, run_after: datetime
+ ) -> str:
+ """
+ Generate Run ID based on Run Type, run_after and logical Date.
+
+ :param run_type: type of DagRun
+ :param logical_date: the logical date
+ :param run_after: the date before which dag run won't start.
+ """
# _Ensure_ run_type is a DagRunType, not just a string from user code
- return DagRunType(run_type).generate_run_id(logical_date)
+ if logical_date:
+ return DagRunType(run_type).generate_run_id(suffix_id=run_after.isoformat())
+ return DagRunType(run_type).generate_run_id(suffix_id=f"{run_after.isoformat()}{get_random_string()}")
Review Comment:
I’d probably just call it `suffix`. Also we should add a `_` between the timestamp and the random string so they’re more obviously separate.
--
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
Re: [PR] AIP-83 amendment: Add logic for generating run_id when logical date is None. [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #46616:
URL: https://github.com/apache/airflow/pull/46616#discussion_r1950580066
##########
kubernetes_tests/test_base.py:
##########
@@ -287,12 +289,13 @@ def start_job_in_kubernetes(self, dag_id, host):
result_json = self.start_dag(dag_id=dag_id, host=host)
dag_runs = result_json["dag_runs"]
assert len(dag_runs) > 0
- logical_date = None
+ run_after = None
Review Comment:
Same, we should not remove `logical_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
Re: [PR] AIP-83 amendment: Add logic for generating run_id when logical date is None. [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #46616:
URL: https://github.com/apache/airflow/pull/46616#discussion_r1950263369
##########
airflow/api_connexion/schemas/dag_run_schema.py:
##########
@@ -78,17 +79,21 @@ class Meta:
@pre_load
def autogenerate(self, data, **kwargs):
- """Auto generate run_id and logical_date if they are not provided."""
- logical_date = data.get("logical_date", _MISSING)
+ """Auto generate run_id and run_after if they are not provided."""
+ run_after = data.get("run_after", _MISSING)
- # Auto-generate logical_date if missing
- if logical_date is _MISSING:
- data["logical_date"] = str(timezone.utcnow())
+ # Auto-generate run_after if missing
+ if run_after is _MISSING:
+ data["run_after"] = str(timezone.utcnow())
if "dag_run_id" not in data:
try:
data["dag_run_id"] = DagRun.generate_run_id(
- DagRunType.MANUAL, timezone.parse(data["logical_date"])
+ run_type=DagRunType.MANUAL,
+ logical_date=timezone.parse(data.get("logical_date"))
+ if data.get("logical_date")
+ else None,
+ run_after=timezone.parse(data["run_after"]),
Review Comment:
Don’t call `.get` twice like this.
```python
if logical_date_str := data.get("logical_date"):
logical_date = timezone.parse(logical_date_str)
else:
logical_date = 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
Re: [PR] AIP-83 amendment: Add logic for generating run_id when logical date is None. [airflow]
Posted by "sunank200 (via GitHub)" <gi...@apache.org>.
sunank200 commented on code in PR #46616:
URL: https://github.com/apache/airflow/pull/46616#discussion_r1950434038
##########
kubernetes_tests/test_base.py:
##########
@@ -264,8 +266,11 @@ def start_dag(self, dag_id, host):
assert result.status_code == 200, f"Could not enable DAG: {result_json}"
post_string = f"http://{host}/api/v1/dags/{dag_id}/dagRuns"
print(f"Calling [start_dag]#2 {post_string}")
+
# Trigger a new dagrun
- result = self.session.post(post_string, json={})
+ result = self.session.post(
+ post_string, json={"logical_date": timezone.coerce_datetime(timezone.utcnow())}
Review Comment:
I have changed it to datetime string like in Docker tests.
--
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
Re: [PR] AIP-83 amendment: Add logic for generating run_id when logical date is None. [airflow]
Posted by "dstandish (via GitHub)" <gi...@apache.org>.
dstandish commented on code in PR #46616:
URL: https://github.com/apache/airflow/pull/46616#discussion_r1951041464
##########
airflow/api/common/trigger_dag.py:
##########
@@ -42,6 +42,7 @@ def _trigger_dag(
dag_bag: DagBag,
*,
triggered_by: DagRunTriggeredByType,
+ run_after: datetime,
Review Comment:
but basically, any public function or public endpoint, my thought is, probably we should not expose this to user as an option / param. essentially it's read only attr i'm thinking.
--
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
Re: [PR] AIP-83 amendment: Add logic for generating run_id when logical date is None. [airflow]
Posted by "Lee-W (via GitHub)" <gi...@apache.org>.
Lee-W commented on code in PR #46616:
URL: https://github.com/apache/airflow/pull/46616#discussion_r1950199442
##########
airflow/models/dagrun.py:
##########
@@ -621,10 +622,20 @@ def find_duplicate(cls, dag_id: str, run_id: str, *, session: Session = NEW_SESS
return session.scalars(select(cls).where(cls.dag_id == dag_id, cls.run_id == run_id)).one_or_none()
@staticmethod
- def generate_run_id(run_type: DagRunType, logical_date: datetime) -> str:
- """Generate Run ID based on Run Type and logical Date."""
+ def generate_run_id(
+ *, run_type: DagRunType, logical_date: datetime | None = None, run_after: datetime
+ ) -> str:
+ """
+ Generate Run ID based on Run Type, run_after and logical Date.
+
+ :param run_type: type of DagRun
+ :param logical_date: the logical date
+ :param run_after: the date before which dag run won't start.
+ """
# _Ensure_ run_type is a DagRunType, not just a string from user code
- return DagRunType(run_type).generate_run_id(logical_date)
+ if logical_date:
+ return DagRunType(run_type).generate_run_id(run_after=logical_date.isoformat())
Review Comment:
I am not sure we should call it `run_after`. looks more like `suffix` to me. as long as @uranusjr 's good with it and users won't access it, I think we're good to keep it as it is
--
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
Re: [PR] AIP-83 amendment: Add logic for generating run_id when logical date is None. [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #46616:
URL: https://github.com/apache/airflow/pull/46616#discussion_r1950266941
##########
airflow/models/dagrun.py:
##########
@@ -621,10 +622,20 @@ def find_duplicate(cls, dag_id: str, run_id: str, *, session: Session = NEW_SESS
return session.scalars(select(cls).where(cls.dag_id == dag_id, cls.run_id == run_id)).one_or_none()
@staticmethod
- def generate_run_id(run_type: DagRunType, logical_date: datetime) -> str:
- """Generate Run ID based on Run Type and logical Date."""
+ def generate_run_id(
+ *, run_type: DagRunType, logical_date: datetime | None = None, run_after: datetime
+ ) -> str:
+ """
+ Generate Run ID based on Run Type, run_after and logical Date.
+
+ :param run_type: type of DagRun
+ :param logical_date: the logical date
+ :param run_after: the date before which dag run won't start.
+ """
# _Ensure_ run_type is a DagRunType, not just a string from user code
- return DagRunType(run_type).generate_run_id(logical_date)
+ if logical_date:
+ return DagRunType(run_type).generate_run_id(run_after=logical_date.isoformat())
Review Comment:
Alternatively (to `run_after`) we can probably use an even vaguer word such as simply `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
Re: [PR] AIP-83 amendment: Add logic for generating run_id when logical date is None. [airflow]
Posted by "sunank200 (via GitHub)" <gi...@apache.org>.
sunank200 commented on code in PR #46616:
URL: https://github.com/apache/airflow/pull/46616#discussion_r1950583310
##########
kubernetes_tests/test_base.py:
##########
@@ -287,12 +289,13 @@ def start_job_in_kubernetes(self, dag_id, host):
result_json = self.start_dag(dag_id=dag_id, host=host)
dag_runs = result_json["dag_runs"]
assert len(dag_runs) > 0
- logical_date = None
+ run_after = None
Review Comment:
Changed 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
Re: [PR] AIP-83 amendment: Add logic for generating run_id when logical date is None. [airflow]
Posted by "dstandish (via GitHub)" <gi...@apache.org>.
dstandish commented on PR #46616:
URL: https://github.com/apache/airflow/pull/46616#issuecomment-2651110552
I think we need to make run_after non public. Feels like we should not expose this to the user as an option -- but rather, it's something that the scheduler should figure out based on the dag / dag run.
--
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
Re: [PR] AIP-83 amendment: Add logic for generating run_id when logical date is None. [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #46616:
URL: https://github.com/apache/airflow/pull/46616#discussion_r1950262249
##########
airflow/api_fastapi/core_api/datamodels/dag_run.py:
##########
@@ -82,9 +82,11 @@ class TriggerDAGRunPostBody(StrictBaseModel):
"""Trigger DAG Run Serializer for POST body."""
dag_run_id: str | None = None
- logical_date: AwareDatetime | None
data_interval_start: AwareDatetime | None = None
data_interval_end: AwareDatetime | None = None
+ logical_date: AwareDatetime | None = None
Review Comment:
```suggestion
logical_date: AwareDatetime | 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
Re: [PR] AIP-83 amendment: Add logic for generating run_id when logical date is None. [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #46616:
URL: https://github.com/apache/airflow/pull/46616#discussion_r1950410179
##########
kubernetes_tests/test_base.py:
##########
@@ -287,12 +292,12 @@ def start_job_in_kubernetes(self, dag_id, host):
result_json = self.start_dag(dag_id=dag_id, host=host)
dag_runs = result_json["dag_runs"]
assert len(dag_runs) > 0
- logical_date = None
+ run_after = None
dag_run_id = None
for dag_run in dag_runs:
if dag_run["dag_id"] == dag_id:
- logical_date = dag_run["logical_date"]
+ run_after = dag_run["run_after"]
dag_run_id = dag_run["dag_run_id"]
break
- assert logical_date is not None, f"No logical_date can be found for the dag with {dag_id}"
- return dag_run_id, logical_date
+ assert run_after is not None, f"No run_after can be found for the dag with {dag_id}"
+ return dag_run_id, run_after
Review Comment:
I don’t think this needs to change? Sicne we set logical_date in the request, logical date should not be None here. (We can probably change the error message).
--
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
Re: [PR] AIP-83 amendment: Add logic for generating run_id when logical date is None. [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #46616:
URL: https://github.com/apache/airflow/pull/46616#discussion_r1948947855
##########
airflow/api_fastapi/core_api/datamodels/dag_run.py:
##########
@@ -99,13 +100,15 @@ def check_data_intervals(cls, values):
@model_validator(mode="after")
def validate_dag_run_id(self):
if not self.dag_run_id:
- self.dag_run_id = DagRun.generate_run_id(DagRunType.MANUAL, self.logical_date)
+ self.dag_run_id = DagRun.generate_run_id(
+ run_type=DagRunType.MANUAL, logical_date=self.logical_date, run_after=self.run_after
+ )
return self
# Mypy issue https://github.com/python/mypy/issues/1362
@computed_field # type: ignore[misc]
@property
- def logical_date(self) -> datetime:
+ def run_after(self) -> datetime:
return timezone.utcnow()
Review Comment:
I think this should be a field instead (can be specified by user, defaults to utcnow).
--
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
Re: [PR] AIP-83 amendment: Add logic for generating run_id when logical date is None. [airflow]
Posted by "dstandish (via GitHub)" <gi...@apache.org>.
dstandish commented on code in PR #46616:
URL: https://github.com/apache/airflow/pull/46616#discussion_r1949880593
##########
airflow/api/common/trigger_dag.py:
##########
@@ -54,6 +55,7 @@ def _trigger_dag(
:param dag_id: DAG ID
:param dag_bag: DAG Bag model
:param triggered_by: the entity which triggers the dag_run
+ :param run_after: the datetime before which dag won't run.
Review Comment:
```suggestion
:param run_after: the datetime before which dag cannot run.
```
--
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
Re: [PR] AIP-83 amendment: Add logic for generating run_id when logical date is None. [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #46616:
URL: https://github.com/apache/airflow/pull/46616#discussion_r1950430586
##########
kubernetes_tests/test_base.py:
##########
@@ -264,8 +266,11 @@ def start_dag(self, dag_id, host):
assert result.status_code == 200, f"Could not enable DAG: {result_json}"
post_string = f"http://{host}/api/v1/dags/{dag_id}/dagRuns"
print(f"Calling [start_dag]#2 {post_string}")
+
# Trigger a new dagrun
- result = self.session.post(post_string, json={})
+ result = self.session.post(
+ post_string, json={"logical_date": timezone.coerce_datetime(timezone.utcnow())}
Review Comment:
Also we need to covert this to a string.
--
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
Re: [PR] AIP-83 amendment: Add logic for generating run_id when logical date is None. [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #46616:
URL: https://github.com/apache/airflow/pull/46616#discussion_r1950257155
##########
airflow/api/common/trigger_dag.py:
##########
@@ -65,26 +67,29 @@ def _trigger_dag(
if dag is None or dag_id not in dag_bag.dags:
raise DagNotFound(f"Dag id {dag_id} not found")
- logical_date = logical_date or timezone.utcnow()
-
- if not timezone.is_localized(logical_date):
- raise ValueError("The logical date should be localized")
-
- if replace_microseconds:
- logical_date = logical_date.replace(microsecond=0)
-
- if dag.default_args and "start_date" in dag.default_args:
- min_dag_start_date = dag.default_args["start_date"]
- if min_dag_start_date and logical_date < min_dag_start_date:
- raise ValueError(
- f"Logical date [{logical_date.isoformat()}] should be >= start_date "
- f"[{min_dag_start_date.isoformat()}] from DAG's default_args"
- )
- coerced_logical_date = timezone.coerce_datetime(logical_date)
-
- data_interval = dag.timetable.infer_manual_data_interval(run_after=coerced_logical_date)
- run_id = run_id or dag.timetable.generate_run_id(
- run_type=DagRunType.MANUAL, logical_date=coerced_logical_date, data_interval=data_interval
+ if logical_date:
+ if not timezone.is_localized(logical_date):
+ raise ValueError("The logical date should be localized")
+
+ if replace_microseconds:
+ logical_date = logical_date.replace(microsecond=0)
+
+ if dag.default_args and "start_date" in dag.default_args:
+ min_dag_start_date = dag.default_args["start_date"]
+ if min_dag_start_date and logical_date < min_dag_start_date:
+ raise ValueError(
+ f"Logical date [{logical_date.isoformat()}] should be >= start_date "
+ f"[{min_dag_start_date.isoformat()}] from DAG's default_args"
+ )
+ coerced_logical_date = timezone.coerce_datetime(logical_date)
+ data_interval = dag.timetable.infer_manual_data_interval(run_after=coerced_logical_date)
+ else:
+ data_interval = None
+
+ run_id = run_id or DagRun.generate_run_id(
+ run_type=DagRunType.MANUAL,
+ logical_date=timezone.coerce_datetime(logical_date),
+ run_after=timezone.coerce_datetime(run_after),
Review Comment:
```suggestion
else:
coerced_logical_date = None
data_interval = None
run_id = run_id or DagRun.generate_run_id(
run_type=DagRunType.MANUAL,
logical_date=coerced_logical_date,
run_after=timezone.coerce_datetime(run_after),
```
So we can avoid one coerce call.
--
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
Re: [PR] AIP-83 amendment: Add logic for generating run_id when logical date is None. [airflow]
Posted by "sunank200 (via GitHub)" <gi...@apache.org>.
sunank200 commented on code in PR #46616:
URL: https://github.com/apache/airflow/pull/46616#discussion_r1950264961
##########
airflow/api_connexion/schemas/dag_run_schema.py:
##########
@@ -78,17 +79,21 @@ class Meta:
@pre_load
def autogenerate(self, data, **kwargs):
- """Auto generate run_id and logical_date if they are not provided."""
- logical_date = data.get("logical_date", _MISSING)
+ """Auto generate run_id and run_after if they are not provided."""
+ run_after = data.get("run_after", _MISSING)
- # Auto-generate logical_date if missing
- if logical_date is _MISSING:
- data["logical_date"] = str(timezone.utcnow())
+ # Auto-generate run_after if missing
+ if run_after is _MISSING:
+ data["run_after"] = str(timezone.utcnow())
if "dag_run_id" not in data:
try:
data["dag_run_id"] = DagRun.generate_run_id(
- DagRunType.MANUAL, timezone.parse(data["logical_date"])
+ run_type=DagRunType.MANUAL,
+ logical_date=timezone.parse(data.get("logical_date"))
+ if data.get("logical_date")
+ else None,
+ run_after=timezone.parse(data["run_after"]),
Review Comment:
changed 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
Re: [PR] AIP-83 amendment: Add logic for generating run_id when logical date is None. [airflow]
Posted by "phanikumv (via GitHub)" <gi...@apache.org>.
phanikumv merged PR #46616:
URL: https://github.com/apache/airflow/pull/46616
--
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
Re: [PR] AIP-83 amendment: Add logic for generating run_id when logical date is None. [airflow]
Posted by "dstandish (via GitHub)" <gi...@apache.org>.
dstandish commented on code in PR #46616:
URL: https://github.com/apache/airflow/pull/46616#discussion_r1951036907
##########
airflow/api/common/trigger_dag.py:
##########
@@ -42,6 +42,7 @@ def _trigger_dag(
dag_bag: DagBag,
*,
triggered_by: DagRunTriggeredByType,
+ run_after: datetime,
Review Comment:
so like e.g. this, not sure it makes sense to have this param
--
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