Add support for using the "normal" Trigger classes for flow.serve and .deploy (#12789)

Author: chrisguidry

src/prefect/cli/deploy.py

@@ -7,22 +7,23 @@
 from datetime import timedelta
 from getpass import GetPassWarning
 from pathlib import Path
-from typing import Any, Dict, List, Optional, Tuple
+from typing import Any, Dict, List, Optional, Tuple, Union
 from uuid import UUID
 
+import typer
+import yaml
+from rich.console import Console
+from rich.panel import Panel
+from rich.table import Table
+from yaml.error import YAMLError
+
 from prefect._internal.pydantic import HAS_PYDANTIC_V2
 
 if HAS_PYDANTIC_V2:
     import pydantic.v1 as pydantic
 else:
     import pydantic
 
-import typer
-import yaml
-from rich.console import Console
-from rich.panel import Panel
-from rich.table import Table
-from yaml.error import YAMLError
 
 import prefect
 from prefect._internal.compatibility.deprecated import (
@@ -62,7 +63,7 @@
     _save_deployment_to_prefect_file,
 )
 from prefect.deployments.steps.core import run_steps
-from prefect.events import DeploymentTriggerTypes
+from prefect.events import DeploymentTriggerTypes, TriggerTypes
 from prefect.exceptions import ObjectNotFound, PrefectHTTPStatusError
 from prefect.flows import load_flow_from_entrypoint
 from prefect.settings import (
@@ -1602,7 +1603,9 @@ def _initialize_deployment_triggers(
 
 
 async def _create_deployment_triggers(
-    client: PrefectClient, deployment_id: UUID, triggers: List[DeploymentTriggerTypes]
+    client: PrefectClient,
+    deployment_id: UUID,
+    triggers: List[Union[DeploymentTriggerTypes, TriggerTypes]],
 ):
     if client.server_type.supports_automations():
         try:

src/prefect/deployments/deployments.py

@@ -16,14 +16,20 @@
 import pendulum
 import yaml
 
+from prefect._internal.pydantic import HAS_PYDANTIC_V2
+
+if HAS_PYDANTIC_V2:
+    from pydantic.v1 import BaseModel, Field, parse_obj_as, root_validator, validator
+else:
+    from pydantic import BaseModel, Field, parse_obj_as, root_validator, validator
+
 from prefect._internal.compatibility.deprecated import (
     DeprecatedInfraOverridesField,
     deprecated_callable,
     deprecated_class,
     deprecated_parameter,
     handle_deprecated_infra_overrides_parameter,
 )
-from prefect._internal.pydantic import HAS_PYDANTIC_V2
 from prefect._internal.schemas.validators import (
     handle_openapi_schema,
     infrastructure_must_have_capabilities,
@@ -32,16 +38,10 @@
     validate_automation_names,
     validate_deprecated_schedule_fields,
 )
-from prefect.client.schemas.actions import DeploymentScheduleCreate
-
-if HAS_PYDANTIC_V2:
-    from pydantic.v1 import BaseModel, Field, parse_obj_as, root_validator, validator
-else:
-    from pydantic import BaseModel, Field, parse_obj_as, root_validator, validator
-
 from prefect.blocks.core import Block
 from prefect.blocks.fields import SecretDict
 from prefect.client.orchestration import PrefectClient, get_client
+from prefect.client.schemas.actions import DeploymentScheduleCreate
 from prefect.client.schemas.objects import (
     FlowRun,
     MinimalDeploymentSchedule,
@@ -53,7 +53,7 @@
     FlexibleScheduleList,
 )
 from prefect.deployments.steps.core import run_steps
-from prefect.events import DeploymentTriggerTypes
+from prefect.events import DeploymentTriggerTypes, TriggerTypes
 from prefect.exceptions import (
     BlockMissingCapabilities,
     ObjectAlreadyExists,
@@ -610,7 +610,7 @@ def _validate_schedule(cls, value):
         description="The parameter schema of the flow, including defaults.",
     )
     timestamp: datetime = Field(default_factory=partial(pendulum.now, "UTC"))
-    triggers: List[DeploymentTriggerTypes] = Field(
+    triggers: List[Union[DeploymentTriggerTypes, TriggerTypes]] = Field(
         default_factory=list,
         description="The triggers that should cause this deployment to run.",
     )

src/prefect/deployments/runner.py

@@ -42,26 +42,19 @@ def fast_flow():
 from rich.progress import Progress, SpinnerColumn, TextColumn, track
 from rich.table import Table
 
-from prefect._internal.concurrency.api import create_call, from_async
 from prefect._internal.pydantic import HAS_PYDANTIC_V2
-from prefect._internal.schemas.validators import (
-    reconcile_paused_deployment,
-    reconcile_schedules_runner,
-    validate_automation_names,
-)
-from prefect.runner.storage import RunnerStorage
-from prefect.settings import (
-    PREFECT_DEFAULT_DOCKER_BUILD_NAMESPACE,
-    PREFECT_DEFAULT_WORK_POOL_NAME,
-    PREFECT_UI_URL,
-)
-from prefect.utilities.collections import get_from_dict, isiterable
 
 if HAS_PYDANTIC_V2:
     from pydantic.v1 import BaseModel, Field, PrivateAttr, root_validator, validator
 else:
     from pydantic import BaseModel, Field, PrivateAttr, root_validator, validator
 
+from prefect._internal.concurrency.api import create_call, from_async
+from prefect._internal.schemas.validators import (
+    reconcile_paused_deployment,
+    reconcile_schedules_runner,
+    validate_automation_names,
+)
 from prefect.client.orchestration import get_client
 from prefect.client.schemas.objects import MinimalDeploymentSchedule
 from prefect.client.schemas.schedules import (
@@ -72,13 +65,20 @@ def fast_flow():
     FlexibleScheduleList,
     create_minimal_deployment_schedule,
 )
-from prefect.events import DeploymentTriggerTypes
+from prefect.events import DeploymentTriggerTypes, TriggerTypes
 from prefect.exceptions import (
     ObjectNotFound,
     PrefectHTTPStatusError,
 )
+from prefect.runner.storage import RunnerStorage
+from prefect.settings import (
+    PREFECT_DEFAULT_DOCKER_BUILD_NAMESPACE,
+    PREFECT_DEFAULT_WORK_POOL_NAME,
+    PREFECT_UI_URL,
+)
 from prefect.utilities.asyncutils import sync_compatible
 from prefect.utilities.callables import ParameterSchema, parameter_schema
+from prefect.utilities.collections import get_from_dict, isiterable
 from prefect.utilities.dockerutils import (
     PushError,
     build_image,
@@ -179,7 +179,7 @@ class Config:
             "The path to the entrypoint for the workflow, relative to the `path`."
         ),
     )
-    triggers: List[DeploymentTriggerTypes] = Field(
+    triggers: List[Union[DeploymentTriggerTypes, TriggerTypes]] = Field(
         default_factory=list,
         description="The triggers that should cause this deployment to run.",
     )
@@ -454,7 +454,7 @@ def from_flow(
         schedule: Optional[SCHEDULE_TYPES] = None,
         is_schedule_active: Optional[bool] = None,
         parameters: Optional[dict] = None,
-        triggers: Optional[List[DeploymentTriggerTypes]] = None,
+        triggers: Optional[List[Union[DeploymentTriggerTypes, TriggerTypes]]] = None,
         description: Optional[str] = None,
         tags: Optional[List[str]] = None,
         version: Optional[str] = None,
@@ -590,7 +590,7 @@ def from_entrypoint(
         schedule: Optional[SCHEDULE_TYPES] = None,
         is_schedule_active: Optional[bool] = None,
         parameters: Optional[dict] = None,
-        triggers: Optional[List[DeploymentTriggerTypes]] = None,
+        triggers: Optional[List[Union[DeploymentTriggerTypes, TriggerTypes]]] = None,
         description: Optional[str] = None,
         tags: Optional[List[str]] = None,
         version: Optional[str] = None,
@@ -688,7 +688,7 @@ async def from_storage(
         schedule: Optional[SCHEDULE_TYPES] = None,
         is_schedule_active: Optional[bool] = None,
         parameters: Optional[dict] = None,
-        triggers: Optional[List[DeploymentTriggerTypes]] = None,
+        triggers: Optional[List[Union[DeploymentTriggerTypes, TriggerTypes]]] = None,
         description: Optional[str] = None,
         tags: Optional[List[str]] = None,
         version: Optional[str] = None,

src/prefect/events/schemas/automations.py

@@ -19,14 +19,14 @@
 from prefect._internal.schemas.validators import validate_trigger_within
 
 if HAS_PYDANTIC_V2:
-    from pydantic.v1 import Field, root_validator, validator
+    from pydantic.v1 import Field, PrivateAttr, root_validator, validator
     from pydantic.v1.fields import ModelField
 else:
-    from pydantic import Field, root_validator, validator
+    from pydantic import Field, PrivateAttr, root_validator, validator
     from pydantic.fields import ModelField
 
 from prefect._internal.schemas.bases import PrefectBaseModel
-from prefect.events.actions import ActionTypes
+from prefect.events.actions import ActionTypes, RunDeployment
 from prefect.utilities.collections import AutoEnum
 
 from .events import ResourceSpecification
@@ -50,6 +50,48 @@ class Trigger(PrefectBaseModel, abc.ABC, extra="ignore"):
     def describe_for_cli(self, indent: int = 0) -> str:
         """Return a human-readable description of this trigger for the CLI"""
 
+    # The following allows the regular Trigger class to be used when serving or
+    # deploying flows, analogous to how the Deployment*Trigger classes work
+
+    _deployment_id: Optional[UUID] = PrivateAttr(default=None)
+
+    def set_deployment_id(self, deployment_id: UUID):
+        self._deployment_id = deployment_id
+
+    def owner_resource(self) -> Optional[str]:
+        return f"prefect.deployment.{self._deployment_id}"
+
+    def actions(self) -> List[RunDeployment]:
+        assert self._deployment_id
+        return [
+            RunDeployment(
+                deployment_id=self._deployment_id,
+                parameters=getattr(self, "parameters", None),
+                job_variables=getattr(self, "job_variables", None),
+            )
+        ]
+
+    def as_automation(self) -> "AutomationCore":
+        assert self._deployment_id
+
+        trigger = self
+
+        # This is one of the Deployment*Trigger classes, so translate it over to a
+        # plain Trigger
+        if hasattr(self, "trigger_type"):
+            trigger = self.trigger_type(**self.dict())
+
+        return AutomationCore(
+            name=(
+                getattr(self, "name", None)
+                or f"Automation for deployment {self._deployment_id}"
+            ),
+            enabled=getattr(self, "enabled", True),
+            trigger=trigger,
+            actions=self.actions(),
+            owner_resource=self.owner_resource(),
+        )
+
 
 class ResourceTrigger(Trigger, abc.ABC):
     """