Repository: PrefectHQ/prefect
File: src/integrations/prefect-kubernetes/prefect_kubernetes/worker.py (Python)

Code before:
        """
        all_annotations = {
            **self.job_manifest["metadata"].get("annotations", {}),
            **self.annotations,
        }
        self.job_manifest["metadata"]["annotations"] = {
            _slugify_label_key(k): v for k, v in all_annotations.items()
        }

    def _propagate_annotations_to_pod(self):
        """Propagates annotations to the pod in the job manifest.

        Unlike labels, there are no automatic base annotations, so we only
        touch pod metadata when the user actually provided annotations.
        """
        merged_annotations = self.job_manifest["metadata"].get("annotations", {})
        if not merged_annotations:
            return

        current_pod_metadata = self.job_manifest["spec"]["template"].get("metadata", {})
        current_pod_annotations = current_pod_metadata.get("annotations", {})
        all_annotations = {**current_pod_annotations, **merged_annotations}

        current_pod_metadata["annotations"] = {
            _slugify_label_key(k): v for k, v in all_annotations.items()
        }
        self.job_manifest["spec"]["template"]["metadata"] = current_pod_metadata


class KubernetesWorkerVariables(BaseVariables):
    """
    Default variables for the Kubernetes worker.

    The schema for this class is used to populate the `variables` section of the default
    base job template.
    """

    annotations: Dict[str, str] = Field(
        default_factory=dict,
        description="Annotations applied to Kubernetes jobs and pods created by the worker.",
    )
    namespace: str = Field(
        default="default", description="The Kubernetes namespace to create jobs within."
    )
    image: Optional[str] = Field(
        default=None,
        description="The image reference of a container image to use for created jobs. "
        "If not set, the latest Prefect image will be used.",
        examples=["docker.io/prefecthq/prefect:3-latest"],
    )
    service_account_name: Optional[str] = Field(

Review comment (suggestion):
Slugifying the annotations on propagation to the pod will result in the base job template annotations being double-slugified. I think we can drop this slugification since the annotations were already slugified before applying them to the job.
```suggestion
        current_pod_metadata["annotations"] = all_annotations
```

Code after:
        """
        all_annotations = {
            **self.job_manifest["metadata"].get("annotations", {}),
            **self.annotations,
        }
        self.job_manifest["metadata"]["annotations"] = {
            _slugify_label_key(k): v for k, v in all_annotations.items()
        }

    def _propagate_annotations_to_pod(self):
        """Propagates annotations to the pod in the job manifest.

        Unlike labels, there are no automatic base annotations, so we only
        touch pod metadata when the user actually provided annotations.
        """
        merged_annotations = self.job_manifest["metadata"].get("annotations", {})
        if not merged_annotations:
            return

        current_pod_metadata = self.job_manifest["spec"]["template"].get("metadata", {})
        current_pod_annotations = current_pod_metadata.get("annotations", {})
        all_annotations = {**current_pod_annotations, **merged_annotations}

        current_pod_metadata["annotations"] = all_annotations
        self.job_manifest["spec"]["template"]["metadata"] = current_pod_metadata


class KubernetesWorkerVariables(BaseVariables):
    """
    Default variables for the Kubernetes worker.

    The schema for this class is used to populate the `variables` section of the default
    base job template.
    """

    annotations: Dict[str, str] = Field(
        default_factory=dict,
        description="Annotations applied to Kubernetes jobs and pods created by the worker.",
    )
    namespace: str = Field(
        default="default", description="The Kubernetes namespace to create jobs within."
    )
    image: Optional[str] = Field(
        default=None,
        description="The image reference of a container image to use for created jobs. "
        "If not set, the latest Prefect image will be used.",
        examples=["docker.io/prefecthq/prefect:3-latest"],
    )
    service_account_name: Optional[str] = Field(
        default=None,
        description="The Kubernetes service account to use for job creation.",