(airflow) 01/01: Apply suggestions from code review

[as...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

ash pushed a commit to branch resume-deferred-operator-tasksdk
in repository https://gitbox.apache.org/repos/asf/airflow.git

commit 857b22a57f60c4791c75072c7b882a781c02700e
Author: Ash Berlin-Taylor <as...@apache.org>
AuthorDate: Tue Feb 25 16:37:26 2025 +0000

    Apply suggestions from code review
    
    Co-authored-by: Amogh Desai <am...@gmail.com>
---
 airflow/api_fastapi/execution_api/datamodels/taskinstance.py | 2 +-
 task_sdk/src/airflow/sdk/definitions/baseoperator.py         | 4 +++-
 task_sdk/src/airflow/sdk/execution_time/supervisor.py        | 2 +-
 3 files changed, 5 insertions(+), 3 deletions(-)

diff --git a/airflow/api_fastapi/execution_api/datamodels/taskinstance.py b/airflow/api_fastapi/execution_api/datamodels/taskinstance.py
index 0bc01f49c5c..ba4573c9d38 100644
--- a/airflow/api_fastapi/execution_api/datamodels/taskinstance.py
+++ b/airflow/api_fastapi/execution_api/datamodels/taskinstance.py
@@ -123,7 +123,7 @@ class TIDeferredStatePayload(StrictBaseModel):
     classpath: str
     trigger_kwargs: Annotated[dict[str, Any] | str, Field(default_factory=dict)]
     """
-    Kwargs to pass to the trigger constructor, either a plain dict or an ecnrypted string.
+    Kwargs to pass to the trigger constructor, either a plain dict or an encrypted string.
 
     Both forms will be passed along to the trigger, the server will not handle either.
     """
diff --git a/task_sdk/src/airflow/sdk/definitions/baseoperator.py b/task_sdk/src/airflow/sdk/definitions/baseoperator.py
index 28e88d0c995..3efbd1ad0b6 100644
--- a/task_sdk/src/airflow/sdk/definitions/baseoperator.py
+++ b/task_sdk/src/airflow/sdk/definitions/baseoperator.py
@@ -1484,10 +1484,12 @@ class BaseOperator(AbstractOperator, metaclass=BaseOperatorMeta):
 
         raise TaskDeferred(trigger=trigger, method_name=method_name, kwargs=kwargs, timeout=timeout)
 
-    def resume_execution(self, next_method: str, next_kwargs: dict[str, Any], context: Context):
+    def resume_execution(self, next_method: str, next_kwargs: dict[str, Any] | None, context: Context):
         """Entrypoint method called by the Task Runner (instead of execute) when this task is resumed."""
         from airflow.exceptions import TaskDeferralError, TaskDeferralTimeout
 
+        if next_kwargs is None:
+            next_kwargs = {}
         # __fail__ is a special signal value for next_method that indicates
         # this task was scheduled specifically to fail.
 
diff --git a/task_sdk/src/airflow/sdk/execution_time/supervisor.py b/task_sdk/src/airflow/sdk/execution_time/supervisor.py
index 7201357c4d9..fada6105bce 100644
--- a/task_sdk/src/airflow/sdk/execution_time/supervisor.py
+++ b/task_sdk/src/airflow/sdk/execution_time/supervisor.py
@@ -963,7 +963,7 @@ def supervise(
     logger: FilteringBoundLogger | None = None
     if log_path:
         # If we are told to write logs to a file, redirect the task logger to it. Make sure we append to the
-        # file though, otherwise when we resume we would loose the logs from the start->deferral segment if it
+        # file though, otherwise when we resume we would lose the logs from the start->deferral segment if it
         # lands on the same node as before.
         from airflow.sdk.log import init_log_file, logging_processors