prefect.workers.base

BaseJobConfiguration

Bases: BaseModel

Source code in src/prefect/workers/base.py

class BaseJobConfiguration(BaseModel):
    command: Optional[str] = Field(
        default=None,
        description=(
            "The command to use when starting a flow run. "
            "In most cases, this should be left blank and the command "
            "will be automatically generated by the worker."
        ),
    )
    env: Dict[str, Optional[str]] = Field(
        default_factory=dict,
        title="Environment Variables",
        description="Environment variables to set when starting a flow run.",
    )
    labels: Dict[str, str] = Field(
        default_factory=dict,
        description=(
            "Labels applied to infrastructure created by the worker using "
            "this job configuration."
        ),
    )
    name: Optional[str] = Field(
        default=None,
        description=(
            "Name given to infrastructure created by the worker using this "
            "job configuration."
        ),
    )

    _related_objects: Dict[str, Any] = PrivateAttr(default_factory=dict)

    @property
    def is_using_a_runner(self):
        return self.command is not None and "prefect flow-run execute" in self.command

    @field_validator("command")
    @classmethod
    def _coerce_command(cls, v):
        return return_v_or_none(v)

    @field_validator("env", mode="before")
    @classmethod
    def _coerce_env(cls, v):
        return {k: str(v) if v is not None else None for k, v in v.items()}

    @staticmethod
    def _get_base_config_defaults(variables: dict) -> dict:
        """Get default values from base config for all variables that have them."""
        defaults = dict()
        for variable_name, attrs in variables.items():
            # We remote `None` values because we don't want to use them in templating.
            # The currently logic depends on keys not existing to populate the correct value
            # in some cases.
            # Pydantic will provide default values if the keys are missing when creating
            # a configuration class.
            if "default" in attrs and attrs.get("default") is not None:
                defaults[variable_name] = attrs["default"]

        return defaults

    @classmethod
    @inject_client
    async def from_template_and_values(
        cls,
        base_job_template: dict,
        values: dict,
        client: Optional["PrefectClient"] = None,
    ):
        """Creates a valid worker configuration object from the provided base
        configuration and overrides.

        Important: this method expects that the base_job_template was already
        validated server-side.
        """
        job_config: Dict[str, Any] = base_job_template["job_configuration"]
        variables_schema = base_job_template["variables"]
        variables = cls._get_base_config_defaults(
            variables_schema.get("properties", {})
        )

        # copy variable defaults for `env` to job config before they're replaced by
        # deployment overrides
        if variables.get("env"):
            job_config["env"] = variables.get("env")

        variables.update(values)

        # deep merge `env`
        if isinstance(job_config.get("env"), dict) and (
            hardcoded_env := variables.get("env")
        ):
            job_config["env"] = hardcoded_env | job_config.get("env")

        populated_configuration = apply_values(template=job_config, values=variables)
        populated_configuration = await resolve_block_document_references(
            template=populated_configuration, client=client
        )
        populated_configuration = await resolve_variables(
            template=populated_configuration, client=client
        )
        return cls(**populated_configuration)

    @classmethod
    def json_template(cls) -> dict:
        """Returns a dict with job configuration as keys and the corresponding templates as values

        Defaults to using the job configuration parameter name as the template variable name.

        e.g.
        {
            key1: '{{ key1 }}',     # default variable template
            key2: '{{ template2 }}', # `template2` specifically provide as template
        }
        """
        configuration = {}
        properties = cls.model_json_schema()["properties"]
        for k, v in properties.items():
            if v.get("template"):
                template = v["template"]
            else:
                template = "{{ " + k + " }}"
            configuration[k] = template

        return configuration

    def prepare_for_flow_run(
        self,
        flow_run: "FlowRun",
        deployment: Optional["DeploymentResponse"] = None,
        flow: Optional["Flow"] = None,
    ):
        """
        Prepare the job configuration for a flow run.

        This method is called by the worker before starting a flow run. It
        should be used to set any configuration values that are dependent on
        the flow run.

        Args:
            flow_run: The flow run to be executed.
            deployment: The deployment that the flow run is associated with.
            flow: The flow that the flow run is associated with.
        """

        self._related_objects = {
            "deployment": deployment,
            "flow": flow,
            "flow-run": flow_run,
        }
        if deployment is not None:
            deployment_labels = self._base_deployment_labels(deployment)
        else:
            deployment_labels = {}

        if flow is not None:
            flow_labels = self._base_flow_labels(flow)
        else:
            flow_labels = {}

        env = {
            **self._base_environment(),
            **self._base_flow_run_environment(flow_run),
            **(self.env if isinstance(self.env, dict) else {}),
        }
        self.env = {key: value for key, value in env.items() if value is not None}
        self.labels = {
            **self._base_flow_run_labels(flow_run),
            **deployment_labels,
            **flow_labels,
            **self.labels,
        }
        self.name = self.name or flow_run.name
        self.command = self.command or self._base_flow_run_command()

    @staticmethod
    def _base_flow_run_command() -> str:
        """
        Generate a command for a flow run job.
        """
        return "prefect flow-run execute"

    @staticmethod
    def _base_flow_run_labels(flow_run: "FlowRun") -> Dict[str, str]:
        """
        Generate a dictionary of labels for a flow run job.
        """
        return {
            "prefect.io/flow-run-id": str(flow_run.id),
            "prefect.io/flow-run-name": flow_run.name,
            "prefect.io/version": prefect.__version__,
        }

    @classmethod
    def _base_environment(cls) -> Dict[str, str]:
        """
        Environment variables that should be passed to all created infrastructure.

        These values should be overridable with the `env` field.
        """
        return get_current_settings().to_environment_variables(exclude_unset=True)

    @staticmethod
    def _base_flow_run_environment(flow_run: "FlowRun") -> Dict[str, str]:
        """
        Generate a dictionary of environment variables for a flow run job.
        """
        return {"PREFECT__FLOW_RUN_ID": str(flow_run.id)}

    @staticmethod
    def _base_deployment_labels(deployment: "DeploymentResponse") -> Dict[str, str]:
        labels = {
            "prefect.io/deployment-id": str(deployment.id),
            "prefect.io/deployment-name": deployment.name,
        }
        if deployment.updated is not None:
            labels["prefect.io/deployment-updated"] = deployment.updated.in_timezone(
                "utc"
            ).to_iso8601_string()
        return labels

    @staticmethod
    def _base_flow_labels(flow: "Flow") -> Dict[str, str]:
        return {
            "prefect.io/flow-id": str(flow.id),
            "prefect.io/flow-name": flow.name,
        }

    def _related_resources(self) -> List[RelatedResource]:
        tags = set()
        related = []

        for kind, obj in self._related_objects.items():
            if obj is None:
                continue
            if hasattr(obj, "tags"):
                tags.update(obj.tags)
            related.append(object_as_related_resource(kind=kind, role=kind, object=obj))

        return related + tags_as_related_resources(tags)

from_template_and_values(base_job_template, values, client=None) async classmethod

Creates a valid worker configuration object from the provided base configuration and overrides.

Important: this method expects that the base_job_template was already validated server-side.

Source code in src/prefect/workers/base.py

@classmethod
@inject_client
async def from_template_and_values(
    cls,
    base_job_template: dict,
    values: dict,
    client: Optional["PrefectClient"] = None,
):
    """Creates a valid worker configuration object from the provided base
    configuration and overrides.

    Important: this method expects that the base_job_template was already
    validated server-side.
    """
    job_config: Dict[str, Any] = base_job_template["job_configuration"]
    variables_schema = base_job_template["variables"]
    variables = cls._get_base_config_defaults(
        variables_schema.get("properties", {})
    )

    # copy variable defaults for `env` to job config before they're replaced by
    # deployment overrides
    if variables.get("env"):
        job_config["env"] = variables.get("env")

    variables.update(values)

    # deep merge `env`
    if isinstance(job_config.get("env"), dict) and (
        hardcoded_env := variables.get("env")
    ):
        job_config["env"] = hardcoded_env | job_config.get("env")

    populated_configuration = apply_values(template=job_config, values=variables)
    populated_configuration = await resolve_block_document_references(
        template=populated_configuration, client=client
    )
    populated_configuration = await resolve_variables(
        template=populated_configuration, client=client
    )
    return cls(**populated_configuration)

json_template() classmethod

Returns a dict with job configuration as keys and the corresponding templates as values

Defaults to using the job configuration parameter name as the template variable name.

e.g. { key1: '{{ key1 }}', # default variable template key2: '{{ template2 }}', # template2 specifically provide as template }

Source code in src/prefect/workers/base.py

@classmethod
def json_template(cls) -> dict:
    """Returns a dict with job configuration as keys and the corresponding templates as values

    Defaults to using the job configuration parameter name as the template variable name.

    e.g.
    {
        key1: '{{ key1 }}',     # default variable template
        key2: '{{ template2 }}', # `template2` specifically provide as template
    }
    """
    configuration = {}
    properties = cls.model_json_schema()["properties"]
    for k, v in properties.items():
        if v.get("template"):
            template = v["template"]
        else:
            template = "{{ " + k + " }}"
        configuration[k] = template

    return configuration

prepare_for_flow_run(flow_run, deployment=None, flow=None)

Prepare the job configuration for a flow run.

This method is called by the worker before starting a flow run. It should be used to set any configuration values that are dependent on the flow run.

Parameters:

Name	Type	Description	Default
flow_run	FlowRun	The flow run to be executed.	required
deployment	Optional[DeploymentResponse]	The deployment that the flow run is associated with.	None
flow	Optional[Flow]	The flow that the flow run is associated with.	None

Source code in src/prefect/workers/base.py

def prepare_for_flow_run(
    self,
    flow_run: "FlowRun",
    deployment: Optional["DeploymentResponse"] = None,
    flow: Optional["Flow"] = None,
):
    """
    Prepare the job configuration for a flow run.

    This method is called by the worker before starting a flow run. It
    should be used to set any configuration values that are dependent on
    the flow run.

    Args:
        flow_run: The flow run to be executed.
        deployment: The deployment that the flow run is associated with.
        flow: The flow that the flow run is associated with.
    """

    self._related_objects = {
        "deployment": deployment,
        "flow": flow,
        "flow-run": flow_run,
    }
    if deployment is not None:
        deployment_labels = self._base_deployment_labels(deployment)
    else:
        deployment_labels = {}

    if flow is not None:
        flow_labels = self._base_flow_labels(flow)
    else:
        flow_labels = {}

    env = {
        **self._base_environment(),
        **self._base_flow_run_environment(flow_run),
        **(self.env if isinstance(self.env, dict) else {}),
    }
    self.env = {key: value for key, value in env.items() if value is not None}
    self.labels = {
        **self._base_flow_run_labels(flow_run),
        **deployment_labels,
        **flow_labels,
        **self.labels,
    }
    self.name = self.name or flow_run.name
    self.command = self.command or self._base_flow_run_command()

BaseVariables

Bases: BaseModel

Source code in src/prefect/workers/base.py

class BaseVariables(BaseModel):
    name: Optional[str] = Field(
        default=None,
        description="Name given to infrastructure created by a worker.",
    )
    env: Dict[str, Optional[str]] = Field(
        default_factory=dict,
        title="Environment Variables",
        description="Environment variables to set when starting a flow run.",
    )
    labels: Dict[str, str] = Field(
        default_factory=dict,
        description="Labels applied to infrastructure created by a worker.",
    )
    command: Optional[str] = Field(
        default=None,
        description=(
            "The command to use when starting a flow run. "
            "In most cases, this should be left blank and the command "
            "will be automatically generated by the worker."
        ),
    )

    @classmethod
    def model_json_schema(
        cls,
        by_alias: bool = True,
        ref_template: str = "#/definitions/{model}",
        schema_generator: Type[GenerateJsonSchema] = GenerateJsonSchema,
        mode: Literal["validation", "serialization"] = "validation",
    ) -> Dict[str, Any]:
        """TODO: stop overriding this method - use GenerateSchema in ConfigDict instead?"""
        schema = super().model_json_schema(
            by_alias, ref_template, schema_generator, mode
        )

        # ensure backwards compatibility by copying $defs into definitions
        if "$defs" in schema:
            schema["definitions"] = schema.pop("$defs")

        # we aren't expecting these additional fields in the schema
        if "additionalProperties" in schema:
            schema.pop("additionalProperties")

        for _, definition in schema.get("definitions", {}).items():
            if "additionalProperties" in definition:
                definition.pop("additionalProperties")

        return schema

model_json_schema(by_alias=True, ref_template='#/definitions/{model}', schema_generator=GenerateJsonSchema, mode='validation') classmethod

Source code in src/prefect/workers/base.py

@classmethod
def model_json_schema(
    cls,
    by_alias: bool = True,
    ref_template: str = "#/definitions/{model}",
    schema_generator: Type[GenerateJsonSchema] = GenerateJsonSchema,
    mode: Literal["validation", "serialization"] = "validation",
) -> Dict[str, Any]:
    """TODO: stop overriding this method - use GenerateSchema in ConfigDict instead?"""
    schema = super().model_json_schema(
        by_alias, ref_template, schema_generator, mode
    )

    # ensure backwards compatibility by copying $defs into definitions
    if "$defs" in schema:
        schema["definitions"] = schema.pop("$defs")

    # we aren't expecting these additional fields in the schema
    if "additionalProperties" in schema:
        schema.pop("additionalProperties")

    for _, definition in schema.get("definitions", {}).items():
        if "additionalProperties" in definition:
            definition.pop("additionalProperties")

    return schema

BaseWorker

Bases: ABC

Source code in src/prefect/workers/base.py

@register_base_type
class BaseWorker(abc.ABC):
    type: str
    job_configuration: Type[BaseJobConfiguration] = BaseJobConfiguration
    job_configuration_variables: Optional[Type[BaseVariables]] = None

    _documentation_url = ""
    _logo_url = ""
    _description = ""

    def __init__(
        self,
        work_pool_name: str,
        work_queues: Optional[List[str]] = None,
        name: Optional[str] = None,
        prefetch_seconds: Optional[float] = None,
        create_pool_if_not_found: bool = True,
        limit: Optional[int] = None,
        heartbeat_interval_seconds: Optional[int] = None,
        *,
        base_job_template: Optional[Dict[str, Any]] = None,
    ):
        """
        Base class for all Prefect workers.

        Args:
            name: The name of the worker. If not provided, a random one
                will be generated. If provided, it cannot contain '/' or '%'.
                The name is used to identify the worker in the UI; if two
                processes have the same name, they will be treated as the same
                worker.
            work_pool_name: The name of the work pool to poll.
            work_queues: A list of work queues to poll. If not provided, all
                work queue in the work pool will be polled.
            prefetch_seconds: The number of seconds to prefetch flow runs for.
            create_pool_if_not_found: Whether to create the work pool
                if it is not found. Defaults to `True`, but can be set to `False` to
                ensure that work pools are not created accidentally.
            limit: The maximum number of flow runs this worker should be running at
                a given time.
            heartbeat_interval_seconds: The number of seconds between worker heartbeats.
            base_job_template: If creating the work pool, provide the base job
                template to use. Logs a warning if the pool already exists.
        """
        if name and ("/" in name or "%" in name):
            raise ValueError("Worker name cannot contain '/' or '%'")
        self.name = name or f"{self.__class__.__name__} {uuid4()}"
        self._started_event: Optional[Event] = None
        self._logger = get_logger(f"worker.{self.__class__.type}.{self.name.lower()}")

        self.is_setup = False
        self._create_pool_if_not_found = create_pool_if_not_found
        self._base_job_template = base_job_template
        self._work_pool_name = work_pool_name
        self._work_queues: Set[str] = set(work_queues) if work_queues else set()

        self._prefetch_seconds: float = (
            prefetch_seconds or PREFECT_WORKER_PREFETCH_SECONDS.value()
        )
        self.heartbeat_interval_seconds = (
            heartbeat_interval_seconds or PREFECT_WORKER_HEARTBEAT_SECONDS.value()
        )

        self.backend_id: Optional[UUID] = None
        self._work_pool: Optional[WorkPool] = None
        self._exit_stack: AsyncExitStack = AsyncExitStack()
        self._runs_task_group: Optional[anyio.abc.TaskGroup] = None
        self._client: Optional[PrefectClient] = None
        self._last_polled_time: pendulum.DateTime = pendulum.now("utc")
        self._limit = limit
        self._limiter: Optional[anyio.CapacityLimiter] = None
        self._submitting_flow_run_ids = set()
        self._cancelling_flow_run_ids = set()
        self._scheduled_task_scopes = set()

    @classmethod
    def get_documentation_url(cls) -> str:
        return cls._documentation_url

    @classmethod
    def get_logo_url(cls) -> str:
        return cls._logo_url

    @classmethod
    def get_description(cls) -> str:
        return cls._description

    @classmethod
    def get_default_base_job_template(cls) -> Dict:
        if cls.job_configuration_variables is None:
            schema = cls.job_configuration.model_json_schema()
            # remove "template" key from all dicts in schema['properties'] because it is not a
            # relevant field
            for key, value in schema["properties"].items():
                if isinstance(value, dict):
                    schema["properties"][key].pop("template", None)
            variables_schema = schema
        else:
            variables_schema = cls.job_configuration_variables.model_json_schema()
        variables_schema.pop("title", None)
        return {
            "job_configuration": cls.job_configuration.json_template(),
            "variables": variables_schema,
        }

    @staticmethod
    def get_worker_class_from_type(type: str) -> Optional[Type["BaseWorker"]]:
        """
        Returns the worker class for a given worker type. If the worker type
        is not recognized, returns None.
        """
        load_prefect_collections()
        worker_registry = get_registry_for_type(BaseWorker)
        if worker_registry is not None:
            return worker_registry.get(type)

    @staticmethod
    def get_all_available_worker_types() -> List[str]:
        """
        Returns all worker types available in the local registry.
        """
        load_prefect_collections()
        worker_registry = get_registry_for_type(BaseWorker)
        if worker_registry is not None:
            return list(worker_registry.keys())
        return []

    def get_name_slug(self):
        return slugify(self.name)

    def get_flow_run_logger(self, flow_run: "FlowRun") -> PrefectLogAdapter:
        return flow_run_logger(flow_run=flow_run).getChild(
            "worker",
            extra={
                "worker_name": self.name,
                "work_pool_name": (
                    self._work_pool_name if self._work_pool else "<unknown>"
                ),
                "work_pool_id": str(getattr(self._work_pool, "id", "unknown")),
            },
        )

    async def start(
        self,
        run_once: bool = False,
        with_healthcheck: bool = False,
        printer: Callable[..., None] = print,
    ):
        """
        Starts the worker and runs the main worker loops.

        By default, the worker will run loops to poll for scheduled/cancelled flow
        runs and sync with the Prefect API server.

        If `run_once` is set, the worker will only run each loop once and then return.

        If `with_healthcheck` is set, the worker will start a healthcheck server which
        can be used to determine if the worker is still polling for flow runs and restart
        the worker if necessary.

        Args:
            run_once: If set, the worker will only run each loop once then return.
            with_healthcheck: If set, the worker will start a healthcheck server.
            printer: A `print`-like function where logs will be reported.
        """
        healthcheck_server = None
        healthcheck_thread = None
        try:
            async with self as worker:
                # wait for an initial heartbeat to configure the worker
                await worker.sync_with_backend()
                # schedule the scheduled flow run polling loop
                async with anyio.create_task_group() as loops_task_group:
                    loops_task_group.start_soon(
                        partial(
                            critical_service_loop,
                            workload=self.get_and_submit_flow_runs,
                            interval=PREFECT_WORKER_QUERY_SECONDS.value(),
                            run_once=run_once,
                            jitter_range=0.3,
                            backoff=4,  # Up to ~1 minute interval during backoff
                        )
                    )
                    # schedule the sync loop
                    loops_task_group.start_soon(
                        partial(
                            critical_service_loop,
                            workload=self.sync_with_backend,
                            interval=self.heartbeat_interval_seconds,
                            run_once=run_once,
                            jitter_range=0.3,
                            backoff=4,
                        )
                    )

                    self._started_event = await self._emit_worker_started_event()

                    if with_healthcheck:
                        from prefect.workers.server import build_healthcheck_server

                        # we'll start the ASGI server in a separate thread so that
                        # uvicorn does not block the main thread
                        healthcheck_server = build_healthcheck_server(
                            worker=worker,
                            query_interval_seconds=PREFECT_WORKER_QUERY_SECONDS.value(),
                        )
                        healthcheck_thread = threading.Thread(
                            name="healthcheck-server-thread",
                            target=healthcheck_server.run,
                            daemon=True,
                        )
                        healthcheck_thread.start()
                    printer(f"Worker {worker.name!r} started!")
        finally:
            if healthcheck_server and healthcheck_thread:
                self._logger.debug("Stopping healthcheck server...")
                healthcheck_server.should_exit = True
                healthcheck_thread.join()
                self._logger.debug("Healthcheck server stopped.")

        printer(f"Worker {worker.name!r} stopped!")

    @abc.abstractmethod
    async def run(
        self,
        flow_run: "FlowRun",
        configuration: BaseJobConfiguration,
        task_status: Optional[anyio.abc.TaskStatus] = None,
    ) -> BaseWorkerResult:
        """
        Runs a given flow run on the current worker.
        """
        raise NotImplementedError(
            "Workers must implement a method for running submitted flow runs"
        )

    @classmethod
    def __dispatch_key__(cls):
        if cls.__name__ == "BaseWorker":
            return None  # The base class is abstract
        return cls.type

    async def setup(self):
        """Prepares the worker to run."""
        self._logger.debug("Setting up worker...")
        self._runs_task_group = anyio.create_task_group()
        self._limiter = (
            anyio.CapacityLimiter(self._limit) if self._limit is not None else None
        )

        if not PREFECT_TEST_MODE and not PREFECT_API_URL.value():
            raise ValueError("`PREFECT_API_URL` must be set to start a Worker.")

        self._client = get_client()
        await self._exit_stack.enter_async_context(self._client)
        await self._exit_stack.enter_async_context(self._runs_task_group)

        self.is_setup = True

    async def teardown(self, *exc_info):
        """Cleans up resources after the worker is stopped."""
        self._logger.debug("Tearing down worker...")
        self.is_setup = False
        for scope in self._scheduled_task_scopes:
            scope.cancel()

        await self._exit_stack.__aexit__(*exc_info)
        if self._started_event:
            await self._emit_worker_stopped_event(self._started_event)
        self._runs_task_group = None
        self._client = None

    def is_worker_still_polling(self, query_interval_seconds: float) -> bool:
        """
        This method is invoked by a webserver healthcheck handler
        and returns a boolean indicating if the worker has recorded a
        scheduled flow run poll within a variable amount of time.

        The `query_interval_seconds` is the same value that is used by
        the loop services - we will evaluate if the _last_polled_time
        was within that interval x 30 (so 10s -> 5m)

        The instance property `self._last_polled_time`
        is currently set/updated in `get_and_submit_flow_runs()`
        """
        threshold_seconds = query_interval_seconds * 30

        seconds_since_last_poll = (
            pendulum.now("utc") - self._last_polled_time
        ).in_seconds()

        is_still_polling = seconds_since_last_poll <= threshold_seconds

        if not is_still_polling:
            self._logger.error(
                f"Worker has not polled in the last {seconds_since_last_poll} seconds "
                "and should be restarted"
            )

        return is_still_polling

    async def get_and_submit_flow_runs(self):
        runs_response = await self._get_scheduled_flow_runs()

        self._last_polled_time = pendulum.now("utc")

        return await self._submit_scheduled_flow_runs(flow_run_response=runs_response)

    async def _update_local_work_pool_info(self):
        try:
            work_pool = await self._client.read_work_pool(
                work_pool_name=self._work_pool_name
            )

        except ObjectNotFound:
            if self._create_pool_if_not_found:
                wp = WorkPoolCreate(
                    name=self._work_pool_name,
                    type=self.type,
                )
                if self._base_job_template is not None:
                    wp.base_job_template = self._base_job_template

                work_pool = await self._client.create_work_pool(work_pool=wp)
                self._logger.info(f"Work pool {self._work_pool_name!r} created.")
            else:
                self._logger.warning(f"Work pool {self._work_pool_name!r} not found!")
                if self._base_job_template is not None:
                    self._logger.warning(
                        "Ignoring supplied base job template because the work pool"
                        " already exists"
                    )
                return

        # if the remote config type changes (or if it's being loaded for the
        # first time), check if it matches the local type and warn if not
        if getattr(self._work_pool, "type", 0) != work_pool.type:
            if work_pool.type != self.__class__.type:
                self._logger.warning(
                    "Worker type mismatch! This worker process expects type "
                    f"{self.type!r} but received {work_pool.type!r}"
                    " from the server. Unexpected behavior may occur."
                )

        # once the work pool is loaded, verify that it has a `base_job_template` and
        # set it if not
        if not work_pool.base_job_template:
            job_template = self.__class__.get_default_base_job_template()
            await self._set_work_pool_template(work_pool, job_template)
            work_pool.base_job_template = job_template

        self._work_pool = work_pool

    async def _send_worker_heartbeat(
        self, get_worker_id: bool = False
    ) -> Optional[UUID]:
        """
        Sends a heartbeat to the API.

        If `get_worker_id` is True, the worker ID will be retrieved from the API.
        """
        if self._work_pool:
            return await self._client.send_worker_heartbeat(
                work_pool_name=self._work_pool_name,
                worker_name=self.name,
                heartbeat_interval_seconds=self.heartbeat_interval_seconds,
                get_worker_id=get_worker_id,
            )

    async def sync_with_backend(self):
        """
        Updates the worker's local information about it's current work pool and
        queues. Sends a worker heartbeat to the API.
        """
        await self._update_local_work_pool_info()
        # Only do this logic if we've enabled the experiment, are connected to cloud and we don't have an ID.
        if (
            get_current_settings().experiments.worker_logging_to_api_enabled
            and (
                self._client.server_type == ServerType.CLOUD
                or get_current_settings().testing.test_mode
            )
            and self.backend_id is None
        ):
            get_worker_id = True
        else:
            get_worker_id = False

        remote_id = await self._send_worker_heartbeat(get_worker_id=get_worker_id)

        if get_worker_id and remote_id is None:
            self._logger.warning(
                "Failed to retrieve worker ID from the Prefect API server."
            )
        else:
            self.backend_id = remote_id

        self._logger.debug(
            f"Worker synchronized with the Prefect API server. Remote ID: {self.backend_id}"
        )

    async def _get_scheduled_flow_runs(
        self,
    ) -> List["WorkerFlowRunResponse"]:
        """
        Retrieve scheduled flow runs from the work pool's queues.
        """
        scheduled_before = pendulum.now("utc").add(seconds=int(self._prefetch_seconds))
        self._logger.debug(
            f"Querying for flow runs scheduled before {scheduled_before}"
        )
        try:
            scheduled_flow_runs = (
                await self._client.get_scheduled_flow_runs_for_work_pool(
                    work_pool_name=self._work_pool_name,
                    scheduled_before=scheduled_before,
                    work_queue_names=list(self._work_queues),
                )
            )
            self._logger.debug(
                f"Discovered {len(scheduled_flow_runs)} scheduled_flow_runs"
            )
            return scheduled_flow_runs
        except ObjectNotFound:
            # the pool doesn't exist; it will be created on the next
            # heartbeat (or an appropriate warning will be logged)
            return []

    async def _submit_scheduled_flow_runs(
        self, flow_run_response: List["WorkerFlowRunResponse"]
    ) -> List["FlowRun"]:
        """
        Takes a list of WorkerFlowRunResponses and submits the referenced flow runs
        for execution by the worker.
        """
        submittable_flow_runs = [entry.flow_run for entry in flow_run_response]

        for flow_run in submittable_flow_runs:
            if flow_run.id in self._submitting_flow_run_ids:
                continue
            try:
                if self._limiter:
                    self._limiter.acquire_on_behalf_of_nowait(flow_run.id)
            except anyio.WouldBlock:
                self._logger.info(
                    f"Flow run limit reached; {self._limiter.borrowed_tokens} flow runs"
                    " in progress."
                )
                break
            else:
                run_logger = self.get_flow_run_logger(flow_run)
                run_logger.info(
                    f"Worker '{self.name}' submitting flow run '{flow_run.id}'"
                )
                self._submitting_flow_run_ids.add(flow_run.id)
                self._runs_task_group.start_soon(
                    self._submit_run,
                    flow_run,
                )

        return list(
            filter(
                lambda run: run.id in self._submitting_flow_run_ids,
                submittable_flow_runs,
            )
        )

    async def _check_flow_run(self, flow_run: "FlowRun") -> None:
        """
        Performs a check on a submitted flow run to warn the user if the flow run
        was created from a deployment with a storage block.
        """
        if flow_run.deployment_id:
            assert (
                self._client and self._client._started
            ), "Client must be started to check flow run deployment."
            deployment = await self._client.read_deployment(flow_run.deployment_id)
            if deployment.storage_document_id:
                raise ValueError(
                    f"Flow run {flow_run.id!r} was created from deployment"
                    f" {deployment.name!r} which is configured with a storage block."
                    " Please use an agent to execute this flow run."
                )

    async def _submit_run(self, flow_run: "FlowRun") -> None:
        """
        Submits a given flow run for execution by the worker.
        """
        run_logger = self.get_flow_run_logger(flow_run)

        try:
            await self._check_flow_run(flow_run)
        except (ValueError, ObjectNotFound):
            self._logger.exception(
                (
                    "Flow run %s did not pass checks and will not be submitted for"
                    " execution"
                ),
                flow_run.id,
            )
            self._submitting_flow_run_ids.remove(flow_run.id)
            return

        ready_to_submit = await self._propose_pending_state(flow_run)

        if ready_to_submit:
            readiness_result = await self._runs_task_group.start(
                self._submit_run_and_capture_errors, flow_run
            )

            if readiness_result and not isinstance(readiness_result, Exception):
                try:
                    await self._client.update_flow_run(
                        flow_run_id=flow_run.id,
                        infrastructure_pid=str(readiness_result),
                    )
                except Exception:
                    run_logger.exception(
                        "An error occurred while setting the `infrastructure_pid` on "
                        f"flow run {flow_run.id!r}. The flow run will "
                        "not be cancellable."
                    )

                run_logger.info(f"Completed submission of flow run '{flow_run.id}'")

            else:
                # If the run is not ready to submit, release the concurrency slot
                if self._limiter:
                    self._limiter.release_on_behalf_of(flow_run.id)

            self._submitting_flow_run_ids.remove(flow_run.id)

    async def _submit_run_and_capture_errors(
        self, flow_run: "FlowRun", task_status: Optional[anyio.abc.TaskStatus] = None
    ) -> Union[BaseWorkerResult, Exception]:
        run_logger = self.get_flow_run_logger(flow_run)

        try:
            configuration = await self._get_configuration(flow_run)
            submitted_event = self._emit_flow_run_submitted_event(configuration)
            result = await self.run(
                flow_run=flow_run,
                task_status=task_status,
                configuration=configuration,
            )
        except Exception as exc:
            if not task_status._future.done():
                # This flow run was being submitted and did not start successfully
                run_logger.exception(
                    f"Failed to submit flow run '{flow_run.id}' to infrastructure."
                )
                # Mark the task as started to prevent agent crash
                task_status.started(exc)
                await self._propose_crashed_state(
                    flow_run, "Flow run could not be submitted to infrastructure"
                )
            else:
                run_logger.exception(
                    f"An error occurred while monitoring flow run '{flow_run.id}'. "
                    "The flow run will not be marked as failed, but an issue may have "
                    "occurred."
                )
            return exc
        finally:
            if self._limiter:
                self._limiter.release_on_behalf_of(flow_run.id)

        if not task_status._future.done():
            run_logger.error(
                f"Infrastructure returned without reporting flow run '{flow_run.id}' "
                "as started or raising an error. This behavior is not expected and "
                "generally indicates improper implementation of infrastructure. The "
                "flow run will not be marked as failed, but an issue may have occurred."
            )
            # Mark the task as started to prevent agent crash
            task_status.started()

        if result.status_code != 0:
            await self._propose_crashed_state(
                flow_run,
                (
                    "Flow run infrastructure exited with non-zero status code"
                    f" {result.status_code}."
                ),
            )

        self._emit_flow_run_executed_event(result, configuration, submitted_event)

        return result

    def get_status(self):
        """
        Retrieves the status of the current worker including its name, current worker
        pool, the work pool queues it is polling, and its local settings.
        """
        return {
            "name": self.name,
            "work_pool": (
                self._work_pool.model_dump(mode="json")
                if self._work_pool is not None
                else None
            ),
            "settings": {
                "prefetch_seconds": self._prefetch_seconds,
            },
        }

    async def _get_configuration(
        self,
        flow_run: "FlowRun",
        deployment: Optional["DeploymentResponse"] = None,
    ) -> BaseJobConfiguration:
        deployment = (
            deployment
            if deployment
            else await self._client.read_deployment(flow_run.deployment_id)
        )
        flow = await self._client.read_flow(flow_run.flow_id)

        deployment_vars = deployment.job_variables or {}
        flow_run_vars = flow_run.job_variables or {}
        job_variables = {**deployment_vars}

        # merge environment variables carefully, otherwise full override
        if isinstance(job_variables.get("env"), dict):
            job_variables["env"].update(flow_run_vars.pop("env", {}))
        job_variables.update(flow_run_vars)

        configuration = await self.job_configuration.from_template_and_values(
            base_job_template=self._work_pool.base_job_template,
            values=job_variables,
            client=self._client,
        )
        configuration.prepare_for_flow_run(
            flow_run=flow_run, deployment=deployment, flow=flow
        )
        return configuration

    async def _propose_pending_state(self, flow_run: "FlowRun") -> bool:
        run_logger = self.get_flow_run_logger(flow_run)
        state = flow_run.state
        try:
            state = await propose_state(
                self._client, Pending(), flow_run_id=flow_run.id
            )
        except Abort as exc:
            run_logger.info(
                (
                    f"Aborted submission of flow run '{flow_run.id}'. "
                    f"Server sent an abort signal: {exc}"
                ),
            )
            return False
        except Exception:
            run_logger.exception(
                f"Failed to update state of flow run '{flow_run.id}'",
            )
            return False

        if not state.is_pending():
            run_logger.info(
                (
                    f"Aborted submission of flow run '{flow_run.id}': "
                    f"Server returned a non-pending state {state.type.value!r}"
                ),
            )
            return False

        return True

    async def _propose_failed_state(self, flow_run: "FlowRun", exc: Exception) -> None:
        run_logger = self.get_flow_run_logger(flow_run)
        try:
            await propose_state(
                self._client,
                await exception_to_failed_state(message="Submission failed.", exc=exc),
                flow_run_id=flow_run.id,
            )
        except Abort:
            # We've already failed, no need to note the abort but we don't want it to
            # raise in the agent process
            pass
        except Exception:
            run_logger.error(
                f"Failed to update state of flow run '{flow_run.id}'",
                exc_info=True,
            )

    async def _propose_crashed_state(self, flow_run: "FlowRun", message: str) -> None:
        run_logger = self.get_flow_run_logger(flow_run)
        try:
            state = await propose_state(
                self._client,
                Crashed(message=message),
                flow_run_id=flow_run.id,
            )
        except Abort:
            # Flow run already marked as failed
            pass
        except Exception:
            run_logger.exception(f"Failed to update state of flow run '{flow_run.id}'")
        else:
            if state.is_crashed():
                run_logger.info(
                    f"Reported flow run '{flow_run.id}' as crashed: {message}"
                )

    async def _mark_flow_run_as_cancelled(
        self, flow_run: "FlowRun", state_updates: Optional[dict] = None
    ) -> None:
        state_updates = state_updates or {}
        state_updates.setdefault("name", "Cancelled")
        state_updates.setdefault("type", StateType.CANCELLED)
        state = flow_run.state.model_copy(update=state_updates)

        await self._client.set_flow_run_state(flow_run.id, state, force=True)

        # Do not remove the flow run from the cancelling set immediately because
        # the API caches responses for the `read_flow_runs` and we do not want to
        # duplicate cancellations.
        await self._schedule_task(
            60 * 10, self._cancelling_flow_run_ids.remove, flow_run.id
        )

    async def _set_work_pool_template(self, work_pool, job_template):
        """Updates the `base_job_template` for the worker's work pool server side."""
        await self._client.update_work_pool(
            work_pool_name=work_pool.name,
            work_pool=WorkPoolUpdate(
                base_job_template=job_template,
            ),
        )

    async def _schedule_task(self, __in_seconds: int, fn, *args, **kwargs):
        """
        Schedule a background task to start after some time.

        These tasks will be run immediately when the worker exits instead of waiting.

        The function may be async or sync. Async functions will be awaited.
        """

        async def wrapper(task_status):
            # If we are shutting down, do not sleep; otherwise sleep until the scheduled
            # time or shutdown
            if self.is_setup:
                with anyio.CancelScope() as scope:
                    self._scheduled_task_scopes.add(scope)
                    task_status.started()
                    await anyio.sleep(__in_seconds)

                self._scheduled_task_scopes.remove(scope)
            else:
                task_status.started()

            result = fn(*args, **kwargs)
            if asyncio.iscoroutine(result):
                await result

        await self._runs_task_group.start(wrapper)

    async def __aenter__(self):
        self._logger.debug("Entering worker context...")
        await self.setup()

        return self

    async def __aexit__(self, *exc_info):
        self._logger.debug("Exiting worker context...")
        await self.teardown(*exc_info)

    def __repr__(self):
        return f"Worker(pool={self._work_pool_name!r}, name={self.name!r})"

    def _event_resource(self):
        return {
            "prefect.resource.id": f"prefect.worker.{self.type}.{self.get_name_slug()}",
            "prefect.resource.name": self.name,
            "prefect.version": prefect.__version__,
            "prefect.worker-type": self.type,
        }

    def _event_related_resources(
        self,
        configuration: Optional[BaseJobConfiguration] = None,
        include_self: bool = False,
    ) -> List[RelatedResource]:
        related = []
        if configuration:
            related += configuration._related_resources()

        if self._work_pool:
            related.append(
                object_as_related_resource(
                    kind="work-pool", role="work-pool", object=self._work_pool
                )
            )

        if include_self:
            worker_resource = self._event_resource()
            worker_resource["prefect.resource.role"] = "worker"
            related.append(RelatedResource.model_validate(worker_resource))

        return related

    def _emit_flow_run_submitted_event(
        self, configuration: BaseJobConfiguration
    ) -> Event:
        return emit_event(
            event="prefect.worker.submitted-flow-run",
            resource=self._event_resource(),
            related=self._event_related_resources(configuration=configuration),
        )

    def _emit_flow_run_executed_event(
        self,
        result: BaseWorkerResult,
        configuration: BaseJobConfiguration,
        submitted_event: Event,
    ):
        related = self._event_related_resources(configuration=configuration)

        for resource in related:
            if resource.role == "flow-run":
                resource["prefect.infrastructure.identifier"] = str(result.identifier)
                resource["prefect.infrastructure.status-code"] = str(result.status_code)

        emit_event(
            event="prefect.worker.executed-flow-run",
            resource=self._event_resource(),
            related=related,
            follows=submitted_event,
        )

    async def _emit_worker_started_event(self) -> Event:
        return emit_event(
            "prefect.worker.started",
            resource=self._event_resource(),
            related=self._event_related_resources(),
        )

    async def _emit_worker_stopped_event(self, started_event: Event):
        emit_event(
            "prefect.worker.stopped",
            resource=self._event_resource(),
            related=self._event_related_resources(),
            follows=started_event,
        )

__init__(work_pool_name, work_queues=None, name=None, prefetch_seconds=None, create_pool_if_not_found=True, limit=None, heartbeat_interval_seconds=None, *, base_job_template=None)

Base class for all Prefect workers.

Parameters:

Name	Type	Description	Default
name	Optional[str]	The name of the worker. If not provided, a random one will be generated. If provided, it cannot contain '/' or '%'. The name is used to identify the worker in the UI; if two processes have the same name, they will be treated as the same worker.	None
work_pool_name	str	The name of the work pool to poll.	required
work_queues	Optional[List[str]]	A list of work queues to poll. If not provided, all work queue in the work pool will be polled.	None
prefetch_seconds	Optional[float]	The number of seconds to prefetch flow runs for.	None
create_pool_if_not_found	bool	Whether to create the work pool if it is not found. Defaults to True, but can be set to False to ensure that work pools are not created accidentally.	True
limit	Optional[int]	The maximum number of flow runs this worker should be running at a given time.	None
heartbeat_interval_seconds	Optional[int]	The number of seconds between worker heartbeats.	None
base_job_template	Optional[Dict[str, Any]]	If creating the work pool, provide the base job template to use. Logs a warning if the pool already exists.	None

Source code in src/prefect/workers/base.py

def __init__(
    self,
    work_pool_name: str,
    work_queues: Optional[List[str]] = None,
    name: Optional[str] = None,
    prefetch_seconds: Optional[float] = None,
    create_pool_if_not_found: bool = True,
    limit: Optional[int] = None,
    heartbeat_interval_seconds: Optional[int] = None,
    *,
    base_job_template: Optional[Dict[str, Any]] = None,
):
    """
    Base class for all Prefect workers.

    Args:
        name: The name of the worker. If not provided, a random one
            will be generated. If provided, it cannot contain '/' or '%'.
            The name is used to identify the worker in the UI; if two
            processes have the same name, they will be treated as the same
            worker.
        work_pool_name: The name of the work pool to poll.
        work_queues: A list of work queues to poll. If not provided, all
            work queue in the work pool will be polled.
        prefetch_seconds: The number of seconds to prefetch flow runs for.
        create_pool_if_not_found: Whether to create the work pool
            if it is not found. Defaults to `True`, but can be set to `False` to
            ensure that work pools are not created accidentally.
        limit: The maximum number of flow runs this worker should be running at
            a given time.
        heartbeat_interval_seconds: The number of seconds between worker heartbeats.
        base_job_template: If creating the work pool, provide the base job
            template to use. Logs a warning if the pool already exists.
    """
    if name and ("/" in name or "%" in name):
        raise ValueError("Worker name cannot contain '/' or '%'")
    self.name = name or f"{self.__class__.__name__} {uuid4()}"
    self._started_event: Optional[Event] = None
    self._logger = get_logger(f"worker.{self.__class__.type}.{self.name.lower()}")

    self.is_setup = False
    self._create_pool_if_not_found = create_pool_if_not_found
    self._base_job_template = base_job_template
    self._work_pool_name = work_pool_name
    self._work_queues: Set[str] = set(work_queues) if work_queues else set()

    self._prefetch_seconds: float = (
        prefetch_seconds or PREFECT_WORKER_PREFETCH_SECONDS.value()
    )
    self.heartbeat_interval_seconds = (
        heartbeat_interval_seconds or PREFECT_WORKER_HEARTBEAT_SECONDS.value()
    )

    self.backend_id: Optional[UUID] = None
    self._work_pool: Optional[WorkPool] = None
    self._exit_stack: AsyncExitStack = AsyncExitStack()
    self._runs_task_group: Optional[anyio.abc.TaskGroup] = None
    self._client: Optional[PrefectClient] = None
    self._last_polled_time: pendulum.DateTime = pendulum.now("utc")
    self._limit = limit
    self._limiter: Optional[anyio.CapacityLimiter] = None
    self._submitting_flow_run_ids = set()
    self._cancelling_flow_run_ids = set()
    self._scheduled_task_scopes = set()

get_all_available_worker_types() staticmethod

Returns all worker types available in the local registry.

Source code in src/prefect/workers/base.py

@staticmethod
def get_all_available_worker_types() -> List[str]:
    """
    Returns all worker types available in the local registry.
    """
    load_prefect_collections()
    worker_registry = get_registry_for_type(BaseWorker)
    if worker_registry is not None:
        return list(worker_registry.keys())
    return []

get_status()

Retrieves the status of the current worker including its name, current worker pool, the work pool queues it is polling, and its local settings.

Source code in src/prefect/workers/base.py

def get_status(self):
    """
    Retrieves the status of the current worker including its name, current worker
    pool, the work pool queues it is polling, and its local settings.
    """
    return {
        "name": self.name,
        "work_pool": (
            self._work_pool.model_dump(mode="json")
            if self._work_pool is not None
            else None
        ),
        "settings": {
            "prefetch_seconds": self._prefetch_seconds,
        },
    }

get_worker_class_from_type(type) staticmethod

Returns the worker class for a given worker type. If the worker type is not recognized, returns None.

Source code in src/prefect/workers/base.py

@staticmethod
def get_worker_class_from_type(type: str) -> Optional[Type["BaseWorker"]]:
    """
    Returns the worker class for a given worker type. If the worker type
    is not recognized, returns None.
    """
    load_prefect_collections()
    worker_registry = get_registry_for_type(BaseWorker)
    if worker_registry is not None:
        return worker_registry.get(type)

is_worker_still_polling(query_interval_seconds)

This method is invoked by a webserver healthcheck handler and returns a boolean indicating if the worker has recorded a scheduled flow run poll within a variable amount of time.

The query_interval_seconds is the same value that is used by the loop services - we will evaluate if the _last_polled_time was within that interval x 30 (so 10s -> 5m)

The instance property self._last_polled_time is currently set/updated in get_and_submit_flow_runs()

Source code in src/prefect/workers/base.py

def is_worker_still_polling(self, query_interval_seconds: float) -> bool:
    """
    This method is invoked by a webserver healthcheck handler
    and returns a boolean indicating if the worker has recorded a
    scheduled flow run poll within a variable amount of time.

    The `query_interval_seconds` is the same value that is used by
    the loop services - we will evaluate if the _last_polled_time
    was within that interval x 30 (so 10s -> 5m)

    The instance property `self._last_polled_time`
    is currently set/updated in `get_and_submit_flow_runs()`
    """
    threshold_seconds = query_interval_seconds * 30

    seconds_since_last_poll = (
        pendulum.now("utc") - self._last_polled_time
    ).in_seconds()

    is_still_polling = seconds_since_last_poll <= threshold_seconds

    if not is_still_polling:
        self._logger.error(
            f"Worker has not polled in the last {seconds_since_last_poll} seconds "
            "and should be restarted"
        )

    return is_still_polling

run(flow_run, configuration, task_status=None) abstractmethod async

Runs a given flow run on the current worker.

Source code in src/prefect/workers/base.py

@abc.abstractmethod
async def run(
    self,
    flow_run: "FlowRun",
    configuration: BaseJobConfiguration,
    task_status: Optional[anyio.abc.TaskStatus] = None,
) -> BaseWorkerResult:
    """
    Runs a given flow run on the current worker.
    """
    raise NotImplementedError(
        "Workers must implement a method for running submitted flow runs"
    )

setup() async

Prepares the worker to run.

Source code in src/prefect/workers/base.py

async def setup(self):
    """Prepares the worker to run."""
    self._logger.debug("Setting up worker...")
    self._runs_task_group = anyio.create_task_group()
    self._limiter = (
        anyio.CapacityLimiter(self._limit) if self._limit is not None else None
    )

    if not PREFECT_TEST_MODE and not PREFECT_API_URL.value():
        raise ValueError("`PREFECT_API_URL` must be set to start a Worker.")

    self._client = get_client()
    await self._exit_stack.enter_async_context(self._client)
    await self._exit_stack.enter_async_context(self._runs_task_group)

    self.is_setup = True

start(run_once=False, with_healthcheck=False, printer=print) async

Starts the worker and runs the main worker loops.

By default, the worker will run loops to poll for scheduled/cancelled flow runs and sync with the Prefect API server.

If run_once is set, the worker will only run each loop once and then return.

If with_healthcheck is set, the worker will start a healthcheck server which can be used to determine if the worker is still polling for flow runs and restart the worker if necessary.

Parameters:

Name	Type	Description	Default
run_once	bool	If set, the worker will only run each loop once then return.	False
with_healthcheck	bool	If set, the worker will start a healthcheck server.	False
printer	Callable[..., None]	A print-like function where logs will be reported.	print

Source code in src/prefect/workers/base.py

async def start(
    self,
    run_once: bool = False,
    with_healthcheck: bool = False,
    printer: Callable[..., None] = print,
):
    """
    Starts the worker and runs the main worker loops.

    By default, the worker will run loops to poll for scheduled/cancelled flow
    runs and sync with the Prefect API server.

    If `run_once` is set, the worker will only run each loop once and then return.

    If `with_healthcheck` is set, the worker will start a healthcheck server which
    can be used to determine if the worker is still polling for flow runs and restart
    the worker if necessary.

    Args:
        run_once: If set, the worker will only run each loop once then return.
        with_healthcheck: If set, the worker will start a healthcheck server.
        printer: A `print`-like function where logs will be reported.
    """
    healthcheck_server = None
    healthcheck_thread = None
    try:
        async with self as worker:
            # wait for an initial heartbeat to configure the worker
            await worker.sync_with_backend()
            # schedule the scheduled flow run polling loop
            async with anyio.create_task_group() as loops_task_group:
                loops_task_group.start_soon(
                    partial(
                        critical_service_loop,
                        workload=self.get_and_submit_flow_runs,
                        interval=PREFECT_WORKER_QUERY_SECONDS.value(),
                        run_once=run_once,
                        jitter_range=0.3,
                        backoff=4,  # Up to ~1 minute interval during backoff
                    )
                )
                # schedule the sync loop
                loops_task_group.start_soon(
                    partial(
                        critical_service_loop,
                        workload=self.sync_with_backend,
                        interval=self.heartbeat_interval_seconds,
                        run_once=run_once,
                        jitter_range=0.3,
                        backoff=4,
                    )
                )

                self._started_event = await self._emit_worker_started_event()

                if with_healthcheck:
                    from prefect.workers.server import build_healthcheck_server

                    # we'll start the ASGI server in a separate thread so that
                    # uvicorn does not block the main thread
                    healthcheck_server = build_healthcheck_server(
                        worker=worker,
                        query_interval_seconds=PREFECT_WORKER_QUERY_SECONDS.value(),
                    )
                    healthcheck_thread = threading.Thread(
                        name="healthcheck-server-thread",
                        target=healthcheck_server.run,
                        daemon=True,
                    )
                    healthcheck_thread.start()
                printer(f"Worker {worker.name!r} started!")
    finally:
        if healthcheck_server and healthcheck_thread:
            self._logger.debug("Stopping healthcheck server...")
            healthcheck_server.should_exit = True
            healthcheck_thread.join()
            self._logger.debug("Healthcheck server stopped.")

    printer(f"Worker {worker.name!r} stopped!")

sync_with_backend() async

Updates the worker's local information about it's current work pool and queues. Sends a worker heartbeat to the API.

Source code in src/prefect/workers/base.py

async def sync_with_backend(self):
    """
    Updates the worker's local information about it's current work pool and
    queues. Sends a worker heartbeat to the API.
    """
    await self._update_local_work_pool_info()
    # Only do this logic if we've enabled the experiment, are connected to cloud and we don't have an ID.
    if (
        get_current_settings().experiments.worker_logging_to_api_enabled
        and (
            self._client.server_type == ServerType.CLOUD
            or get_current_settings().testing.test_mode
        )
        and self.backend_id is None
    ):
        get_worker_id = True
    else:
        get_worker_id = False

    remote_id = await self._send_worker_heartbeat(get_worker_id=get_worker_id)

    if get_worker_id and remote_id is None:
        self._logger.warning(
            "Failed to retrieve worker ID from the Prefect API server."
        )
    else:
        self.backend_id = remote_id

    self._logger.debug(
        f"Worker synchronized with the Prefect API server. Remote ID: {self.backend_id}"
    )

teardown(*exc_info) async

Cleans up resources after the worker is stopped.

Source code in src/prefect/workers/base.py

async def teardown(self, *exc_info):
    """Cleans up resources after the worker is stopped."""
    self._logger.debug("Tearing down worker...")
    self.is_setup = False
    for scope in self._scheduled_task_scopes:
        scope.cancel()

    await self._exit_stack.__aexit__(*exc_info)
    if self._started_event:
        await self._emit_worker_stopped_event(self._started_event)
    self._runs_task_group = None
    self._client = None