content: #  Copyright (c) ZenML GmbH 2025. All Rights Reserved.
  #
  #  Licensed under the Apache License, Version 2.0 (the "License");
  #  you may not use this file except in compliance with the License.
  #  You may obtain a copy of the License at:
  #
  #       https://www.apache.org/licenses/LICENSE-2.0
  #
  #  Unless required by applicable law or agreed to in writing, software
  #  distributed under the License is distributed on an "AS IS" BASIS,
  #  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
  #  or implied. See the License for the specific language governing
  #  permissions and limitations under the License.
  """Implementation of the AWS App Runner deployer."""
  
  import datetime
  import json
  import re
  from typing import (
      TYPE_CHECKING,
      Any,
      Dict,
      Generator,
      List,
      Optional,
      Tuple,
      Type,
      cast,
  )
  from uuid import UUID
  
  import boto3
  from botocore.exceptions import BotoCoreError, ClientError
  from pydantic import BaseModel
  
  from zenml.config.base_settings import BaseSettings
  from zenml.config.resource_settings import ResourceSettings
  from zenml.deployers.containerized_deployer import ContainerizedDeployer
  from zenml.deployers.exceptions import (
      DeployerError,
      DeploymentDeprovisionError,
      DeploymentLogsNotFoundError,
      DeploymentNotFoundError,
      DeploymentProvisionError,
  )
  from zenml.deployers.server.entrypoint_configuration import (
      DEPLOYMENT_ID_OPTION,
      DeploymentEntrypointConfiguration,
  )
  from zenml.enums import DeploymentStatus, StackComponentType
  from zenml.integrations.aws.flavors.aws_deployer_flavor import (
      AWSDeployerConfig,
      AWSDeployerSettings,
  )
  from zenml.logger import get_logger
  from zenml.models import (
      DeploymentOperationalState,
      DeploymentResponse,
  )
  from zenml.stack import StackValidator
  
  if TYPE_CHECKING:
      from zenml.stack import Stack
  
  logger = get_logger(__name__)
  
  # Default resource and scaling configuration constants
  DEFAULT_CPU = 0.25  # vCPU
  DEFAULT_MEMORY = 0.5  # GB
  DEFAULT_MIN_REPLICAS = 1
  DEFAULT_MAX_REPLICAS = 25
  DEFAULT_MAX_CONCURRENCY = 100
  
  # AWS App Runner built-in limits
  AWS_APP_RUNNER_MAX_SIZE = 1000
  AWS_APP_RUNNER_MAX_CONCURRENCY = 1000
  
  
  class AppRunnerDeploymentMetadata(BaseModel):
      """Metadata for an App Runner deployment."""
  
      service_name: Optional[str] = None
      service_arn: Optional[str] = None
      service_url: Optional[str] = None
      region: Optional[str] = None
      service_id: Optional[str] = None
      status: Optional[str] = None
      source_configuration: Optional[Dict[str, Any]] = None
      instance_configuration: Optional[Dict[str, Any]] = None
      auto_scaling_configuration_summary: Optional[Dict[str, Any]] = None
      auto_scaling_configuration_arn: Optional[str] = None
      health_check_configuration: Optional[Dict[str, Any]] = None
      network_configuration: Optional[Dict[str, Any]] = None
      observability_configuration: Optional[Dict[str, Any]] = None
      encryption_configuration: Optional[Dict[str, Any]] = None
      cpu: Optional[str] = None
      memory: Optional[str] = None
      port: Optional[int] = None
      auto_scaling_max_concurrency: Optional[int] = None
      auto_scaling_max_size: Optional[int] = None
      auto_scaling_min_size: Optional[int] = None
      is_publicly_accessible: Optional[bool] = None
      health_check_grace_period_seconds: Optional[int] = None
      health_check_interval_seconds: Optional[int] = None
      health_check_path: Optional[str] = None
      health_check_protocol: Optional[str] = None
      health_check_timeout_seconds: Optional[int] = None
      health_check_healthy_threshold: Optional[int] = None
      health_check_unhealthy_threshold: Optional[int] = None
      tags: Optional[Dict[str, str]] = None
      traffic_allocation: Optional[Dict[str, int]] = None
      created_at: Optional[str] = None
      updated_at: Optional[str] = None
      deleted_at: Optional[str] = None
      secret_arn: Optional[str] = None
  
      @classmethod
      def from_app_runner_service(
          cls,
          service: Dict[str, Any],
          region: str,
          secret_arn: Optional[str] = None,
      ) -> "AppRunnerDeploymentMetadata":
          """Create metadata from an App Runner service.
  
          Args:
              service: The App Runner service dictionary from describe_service.
              region: The AWS region.
              secret_arn: The AWS Secrets Manager secret ARN for the deployment.
  
          Returns:
              The metadata for the App Runner service.
          """
          instance_config = service.get("InstanceConfiguration", {})
          cpu = instance_config.get("Cpu")
          memory = instance_config.get("Memory")
  
          auto_scaling_config = service.get(
              "AutoScalingConfigurationSummary", {}
          )
          auto_scaling_configuration_arn = auto_scaling_config.get(
              "AutoScalingConfigurationArn"
          )
          auto_scaling_max_concurrency = auto_scaling_config.get(
              "MaxConcurrency"
          )
          auto_scaling_max_size = auto_scaling_config.get("MaxSize")
          auto_scaling_min_size = auto_scaling_config.get("MinSize")
  
          health_check_config = service.get("HealthCheckConfiguration", {})
          health_check_grace_period = health_check_config.get(
              "HealthCheckGracePeriodSeconds"
          )
          health_check_interval = health_check_config.get("Interval")
          health_check_path = health_check_config.get("Path")
          health_check_protocol = health_check_config.get("Protocol")
          health_check_timeout = health_check_config.get("Timeout")
          health_check_healthy_threshold = health_check_config.get(
              "HealthyThreshold"
          )
          health_check_unhealthy_threshold = health_check_config.get(
              "UnhealthyThreshold"
          )
  
          network_config = service.get("NetworkConfiguration", {})
          is_publicly_accessible = network_config.get(
              "IngressConfiguration", {}
          ).get("IsPubliclyAccessible")
  
          source_config = service.get("SourceConfiguration", {})
          image_repo = source_config.get("ImageRepository", {})
          image_config = image_repo.get("ImageConfiguration", {})
  
          port = None
          if image_config:
              port = image_config.get("Port")
  
          traffic_allocation = {}
          traffic_config = service.get("TrafficConfiguration", [])
          for traffic in traffic_config:
              if traffic.get("Type") == "LATEST":
                  traffic_allocation["LATEST"] = traffic.get("Percent", 0)
              elif traffic.get("Revision"):
                  traffic_allocation[traffic["Revision"]] = traffic.get(
                      "Percent", 0
                  )
              elif traffic.get("Tag"):
                  traffic_allocation[f"tag:{traffic['Tag']}"] = traffic.get(
                      "Percent", 0
                  )
  
          # Extract timestamps
          created_at = service.get("CreatedAt")
          updated_at = service.get("UpdatedAt")
          deleted_at = service.get("DeletedAt")
  
          return cls(
              service_name=service.get("ServiceName"),
              service_arn=service.get("ServiceArn"),
              service_url=service.get("ServiceUrl"),
              region=region,
              service_id=service.get("ServiceId"),
              status=service.get("Status"),
              source_configuration=source_config,
              instance_configuration=instance_config,
              auto_scaling_configuration_summary=auto_scaling_config,
              auto_scaling_configuration_arn=auto_scaling_configuration_arn,
              health_check_configuration=health_check_config,
              network_configuration=network_config,
              observability_configuration=service.get(
                  "ObservabilityConfiguration"
              ),
              encryption_configuration=service.get("EncryptionConfiguration"),
              cpu=cpu,
              memory=memory,
              port=port,
              auto_scaling_max_concurrency=auto_scaling_max_concurrency,
              auto_scaling_max_size=auto_scaling_max_size,
              auto_scaling_min_size=auto_scaling_min_size,
              is_publicly_accessible=is_publicly_accessible,
              health_check_grace_period_seconds=health_check_grace_period,
              health_check_interval_seconds=health_check_interval,
              health_check_path=health_check_path,
              health_check_protocol=health_check_protocol,
              health_check_timeout_seconds=health_check_timeout,
              health_check_healthy_threshold=health_check_healthy_threshold,
              health_check_unhealthy_threshold=health_check_unhealthy_threshold,
              tags=dict(service.get("Tags", {})),
              traffic_allocation=traffic_allocation
              if traffic_allocation
              else None,
              created_at=created_at.isoformat() if created_at else None,
              updated_at=updated_at.isoformat() if updated_at else None,
              deleted_at=deleted_at.isoformat() if deleted_at else None,
              secret_arn=secret_arn,
          )
  
      @classmethod
      def from_deployment(
          cls, deployment: DeploymentResponse
      ) -> "AppRunnerDeploymentMetadata":
          """Create metadata from a deployment.
  
          Args:
              deployment: The deployment to get the metadata for.
  
          Returns:
              The metadata for the deployment.
          """
          return cls.model_validate(deployment.deployment_metadata)
  
  
  class AWSDeployer(ContainerizedDeployer):
      """Deployer responsible for deploying pipelines on AWS App Runner."""
  
      _boto_session: Optional[boto3.Session] = None
      _region: Optional[str] = None
      _app_runner_client: Optional[Any] = None
      _secrets_manager_client: Optional[Any] = None
      _logs_client: Optional[Any] = None
  
      @property
      def config(self) -> AWSDeployerConfig:
          """Returns the `AWSDeployerConfig` config.
  
          Returns:
              The configuration.
          """
          return cast(AWSDeployerConfig, self._config)
  
      @property
      def settings_class(self) -> Optional[Type["BaseSettings"]]:
          """Settings class for the AWS deployer.
  
          Returns:
              The settings class.
          """
          return AWSDeployerSettings
  
      @property
      def validator(self) -> Optional[StackValidator]:
          """Ensures there is an image builder in the stack.
  
          Returns:
              A `StackValidator` instance.
          """
          return StackValidator(
              required_components={
                  StackComponentType.IMAGE_BUILDER,
                  StackComponentType.CONTAINER_REGISTRY,
              }
          )
  
      def _get_boto_session_and_region(self) -> Tuple[boto3.Session, str]:
          """Get an authenticated boto3 session and determine the region.
  
          Returns:
              A tuple containing the boto3 session and the AWS region.
  
          Raises:
              RuntimeError: If the service connector returns an unexpected type.
          """
          if (
              self._boto_session is not None
              and self._region is not None
              and not self.connector_has_expired()
          ):
              return self._boto_session, self._region
  
          # Option 1: Service connector
          if connector := self.get_connector():
              boto_session = connector.connect()
              if not isinstance(boto_session, boto3.Session):
                  raise RuntimeError(
                      f"Expected to receive a `boto3.Session` object from the "
                      f"linked connector, but got type `{type(boto_session)}`."
                  )
  
              region = boto_session.region_name
              if not region:
                  # Fallback to config region or default
                  region = self.config.region or "us-east-1"
                  logger.warning(
                      f"No region found in boto3 session, using {region}"
                  )
  
          # Option 2: Implicit configuration
          else:
              boto_session = boto3.Session(region_name=self.config.region)
  
          self._boto_session = boto_session
          self._region = region
          return boto_session, region
  
      @property
      def app_runner_client(self) -> Any:
          """Get the App Runner client.
  
          Returns:
              The App Runner client.
          """
          if self._app_runner_client is None or self.connector_has_expired():
              session, region = self._get_boto_session_and_region()
              self._app_runner_client = session.client(
                  "apprunner", region_name=region
              )
          return self._app_runner_client
  
      @property
      def secrets_manager_client(self) -> Any:
          """Get the Secrets Manager client.
  
          Returns:
              The Secrets Manager client.
          """
          if (
              self._secrets_manager_client is None
              or self.connector_has_expired()
          ):
              session, region = self._get_boto_session_and_region()
              self._secrets_manager_client = session.client(
                  "secretsmanager", region_name=region
              )
          return self._secrets_manager_client
  
      @property
      def logs_client(self) -> Any:
          """Get the CloudWatch Logs client.
  
          Returns:
              The CloudWatch Logs client.
          """
          if self._logs_client is None or self.connector_has_expired():
              session, region = self._get_boto_session_and_region()
              self._logs_client = session.client("logs", region_name=region)
          return self._logs_client
  
      @property
      def region(self) -> str:
          """Get the AWS region.
  
          Returns:
              The AWS region.
          """
          _, region = self._get_boto_session_and_region()
          return region
  
      def get_tags(
          self,
          deployment: DeploymentResponse,
          settings: AWSDeployerSettings,
      ) -> List[Dict[str, str]]:
          """Get the tags for a deployment to be used for AWS resources.
  
          Args:
              deployment: The deployment.
              settings: The deployer settings.
  
          Returns:
              The tags for the deployment.
          """
          tags = {
              **settings.tags,
              "zenml-deployment-id": str(deployment.id),
              "zenml-deployment-name": deployment.name,
              "zenml-deployer-name": str(self.name),
              "zenml-deployer-id": str(self.id),
              "managed-by": "zenml",
          }
  
          return [{"Key": k, "Value": v} for k, v in tags.items()]
  
      def _sanitize_name(
          self,
          name: str,
          random_suffix: str,
          max_length: int = 32,
          extra_allowed_characters: str = "-_",
      ) -> str:
          """Sanitize a name to comply with AWS naming requirements.
  
          Common AWS naming requirements:
          - Length: 4-max_length characters
          - Characters: letters (a-z, A-Z), numbers (0-9), configured extra
          allowed characters (e.g. dashes and underscores)
          - Must start and end with a letter or number
          - Cannot contain consecutive extra_allowed_characters
  
          Args:
              name: The raw name to sanitize.
              random_suffix: A random suffix to add to the name to ensure
                  uniqueness.
              max_length: The maximum length of the name.
              extra_allowed_characters: Extra allowed characters in the name.
  
          Returns:
              A sanitized name that complies with AWS requirements.
  
          Raises:
              RuntimeError: If the random suffix is invalid.
              ValueError: If the service name is invalid.
          """
          if (
              not re.match(r"^[a-zA-Z0-9]+$", random_suffix)
              or len(random_suffix) < 4
          ):
              raise RuntimeError(
                  f"Invalid random suffix: {random_suffix}. Must contain only "
                  "letters and numbers and be at least 4 characters long."
              )
  
          # Use the first extra allowed character as the separator
          separator = extra_allowed_characters[0]
  
          # Replace all disallowed characters with the separator
          sanitized = re.sub(
              rf"[^a-zA-Z0-9{extra_allowed_characters}]",
              separator,
              name,
          )
  
          # Remove consecutive extra allowed characters
          for char in extra_allowed_characters:
              sanitized = re.sub(
                  rf"[{char}]+",
                  char,
                  sanitized,
              )
  
          # Remove leading and trailing extra allowed characters before truncating
          sanitized = re.sub(
              r"^[^a-zA-Z0-9]+|[^a-zA-Z0-9]+$",
              "",
              sanitized,
          )
  
          # Truncate to fit within max_length character limit including suffix
          max_base_length = (
              max_length - len(random_suffix) - 1
          )  # -1 for the separator
          if len(sanitized) > max_base_length:
              sanitized = sanitized[:max_base_length]
  
          # Ensure it starts and ends with alphanumeric characters
          sanitized = re.sub(
              r"^[^a-zA-Z0-9]+|[^a-zA-Z0-9]+$",
              "",
              sanitized,
          )
  
          # Ensure we have at least one character after cleanup
          if not sanitized:
              raise ValueError(
                  f"Invalid name: {name}. Must contain at least one "
                  "alphanumeric character."
              )
  
          return f"{sanitized}{separator}{random_suffix}"
  
      def _get_service_name(
          self, deployment_name: str, deployment_id: UUID, prefix: str
      ) -> str:
          """Get the App Runner service name for a deployment.
  
          Args:
              deployment_name: The deployment name.
              deployment_id: The deployment ID.
              prefix: The prefix to use for the service name.
  
          Returns:
              The App Runner service name that complies with all naming
              requirements.
          """
          # We use the first 8 characters of the deployment UUID as a random
          # suffix to ensure uniqueness.
          deployment_id_short = str(deployment_id)[:8]
          raw_name = f"{prefix}{deployment_name}"
  
          return self._sanitize_name(
              raw_name,
              random_suffix=deployment_id_short,
              max_length=40,
              extra_allowed_characters="-_",
          )
  
      def _get_secret_name(
          self,
          deployment_name: str,
          deployment_id: UUID,
          prefix: str,
      ) -> str:
          """Get the Secrets Manager secret name for a deployment.
  
          Args:
              deployment_name: The deployment name.
              deployment_id: The deployment ID.
              prefix: The prefix to use for the secret name.
  
          Returns:
              The Secrets Manager secret name.
          """
          # We use the first 8 characters of the deployment UUID as a random
          # suffix to ensure uniqueness.
          deployment_id_short = str(deployment_id)[:8]
          raw_name = f"{prefix}{deployment_name}"
  
          return self._sanitize_name(
              raw_name,
              random_suffix=deployment_id_short,
              max_length=512,
              extra_allowed_characters="-_./",
          )
  
      def _create_or_update_secret(
          self,
          secret_name: str,
          secret_value: str,
          deployment: DeploymentResponse,
          settings: AWSDeployerSettings,
      ) -> str:
          """Create or update a secret in Secrets Manager.
  
          Args:
              secret_name: The name of the secret.
              secret_value: The value to store.
              deployment: The deployment.
              settings: The deployer settings.
  
          Returns:
              The secret ARN.
  
          Raises:
              ClientError: If the secret cannot be updated.
              DeployerError: If secret creation/update fails.
          """
          try:
              try:
                  response = self.secrets_manager_client.update_secret(
                      SecretId=secret_name,
                      SecretString=secret_value,
                  )
                  self.secrets_manager_client.tag_resource(
                      SecretId=response["ARN"],
                      Tags=self.get_tags(deployment, settings),
                  )
                  logger.debug(f"Updated existing secret {secret_name}")
                  return response["ARN"]  # type: ignore[no-any-return]
              except ClientError as e:
                  if e.response["Error"]["Code"] == "ResourceNotFoundException":
                      logger.debug(f"Creating new secret {secret_name}")
                      response = self.secrets_manager_client.create_secret(
                          Name=secret_name,
                          SecretString=secret_value,
                          Description=f"ZenML deployment secret for {deployment.name}",
                          Tags=self.get_tags(deployment, settings),
                      )
                      logger.debug(f"Created new secret {secret_name}")
                      return response["ARN"]  # type: ignore[no-any-return]
                  else:
                      raise
  
          except (ClientError, BotoCoreError) as e:
              raise DeployerError(
                  f"Failed to create/update secret {secret_name}: {e}"
              )
  
      def _get_secret_arn(self, deployment: DeploymentResponse) -> Optional[str]:
          """Get the existing AWS Secrets Manager secret ARN for a deployment.
  
          Args:
              deployment: The deployment.
  
          Returns:
              The existing AWS Secrets Manager secret ARN for the deployment,
              or None if no secret exists.
          """
          metadata = AppRunnerDeploymentMetadata.from_deployment(deployment)
  
          if not metadata.secret_arn:
              return None
  
          try:
              self.secrets_manager_client.describe_secret(
                  SecretId=metadata.secret_arn
              )
              return metadata.secret_arn
          except ClientError as e:
              if e.response["Error"]["Code"] == "ResourceNotFoundException":
                  return None
              logger.exception(f"Failed to verify secret {metadata.secret_arn}")
              return None
  
      def _delete_secret(self, secret_arn: str) -> None:
          """Delete a secret from Secrets Manager.
  
          Args:
              secret_arn: The ARN of the secret to delete.
          """
          try:
              self.secrets_manager_client.delete_secret(
                  SecretId=secret_arn,
                  ForceDeleteWithoutRecovery=True,
              )
              logger.debug(f"Deleted secret {secret_arn}")
          except ClientError as e:
              if e.response["Error"]["Code"] == "ResourceNotFoundException":
                  logger.debug(
                      f"Secret {secret_arn} not found, skipping deletion"
                  )
              else:
                  logger.exception(f"Failed to delete secret {secret_arn}")
  
      def _cleanup_deployment_secrets(
          self,
          deployment: DeploymentResponse,
      ) -> None:
          """Clean up the secret associated with a deployment.
  
          Args:
              deployment: The deployment.
          """
          secret_arn = self._get_secret_arn(deployment)
  
          if secret_arn:
              self._delete_secret(secret_arn)
  
      def _get_auto_scaling_config_name(
          self, deployment_name: str, deployment_id: UUID
      ) -> str:
          """Get the auto-scaling configuration name for a deployment.
  
          Args:
              deployment_name: The deployment name.
              deployment_id: The deployment ID.
  
          Returns:
              The auto-scaling configuration name.
          """
          # We use the first 8 characters of the deployment UUID as a random
          # suffix to ensure uniqueness.
          deployment_id_short = str(deployment_id)[:8]
          raw_name = f"zenml-{deployment_name}-{deployment_id_short}"
  
          return self._sanitize_name(
              raw_name,
              random_suffix=deployment_id_short,
              max_length=32,
              extra_allowed_characters="-_",
          )
  
      def _create_or_update_auto_scaling_config(
          self,
          config_name: str,
          min_size: int,
          max_size: int,
          max_concurrency: int,
          deployment: DeploymentResponse,
          settings: AWSDeployerSettings,
      ) -> str:
          """Create or update an auto-scaling configuration for App Runner.
  
          Args:
              config_name: The name for the auto-scaling configuration.
              min_size: Minimum number of instances.
              max_size: Maximum number of instances.
              max_concurrency: Maximum concurrent requests per instance.
              deployment: The deployment.
              settings: The deployer settings.
  
          Returns:
              The ARN of the created/updated auto-scaling configuration.
  
          Raises:
              ClientError: If the auto-scaling configuration cannot be described.
              DeployerError: If auto-scaling configuration creation/update fails.
          """
          try:
              metadata = AppRunnerDeploymentMetadata.from_deployment(deployment)
              existing_arn = metadata.auto_scaling_configuration_arn
  
              if existing_arn:
                  try:
                      response = self.app_runner_client.describe_auto_scaling_configuration(
                          AutoScalingConfigurationArn=existing_arn
                      )
                      existing_config = response["AutoScalingConfiguration"]
  
                      if (
                          existing_config["MaxConcurrency"] == max_concurrency
                          and existing_config["MaxSize"] == max_size
                          and existing_config["MinSize"] == min_size
                      ):
                          logger.debug(
                              f"Auto-scaling configuration {existing_arn} is up "
                              "to date"
                          )
                          return existing_arn
  
                      logger.debug(
                          f"Auto-scaling configuration {existing_arn} is out of "
                          "date, updating it"
                      )
  
                  except ClientError as e:
                      if (
                          e.response["Error"]["Code"]
                          != "InvalidRequestException"
                      ):
                          raise
                      logger.debug(
                          f"Existing auto-scaling configuration {existing_arn} "
                          "not found, creating new one"
                      )
              else:
                  logger.debug(
                      f"Creating auto-scaling configuration {config_name}"
                  )
  
              # The create_auto_scaling_configuration call is used to both create
              # a new auto-scaling configuration and update an existing one.
              # It is possible to create multiple revisions of the same
              # configuration by calling create_auto_scaling_configuration
              # multiple times using the same AutoScalingConfigurationName.
              response = (
                  self.app_runner_client.create_auto_scaling_configuration(
                      AutoScalingConfigurationName=config_name,
                      MaxConcurrency=max_concurrency,
                      MaxSize=max_size,
                      MinSize=min_size,
                      Tags=self.get_tags(deployment, settings),
                  )
              )
  
              return response["AutoScalingConfiguration"][  # type: ignore[no-any-return]
                  "AutoScalingConfigurationArn"
              ]
  
          except (ClientError, BotoCoreError) as e:
              raise DeployerError(
                  f"Failed to create/update auto-scaling configuration "
                  f"{config_name}: {e}"
              )
  
      def _cleanup_deployment_auto_scaling_config(
          self, deployment: DeploymentResponse
      ) -> None:
          """Clean up the auto-scaling configuration associated with a deployment.
  
          Args:
              deployment: The deployment.
          """
          metadata = AppRunnerDeploymentMetadata.from_deployment(deployment)
          config_arn = metadata.auto_scaling_configuration_arn
          if not config_arn:
              return
  
          try:
              logger.debug(f"Deleting auto-scaling configuration {config_arn}")
              self.app_runner_client.delete_auto_scaling_configuration(
                  AutoScalingConfigurationArn=config_arn
              )
          except ClientError as e:
              if e.response["Error"]["Code"] == "ResourceNotFoundException":
                  logger.debug(
                      f"Auto-scaling configuration {config_arn} not found, "
                      "skipping deletion"
                  )
              else:
                  logger.warning(
                      f"Failed to delete auto-scaling configuration {config_arn}: "
                      f"{e}"
                  )
          except Exception as e:
              logger.warning(
                  f"Failed to delete auto-scaling configuration {config_arn}: "
                  f"{e}"
              )
  
      def _prepare_environment_variables(
          self,
          deployment: DeploymentResponse,
          environment: Dict[str, str],
          secrets: Dict[str, str],
          settings: AWSDeployerSettings,
      ) -> Tuple[Dict[str, str], Dict[str, str], Optional[str]]:
          """Prepare environment variables for App Runner, handling secrets appropriately.
  
          Args:
              deployment: The deployment.
              environment: Regular environment variables.
              secrets: Sensitive environment variables.
              settings: The deployer settings.
  
          Returns:
              Tuple containing:
              - Dictionary of regular environment variables.
              - Dictionary of secret environment variables (key -> secret ARN).
              - Optional secret ARN (None if no secrets or fallback to env vars).
          """
          secret_refs = {}
          active_secret_arn: Optional[str] = None
  
          env_vars = {**settings.environment_variables, **environment}
  
          if secrets:
              if settings.use_secrets_manager:
                  # Always store secrets as single JSON secret and reference their
                  # keys in the App Runner service configuration environment
                  # variables.
  
                  secret_name = self._get_secret_name(
                      deployment.name, deployment.id, settings.secret_name_prefix
                  )
  
                  try:
                      secret_value = json.dumps(secrets)
                      secret_arn = self._create_or_update_secret(
                          secret_name, secret_value, deployment, settings
                      )
                      active_secret_arn = secret_arn
  
                      for key in secrets.keys():
                          secret_refs[key] = f"{secret_arn}:{key}::"
  
                      logger.debug(
                          f"Secret {secret_name} stored with ARN {secret_arn} "
                          f"containing {len(secrets)} secret(s)"
                      )
  
                  except Exception as e:
                      logger.warning(
                          f"Failed to create secret, falling back "
                          f"to direct env vars: {e}"
                      )
                      env_vars.update(secrets)
  
                  existing_secret_arn = self._get_secret_arn(deployment)
                  if (
                      existing_secret_arn
                      and existing_secret_arn != active_secret_arn
                  ):
                      # Sometimes the previous secret resource is different from
                      # the new secret resource, e.g. if the secret name changed.
                      # In this case, we need to delete the old secret resource.
                      self._delete_secret(existing_secret_arn)
              else:
                  logger.warning(
                      "Storing secrets directly in environment variables. "
                      "Consider enabling use_secrets_manager for better security."
                  )
                  env_vars.update(secrets)
  
          return env_vars, secret_refs, active_secret_arn
  
      def _get_app_runner_service(
          self, deployment: DeploymentResponse
      ) -> Optional[Dict[str, Any]]:
          """Get an existing App Runner service for a deployment.
  
          Args:
              deployment: The deployment.
  
          Returns:
              The App Runner service dictionary, or None if it doesn't exist.
  
          Raises:
              ClientError: If the App Runner service cannot be described.
          """
          existing_metadata = AppRunnerDeploymentMetadata.from_deployment(
              deployment
          )
  
          if not existing_metadata.service_arn:
              return None
  
          try:
              response = self.app_runner_client.describe_service(
                  ServiceArn=existing_metadata.service_arn
  