jedcunningham commented on code in PR #24912:
URL: https://github.com/apache/airflow/pull/24912#discussion_r916446383
##########
airflow/jobs/scheduler_job.py:
##########
@@ -790,7 +794,7 @@ def _update_dag_run_state_for_paused_dags(self):
dag = SerializedDagModel.get_dag(dag_id)
if dag is None:
continue
- dag_runs = DagRun.find(dag_id=dag_id, state=State.RUNNING)
+ dag_runs = DagRun.find(dag_id=dag_id,
state=DagRunState.RUNNING)
Review Comment:
Also here, using the wrong one.
##########
airflow/models/taskinstance.py:
##########
@@ -790,11 +791,11 @@ def log_url(self):
)
@property
- def mark_success_url(self):
+ def mark_success_url(self) -> str:
"""URL to mark TI success"""
base_url = conf.get('webserver', 'BASE_URL')
- return base_url + (
- "/confirm"
+ return (
+ f"{base_url}/confirm"
Review Comment:
This avoids the "Optional" aspect of `conf.get`, plus I think it's easier to
read.
##########
airflow/jobs/scheduler_job.py:
##########
@@ -444,9 +448,9 @@ def _executable_task_instances_to_queued(self, max_tis:
int, session: Session =
dag_id,
task_instance,
)
- session.query(TI).filter(TI.dag_id == dag_id, TI.state
== State.SCHEDULED).update(
- {TI.state: State.FAILED},
synchronize_session='fetch'
- )
+ session.query(TI).filter(
+ TI.dag_id == dag_id, TI.state ==
TaskInstanceState.SCHEDULED
+ ).update({TI.state: TaskInstanceState.FAILED},
synchronize_session='fetch')
Review Comment:
Technically just a typing change, but we were using the wrong 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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
