Rework the TriggererJobRunner to run triggers in a process without DB access

.github/boring-cyborg.yml

@@ -405,7 +405,6 @@ labelPRBasedOnFilePath:
     - tests/cli/commands/local_commands/test_triggerer_command.py
     - tests/jobs/test_triggerer_job.py
     - tests/models/test_trigger.py
-    - tests/jobs/test_triggerer_job_logging.py
     - providers/standard/tests/unit/standard/triggers/**/*
 
   area:Serialization:

airflow/dag_processing/processor.py

@@ -20,7 +20,7 @@
 import sys
 import traceback
 from pathlib import Path
-from typing import TYPE_CHECKING, Annotated, Callable, Literal, Union
+from typing import TYPE_CHECKING, Annotated, Callable, ClassVar, Literal, Union
 
 import attrs
 from pydantic import BaseModel, Field, TypeAdapter
@@ -207,7 +207,7 @@ class DagFileProcessorProcess(WatchedSubprocess):
     """
 
     parsing_result: DagFileParsingResult | None = None
-    decoder: TypeAdapter[ToParent] = TypeAdapter[ToParent](ToParent)
+    decoder: ClassVar[TypeAdapter[ToParent]] = TypeAdapter[ToParent](ToParent)
 
     @classmethod
     def start(  # type: ignore[override]

airflow/executors/local_executor.py

@@ -36,15 +36,14 @@
 from setproctitle import setproctitle
 
 from airflow import settings
+from airflow.executors import workloads
 from airflow.executors.base_executor import PARALLELISM, BaseExecutor
 from airflow.utils.session import NEW_SESSION, provide_session
 from airflow.utils.state import TaskInstanceState
 
 if TYPE_CHECKING:
     from sqlalchemy.orm import Session
 
-    from airflow.executors import workloads
-
     TaskInstanceStateType = tuple[workloads.TaskInstance, TaskInstanceState, Optional[Exception]]
 
 
@@ -82,6 +81,9 @@ def _run_worker(
             # Received poison pill, no more tasks to run
             return
 
+        if not isinstance(workload, workloads.ExecuteTask):
+            raise ValueError(f"LocalExecutor does not now how to handle {type(workload)}")
+
         # Decrement this as soon as we pick up a message off the queue
         with unread_messages:
             unread_messages.value -= 1

airflow/executors/workloads.py

@@ -18,8 +18,9 @@
 
 import os
 import uuid
+from datetime import datetime
 from pathlib import Path
-from typing import TYPE_CHECKING, Literal, Union
+from typing import TYPE_CHECKING, Annotated, Literal, Union
 
 from pydantic import BaseModel, Field
 
@@ -106,4 +107,33 @@ def make(cls, ti: TIModel, dag_rel_path: Path | None = None) -> ExecuteTask:
         return cls(ti=ser_ti, dag_rel_path=path, token="", log_path=fname, bundle_info=bundle_info)
 
 
-All = Union[ExecuteTask]
+class RunTrigger(BaseModel):
+    """Execute an async "trigger" process that yields events."""
+
+    id: int
+
+    ti: TaskInstance | None
+    """
+    The task instance associated with this trigger.
+
+    Could be none for asset-based triggers.
+    """
+
+    classpath: str
+    """
+    Dot-separated name of the module+fn to import and run this workload.
+
+    Consumers of this Workload must perform their own validation of this input.
+    """
+
+    encrypted_kwargs: str
+
+    timeout_after: datetime | None = None
+
+    kind: Literal["RunTrigger"] = Field(init=False, default="RunTrigger")
+
+
+All = Annotated[
+    Union[ExecuteTask, RunTrigger],
+    Field(discriminator="kind"),
+]