Raise type completeness to 95% (#16723)

src/prefect/docker/docker_image.py

@@ -1,5 +1,5 @@
 from pathlib import Path
-from typing import Optional
+from typing import Any, Optional
 
 from pendulum import now as pendulum_now
 
@@ -34,7 +34,11 @@ class DockerImage:
     """
 
     def __init__(
-        self, name: str, tag: Optional[str] = None, dockerfile="auto", **build_kwargs
+        self,
+        name: str,
+        tag: Optional[str] = None,
+        dockerfile: str = "auto",
+        **build_kwargs: Any,
     ):
         image_name, image_tag = parse_image_tag(name)
         if tag and image_tag:
@@ -49,16 +53,16 @@ def __init__(
             namespace = PREFECT_DEFAULT_DOCKER_BUILD_NAMESPACE.value()
         # join the namespace and repository to create the full image name
         # ignore namespace if it is None
-        self.name = "/".join(filter(None, [namespace, repository]))
-        self.tag = tag or image_tag or slugify(pendulum_now("utc").isoformat())
-        self.dockerfile = dockerfile
-        self.build_kwargs = build_kwargs
+        self.name: str = "/".join(filter(None, [namespace, repository]))
+        self.tag: str = tag or image_tag or slugify(pendulum_now("utc").isoformat())
+        self.dockerfile: str = dockerfile
+        self.build_kwargs: dict[str, Any] = build_kwargs
 
     @property
-    def reference(self):
+    def reference(self) -> str:
         return f"{self.name}:{self.tag}"
 
-    def build(self):
+    def build(self) -> None:
         full_image_name = self.reference
         build_kwargs = self.build_kwargs.copy()
         build_kwargs["context"] = Path.cwd()
@@ -72,7 +76,7 @@ def build(self):
             build_kwargs["dockerfile"] = self.dockerfile
             build_image(**build_kwargs)
 
-    def push(self):
+    def push(self) -> None:
         with docker_client() as client:
             events = client.api.push(
                 repository=self.name, tag=self.tag, stream=True, decode=True

src/prefect/engine.py

@@ -1,6 +1,6 @@
 import os
 import sys
-from typing import Any, Callable
+from typing import TYPE_CHECKING, Any, Callable
 from uuid import UUID
 
 from prefect._internal.compatibility.migration import getattr_migration
@@ -15,12 +15,19 @@
     run_coro_as_sync,
 )
 
-engine_logger = get_logger("engine")
+if TYPE_CHECKING:
+    import logging
+
+    from prefect.flow_engine import FlowRun
+    from prefect.flows import Flow
+    from prefect.logging.loggers import LoggingAdapter
+
+engine_logger: "logging.Logger" = get_logger("engine")
 
 
 if __name__ == "__main__":
     try:
-        flow_run_id = UUID(
+        flow_run_id: UUID = UUID(
             sys.argv[1] if len(sys.argv) > 1 else os.environ.get("PREFECT__FLOW_RUN_ID")
         )
     except Exception:
@@ -37,11 +44,11 @@
             run_flow,
         )
 
-        flow_run = load_flow_run(flow_run_id=flow_run_id)
-        run_logger = flow_run_logger(flow_run=flow_run)
+        flow_run: "FlowRun" = load_flow_run(flow_run_id=flow_run_id)
+        run_logger: "LoggingAdapter" = flow_run_logger(flow_run=flow_run)
 
         try:
-            flow = load_flow(flow_run)
+            flow: "Flow[..., Any]" = load_flow(flow_run)
         except Exception:
             run_logger.error(
                 "Unexpected exception encountered when trying to load flow",
@@ -55,15 +62,17 @@
         else:
             run_flow(flow, flow_run=flow_run, error_logger=run_logger)
 
-    except Abort as exc:
+    except Abort as abort_signal:
+        abort_signal: Abort
         engine_logger.info(
             f"Engine execution of flow run '{flow_run_id}' aborted by orchestrator:"
-            f" {exc}"
+            f" {abort_signal}"
         )
         exit(0)
-    except Pause as exc:
+    except Pause as pause_signal:
+        pause_signal: Pause
         engine_logger.info(
-            f"Engine execution of flow run '{flow_run_id}' is paused: {exc}"
+            f"Engine execution of flow run '{flow_run_id}' is paused: {pause_signal}"
         )
         exit(0)
     except Exception:

src/prefect/events/schemas/automations.py

@@ -52,7 +52,7 @@ def describe_for_cli(self, indent: int = 0) -> str:
 
     _deployment_id: Optional[UUID] = PrivateAttr(default=None)
 
-    def set_deployment_id(self, deployment_id: UUID):
+    def set_deployment_id(self, deployment_id: UUID) -> None:
         self._deployment_id = deployment_id
 
     def owner_resource(self) -> Optional[str]:
@@ -277,7 +277,7 @@ class MetricTriggerQuery(PrefectBaseModel):
     )
 
     @field_validator("range", "firing_for")
-    def enforce_minimum_range(cls, value: timedelta):
+    def enforce_minimum_range(cls, value: timedelta) -> timedelta:
         if value < timedelta(seconds=300):
             raise ValueError("The minimum range is 300 seconds (5 minutes)")
         return value
@@ -404,21 +404,25 @@ class AutomationCore(PrefectBaseModel, extra="ignore"):  # type: ignore[call-arg
     """Defines an action a user wants to take when a certain number of events
     do or don't happen to the matching resources"""
 
-    name: str = Field(..., description="The name of this automation")
-    description: str = Field("", description="A longer description of this automation")
+    name: str = Field(default=..., description="The name of this automation")
+    description: str = Field(
+        default="", description="A longer description of this automation"
+    )
 
-    enabled: bool = Field(True, description="Whether this automation will be evaluated")
+    enabled: bool = Field(
+        default=True, description="Whether this automation will be evaluated"
+    )
 
     trigger: TriggerTypes = Field(
-        ...,
+        default=...,
         description=(
             "The criteria for which events this Automation covers and how it will "
             "respond to the presence or absence of those events"
         ),
     )
 
     actions: List[ActionTypes] = Field(
-        ...,
+        default=...,
         description="The actions to perform when this Automation triggers",
     )
 
@@ -438,4 +442,4 @@ class AutomationCore(PrefectBaseModel, extra="ignore"):  # type: ignore[call-arg
 
 
 class Automation(AutomationCore):
-    id: UUID = Field(..., description="The ID of this automation")
+    id: UUID = Field(default=..., description="The ID of this automation")