import os
import re
import sys
from datetime import datetime
from typing import Any, Dict, List, Literal, Optional, Union

import requests

from lightning_sdk.api import CloudAccountApi
from lightning_sdk.api.deployment_api import (
    Auth,
    AutoScaleConfig,
    AutoScalingMetric,
    BasicAuth,
    DateTimeLike,
    DeploymentApi,
    Env,
    ExecHealthCheck,
    HttpHealthCheck,
    PathLike,
    ReleaseStrategy,
    RequestCaptureExportResult,
    RollingUpdateReleaseStrategy,
    Secret,
    TokenAuth,
    compose_commands,
    restore_auth,
    restore_autoscale,
    restore_env,
    restore_health_check,
    restore_release_strategy,
    to_autoscaling,
    to_byom_spec,
    to_endpoint,
    to_spec,
    to_strategy,
)
from lightning_sdk.api.logs_api import LogsApi
from lightning_sdk.api.utils import (
    AccessibleResource,
    logs_filename,
    raise_access_error_if_not_allowed,
    resolve_logs_path,
)
from lightning_sdk.lightning_cloud import login
from lightning_sdk.lightning_cloud.openapi import V1Deployment, V1DeploymentState
from lightning_sdk.machine import CloudProvider, Machine
from lightning_sdk.organization import Organization
from lightning_sdk.studio import Studio
from lightning_sdk.teamspace import Teamspace
from lightning_sdk.user import User
from lightning_sdk.utils.logging import TrackCallsMeta
from lightning_sdk.utils.resolve import (
    _get_cluster,
    _resolve_default_cloud_account,
    _resolve_org,
    _resolve_teamspace,
    _resolve_user,
)

__all__ = [
    "AutoScaleConfig",
    "AutoScalingMetric",
    "BasicAuth",
    "Deployment",
    "Env",
    "ExecHealthCheck",
    "HttpHealthCheck",
    "ReleaseStrategy",
    "RequestCaptureExportResult",
    "RollingUpdateReleaseStrategy",
    "Secret",
    "TokenAuth",
]

_ENV_NAME_PATTERN = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*$")
_ENV_NAME_ERROR = (
    "Environment variable names must start with a letter or underscore "
    "and contain only letters, numbers, and underscores."
)

# A running replica whose logs are not in the logs API yet has no saved history; its live
# stream is tailed for this many seconds of silence so a download still captures something.
_REPLICA_LOGS_IDLE_TIMEOUT = 8.0


def _validate_env_names(env: Dict[str, str]) -> None:
    if any(_ENV_NAME_PATTERN.fullmatch(name) is None for name in env):
        raise ValueError(_ENV_NAME_ERROR)


class Deployment(metaclass=TrackCallsMeta):
    """The Lightning AI Deployment.

    Allows to fully control a deployment, including retrieving the status, making new release
    and switching machine types, etc..

    Args:
        name: The name or the id of the deployment.
        teamspace: The teamspace in which you want to deploy. Accepts a bare name or an ``owner/teamspace`` slug.
        org: The name of the organization owning the :param`teamspace` in case it is owned by an org. Deprecated —
            pass the owner as part of ``teamspace`` instead, e.g. ``teamspace="owner/teamspace"``.
        user: The name of the user owning the :param`teamspace` in case it is owned directly by a user instead of
            an org. Deprecated — pass the owner as part of ``teamspace`` instead, e.g. ``teamspace="owner/teamspace"``.

    Note:
        Since a teamspace can either be owned by an org or by a user directly,
        only one of the arguments can be provided.

    """

    def __init__(
        self,
        name: Optional[str] = None,
        teamspace: Optional[Union[str, Teamspace]] = None,
        org: Optional[Union[str, Organization]] = None,
        user: Optional[Union[str, User]] = None,
    ) -> None:
        self._request_session: Optional[requests.Session] = None
        self._cloud_account_api = CloudAccountApi()

        # Auth is applied automatically on every API request, so there is no need to
        # authenticate (or resolve the authenticated user) here. The deployment owner is
        # taken from the explicitly provided teamspace/org/user (or the matching env vars
        # / config) below - org/teamspace-scoped API keys have no associated user.
        self._auth = login.Auth()

        if name is None:
            name = "dep_" + datetime.now().strftime("%m-%d_%H:%M:%S")

        self._name = name
        self._user = _resolve_user(user)
        self._org = _resolve_org(org)

        resolved_teamspace = _resolve_teamspace(
            teamspace=teamspace,
            org=org,
            user=user,
        )
        if resolved_teamspace is None:
            raise ValueError(
                "Could not determine the teamspace for your deployment. Pass a teamspace as "
                "'owner/teamspace', set one of LIGHTNING_TEAMSPACE / LIGHTNING_ORG (or LIGHTNING_USERNAME), "
                "or configure a default with 'lightning config set teamspace'."
            )
        self._teamspace = resolved_teamspace

        raise_access_error_if_not_allowed(AccessibleResource.Deployments, self._teamspace.id)

        self._deployment_api = DeploymentApi()
        self._cloud_account = _get_cluster(client=self._deployment_api._client, project_id=self._teamspace.id)
        self._is_created = False

        if name.startswith("dep_"):
            deployment = self._deployment_api.get_deployment_by_id(name, self._teamspace.id)
        else:
            deployment = self._deployment_api.get_deployment_by_name(name, self._teamspace.id)

        # Deployment payloads are generated API models and remain opaque here.
        self._deployment: Any
        if deployment:
            self._name = deployment.name
            self._is_created = True
            self._deployment = deployment
        else:
            self._deployment = None

    def start(
        self,
        studio: Optional[Union[str, Studio]] = None,
        cloud: Optional[Union[CloudProvider, str]] = None,
        machine: Optional[Machine] = None,
        image: Optional[str] = None,
        autoscale: Optional[AutoScaleConfig] = None,
        ports: Optional[Union[float, List[float]]] = None,
        release_strategy: Optional[ReleaseStrategy] = None,
        entrypoint: Optional[str] = None,
        command: Optional[str] = None,
        commands: Optional[List[str]] = None,
        env: Union[List[Union[Secret, Env]], Dict[str, str], None] = None,
        spot: Optional[bool] = None,
        replicas: Optional[int] = None,
        health_check: Optional[Union[HttpHealthCheck, ExecHealthCheck]] = None,
        auth: Optional[Union[BasicAuth, TokenAuth]] = None,
        custom_domain: Optional[str] = None,
        cloudspace_id: Optional[str] = None,
        quantity: Optional[int] = None,
        include_credentials: Optional[bool] = None,
        from_onboarding: Optional[bool] = None,
        from_litserve: Optional[bool] = None,
        max_runtime: Optional[int] = None,
        path_mappings: Optional[Dict[str, str]] = None,
        model: Optional[str] = None,
        hf_token_secret: Optional[str] = None,
        base_image_variant: Optional[str] = None,
        tensor_parallel_size: Optional[int] = None,
        max_model_len: Optional[int] = None,
        gpu_memory_utilization: Optional[float] = None,
        quantization: Optional[str] = None,
        dtype: Optional[str] = None,
        extra_vllm_args: Optional[List[str]] = None,
        enable_weight_reload: Optional[bool] = None,
        acknowledged_warnings: Optional[List[str]] = None,
    ) -> None:
        """The Lightning AI Deployment.

        This method creates the first release of the deployment.
        If the deployment already exists, it sets the deployment to the running state.

        Args:
            name: The name of the deployment.
            cloud: Cloud provider or cloud account to run replicas on.
            machine: The machine used by the deployment replicas.
            autoscale: The list of the metrics to autoscale on.
            ports: The ports to reach your replica services.
            image: The environement used by the deployment. Currentely, only docker images.
            release_strategy: The release strategy to use when changing core deployment specs.
            entrypoint: The docker container entrypoint.
            command: The docker container command.
            env: The environment variables or secrets to use.
            spot: Whether to use spot instances for the replicas.
            replicas: The number of replicas to deploy with.
            health_check: The health check config to know whether your service is ready to receive traffic.
            auth: The auth config to protect your services. Only Basic and Token supported.
            custom_domain: Whether your service would be referenced under a custom domain.
            cloudspace_id: Connect deployment to a Studio.
            quantity: The number of machines per replica to deploy.
            include_credentials: Whether to include the environment variables for the SDK to authenticate
            from_onboarding: Whether the deployment is from onboarding.
            max_runtime: the duration (in seconds) for which to allocate the machine.
                Irrelevant for most machines, required for some of the top-end machines on GCP.
                If in doubt, set it. Won't have an effect on machines not requiring it.
                Defaults to 3h
            path_mappings: Maps container paths to data-connection paths in the form
                ``{"<CONTAINER_PATH>": "<CONNECTION_NAME>:<PATH>"}`` or ``{"<CONTAINER_PATH>": "<CONNECTION_NAME>"}``
                for the root of a connection. Only applicable when deploying docker containers.

        """
        raise_access_error_if_not_allowed(AccessibleResource.Deployments, self._teamspace.id)
        if self._is_created:
            # When restarting a stopped deployment, apply the same defaults as 'Start' button in UI
            # and first-time deployment creation to ensure non-zero replica count
            if replicas is None and autoscale is None:
                replicas = 1
            if autoscale is None:
                autoscale = AutoScaleConfig(
                    min_replicas=0,
                    max_replicas=1,
                    metric="CPU",
                    threshold=90,
                )

            self._deployment.desired_state = V1DeploymentState.RUNNING
            self.update(
                machine=machine,
                cloud=cloud,
                image=image,
                entrypoint=entrypoint,
                command=command,
                commands=commands,
                env=env,
                spot=spot,
                health_check=health_check,
                release_strategy=release_strategy,
                replicas=replicas,
                min_replicas=autoscale.min_replicas if autoscale else None,
                max_replicas=autoscale.max_replicas if autoscale else None,
                ports=ports if isinstance(ports, list) else ([ports] if ports else None),
                auth=auth,
                custom_domain=custom_domain,
                quantity=quantity,
                include_credentials=include_credentials,
                max_runtime=max_runtime,
                path_mappings=path_mappings,
            )
            return

        machine_image_version = None
        cloud_account = None

        if isinstance(studio, Studio):
            cloudspace_id = studio._studio.id
            cloud_account = studio._studio.cluster_id
            machine_image_version = studio._studio.machine_image_version

        if isinstance(studio, str):
            studio = Studio(studio)
            cloudspace_id = studio._studio.id
            cloud_account = studio._studio.cluster_id
            machine_image_version = studio._studio.machine_image_version

        if cloud_account is None:
            cloud_account = _resolve_default_cloud_account(None)
        if cloud is not None:
            cloud_account = None

        if cloud_account is None and self._cloud_account is not None and cloud is None:
            # informational -> stderr, so callers reading stdout (e.g. `--json`) get clean output
            print(f"No cloud account was provided, defaulting to {self._cloud_account.cluster_id}", file=sys.stderr)
            cloud_account = os.getenv("LIGHTNING_CLUSTER_ID") or self._cloud_account.cluster_id

        resolve_cloud = cloud if cloud_account is None else None
        _cloud_account = self._cloud_account_api.resolve_cloud_account(
            self.teamspace.id,
            cloud=resolve_cloud or cloud_account,
            default_cloud_account=self._teamspace.default_cloud_account,
        )

        if isinstance(ports, float):
            ports = [ports]

        if replicas is None and autoscale is None:
            replicas = 1

        if machine is None:
            machine = Machine.CPU

        if commands is not None and command is not None:
            raise ValueError("Commands and command are mutually exclusive")

        if commands is not None:
            command = compose_commands(commands)

        autoscaling_metric_name: Literal["CPU", "GPU"] = (
            ("CPU" if machine.is_cpu() else "GPU") if isinstance(machine, Machine) else "CPU"
        )

        if autoscale is None:
            autoscale = AutoScaleConfig(
                min_replicas=0,
                max_replicas=1,
                metric=autoscaling_metric_name,
                threshold=90,
            )

        self._deployment = self._deployment_api.create_deployment(
            V1Deployment(
                autoscaling=to_autoscaling(autoscale, replicas),
                endpoint=to_endpoint(ports, auth, custom_domain),
                name=self._name,
                project_id=self._teamspace.id,
                replicas=replicas,
                cloudspace_id=cloudspace_id,
                spec=to_spec(
                    cloud_account=_cloud_account,
                    command=command,
                    entrypoint=entrypoint,
                    env=env,
                    image=image,
                    spot=spot,
                    machine=machine,
                    health_check=health_check,
                    quantity=quantity,
                    include_credentials=include_credentials if include_credentials is not None else True,
                    cloudspace_id=cloudspace_id,
                    max_runtime=max_runtime,
                    machine_image_version=machine_image_version,
                    path_mappings=path_mappings,
                    byom=model is not None,
                ),
                strategy=to_strategy(release_strategy),
                byom_spec=to_byom_spec(
                    model,
                    hf_token_secret=hf_token_secret,
                    base_image_variant=base_image_variant,
                    tensor_parallel_size=tensor_parallel_size,
                    max_model_len=max_model_len,
                    gpu_memory_utilization=gpu_memory_utilization,
                    quantization=quantization,
                    dtype=dtype,
                    extra_vllm_args=extra_vllm_args,
                    enable_weight_reload=enable_weight_reload,
                ),
                acknowledged_warnings=acknowledged_warnings,
            ),
            from_onboarding=from_onboarding,
            from_litserve=from_litserve,
        )

        # Overrides the name
        self._name = self._deployment._name
        self._is_created = True

    def update(
        self,
        # Changing those arguments create a new release
        machine: Optional[Machine] = None,
        cloud: Optional[Union[CloudProvider, str]] = None,
        image: Optional[str] = None,
        entrypoint: Optional[str] = None,
        command: Optional[str] = None,
        commands: Optional[List[str]] = None,
        env: Union[List[Union[Env, Secret]], Dict[str, str], None] = None,
        spot: Optional[bool] = None,
        health_check: Optional[Union[HttpHealthCheck, ExecHealthCheck]] = None,
        # Changing those arguments don't create a new release
        min_replicas: Optional[int] = None,
        max_replicas: Optional[int] = None,
        name: Optional[str] = None,
        ports: Optional[List[float]] = None,
        release_strategy: Optional[ReleaseStrategy] = None,
        replicas: Optional[int] = None,
        auth: Optional[Auth] = None,
        custom_domain: Optional[str] = None,
        quantity: Optional[int] = None,
        include_credentials: Optional[bool] = None,
        max_runtime: Optional[int] = None,
        path_mappings: Optional[Dict[str, str]] = None,
    ) -> None:
        """Update the deployment configuration.

        Changes to ``machine``, ``image``, ``entrypoint``, ``command``/``commands``, ``env``,
        ``spot``, ``cloud``, or ``health_check`` create a new release.  All other
        arguments (replica counts, ports, auth, etc.) are applied in-place without a new release.

        Args:
            machine: New machine type for the replicas.
            cloud: Cloud provider or cloud account to run replicas on.
            image: New docker image for the replicas.
            entrypoint: New container entrypoint.
            command: New container command string.
            commands: List of command strings, joined into a single command.  Mutually exclusive
                with ``command``.
            env: Environment variables or secrets to inject into replicas.
            spot: Whether to use spot/interruptible instances.
            health_check: Readiness probe configuration.
            min_replicas: New minimum replica count for autoscaling.
            max_replicas: New maximum replica count for autoscaling.
            name: New display name for the deployment.
            ports: Exposed port numbers.
            release_strategy: How to roll out the new release (e.g. blue/green).
            replicas: Fixed number of replicas (overrides autoscaling).
            auth: Authentication configuration for the deployment endpoint.
            custom_domain: Custom domain to attach to the deployment.
            quantity: Number of machines per replica.
            include_credentials: Whether to inject SDK authentication env vars.
            max_runtime: Maximum machine allocation duration in seconds.
            path_mappings: Container-path → data-connection-path mappings for docker jobs.
        """
        raise_access_error_if_not_allowed(AccessibleResource.Deployments, self._teamspace.id)
        cloud_account = _resolve_default_cloud_account(None)
        if cloud is not None:
            cloud_account = None
        if cloud is not None:
            cloud_account = self._cloud_account_api.resolve_cloud_account(
                self.teamspace.id,
                cloud=cloud,
                default_cloud_account=self._teamspace.default_cloud_account,
            )

        if command is None and commands is not None:
            command = compose_commands(commands)

        self._deployment = self._deployment_api.update_deployment(
            self._deployment,
            name=name or self._name,
            spot=spot,
            replicas=replicas,
            min_replicas=min_replicas,
            max_replicas=max_replicas,
            cloud_account=cloud_account,
            machine=machine,
            image=image,
            entrypoint=entrypoint,
            command=command,
            ports=ports,
            custom_domain=custom_domain,
            auth=auth,
            env=env,
            health_check=health_check,
            release_strategy=release_strategy,
            quantity=quantity,
            include_credentials=include_credentials,
            max_runtime=max_runtime,
            path_mappings=path_mappings,
        )
        if self._deployment is not None:
            self._name = self._deployment.name

    def stop(self) -> None:
        """All the deployment replicas will be stopped and all their traffic blocked."""
        self._deployment = self._deployment_api.stop(self._deployment)

    def export_request_captures(
        self,
        *,
        start: DateTimeLike,
        end: DateTimeLike,
        output_dir: PathLike,
        paths: Optional[List[str]] = None,
        include_all_paths: bool = False,
        status_codes: Optional[List[int]] = None,
        strict: bool = False,
        request_timeout: Union[float, tuple] = 60,
        overwrite: bool = False,
        remote_path: Optional[str] = None,
    ) -> RequestCaptureExportResult:
        """Export captured request telemetry to local artifacts.

        The export is always written to ``output_dir``. If ``remote_path`` is provided,
        the same artifacts are also uploaded to that remote destination. This currently
        supports ``lightning_storage`` targets only.

        Args:
            start: Start of the time range to export captures for.
            end: End of the time range to export captures for.
            output_dir: Local directory where exported artifacts will be written.
            paths: Optional list of request paths to filter exports by.
            include_all_paths: If ``True``, export captures for all request paths.
            status_codes: Optional list of HTTP status codes to filter exports by.
            strict: If ``True``, raise an error on any export failure instead of skipping.
            request_timeout: Timeout in seconds (or ``(connect, read)`` tuple) for each request.
            overwrite: If ``True``, overwrite existing files in ``output_dir``.
            remote_path: Optional remote destination path for uploading exported artifacts.

        Returns:
            RequestCaptureExportResult: Summary of the export operation.

        Raises:
            RuntimeError: If the deployment has not been created yet.
        """
        if not self._deployment:
            raise RuntimeError("This deployment has not been created.")

        return self._deployment_api.export_request_captures(
            self._deployment,
            start=start,
            end=end,
            output_dir=output_dir,
            paths=paths,
            include_all_paths=include_all_paths,
            status_codes=status_codes,
            strict=strict,
            request_timeout=request_timeout,
            overwrite=overwrite,
            remote_path=remote_path,
        )

    def download_logs(
        self,
        *,
        job_id: Optional[str] = None,
        timestamps: bool = False,
    ) -> str:
        """Download a deployment replica's stored logs to a file and return the written path."""
        if not self._deployment or not self.id:
            raise RuntimeError("This deployment has not been created.")

        teamspace_id = self._teamspace.id
        if job_id is None:
            jobs = self._deployment_api.list_deployment_jobs(teamspace_id, self.id, limit=100)
            if not jobs:
                raise RuntimeError(f"Deployment '{self.name}' has no replicas to download logs from.")
            if len(jobs) > 1:
                available = ", ".join(job.id for job in jobs)
                raise ValueError(
                    f"Deployment '{self.name}' has {len(jobs)} replicas; pass job_id to pick one. "
                    f"Available replica ids: {available}."
                )
            job_id = jobs[0].id

        entries = LogsApi().stream(
            teamspace_id,
            job_ids=[job_id],
            # A running replica whose lines are not in the logs API yet has no saved history;
            # tail its live stream briefly so the download still captures what is there.
            idle_timeout=_REPLICA_LOGS_IDLE_TIMEOUT,
            fallback_to_live=True,
        )
        lines = [entry.format(timestamps=timestamps) for entry in entries]
        text = "\n".join(lines)
        if lines:
            text += "\n"
        dest = resolve_logs_path(None, logs_filename("deployment", self.name or self.id))
        dest.write_text(text, encoding="utf-8")
        return str(dest)

    @property
    def id(self) -> Optional[str]:
        """The deployment's unique identifier, or None before it is created."""
        return self._deployment.id if self._deployment else None

    @property
    def name(self) -> Optional[str]:
        if self._deployment:
            self._deployment = self._deployment_api.get_deployment_by_name(self._name, self._teamspace.id)
            self._name = self._deployment.name
        return self._name

    @property
    def replicas(self) -> Optional[int]:
        """The default number of replicas the release starts with.

        Returns:
            Optional[int]: The replica count, or ``None`` if the deployment has not been started.
        """
        if self._deployment:
            self._deployment = self._deployment_api.get_deployment_by_name(self._name, self._teamspace.id)
            return self._deployment.replicas
        return None

    @property
    def min_replicas(self) -> Optional[int]:
        """The minimum number of replicas.

        Returns:
            Optional[int]: The autoscaling minimum replica count, or ``None`` if not started.
        """
        if self._deployment:
            self._deployment = self._deployment_api.get_deployment_by_name(self._name, self._teamspace.id)
            return int(self._deployment.autoscaling.min_replicas)
        return None

    @property
    def max_replicas(self) -> Optional[int]:
        """The maximum number of replicas.

        Returns:
            Optional[int]: The autoscaling maximum replica count, or ``None`` if not started.
        """
        if self._deployment:
            self._deployment = self._deployment_api.get_deployment_by_name(self._name, self._teamspace.id)
            return int(self._deployment.autoscaling.max_replicas)
        return None

    @property
    def ports(self) -> Optional[List[int]]:
        """The exposed ports on which you can reach your deployment.

        Returns:
            Optional[List[int]]: List of exposed port numbers, or ``None`` if not started.
        """
        if self._deployment:
            self._deployment = self._deployment_api.get_deployment_by_name(self._name, self._teamspace.id)
            return [int(p) for p in self._deployment.endpoint.ports]
        return None

    @property
    def release_strategy(self) -> Optional[ReleaseStrategy]:
        """The release strategy of the deployment.

        Returns:
            Optional[ReleaseStrategy]: The current release strategy, or ``None`` if not started.
        """
        if self._deployment:
            self._deployment = self._deployment_api.get_deployment_by_name(self._name, self._teamspace.id)
            return restore_release_strategy(self._deployment.strategy)
        return None

    @property
    def health_check(self) -> Optional[Union[HttpHealthCheck, ExecHealthCheck]]:
        """The health check to validate the replicas are ready to receive traffic.

        Returns:
            Optional[Union[HttpHealthCheck, ExecHealthCheck]]: The readiness probe configuration,
            or ``None`` if not started.
        """
        if self._deployment:
            self._deployment = self._deployment_api.get_deployment_by_name(self._name, self._teamspace.id)
            return restore_health_check(self._deployment.spec.readiness_probe)
        return None

    @property
    def auth(self) -> Optional[Auth]:
        """The authentification configuration of the deployment.

        Returns:
            Optional[Auth]: The authentication configuration, or ``None`` if not started.
        """
        if self._deployment:
            self._deployment = self._deployment_api.get_deployment_by_name(self._name, self._teamspace.id)
            return restore_auth(self._deployment.endpoint.auth)
        return None

    @property
    def autoscale(self) -> Optional[AutoScaleConfig]:
        """The autoscaling configuration of the deployment.

        Returns:
            Optional[AutoScaleConfig]: The autoscaling configuration, or ``None`` if not started.
        """
        if self._deployment:
            self._deployment = self._deployment_api.get_deployment_by_name(self._name, self._teamspace.id)
            return restore_autoscale(self._deployment.autoscaling)
        return None

    @property
    def env(self) -> Optional[List[Union[Secret, Env]]]:
        """The env configuration of the deployment.

        Returns:
            Optional[List[Union[Secret, Env]]]: The environment variables and secrets, or ``None`` if not started.
        """
        if self._deployment:
            self._deployment = self._deployment_api.get_deployment_by_name(self._name, self._teamspace.id)
            return restore_env(self._deployment.spec.env)
        return None

    def set_env(self, new_env: Dict[str, str], partial: bool = True) -> None:
        """Set directly configured literal Deployment environment variables.

        Args:
            new_env: Literal environment variable names and values.
            partial: Preserve unrelated literal variables when ``True``. Secret references
                are preserved unless shadowed by a new literal in either mode.

        Raises:
            ValueError: If a name is invalid or the Deployment does not exist.
        """
        _validate_env_names(new_env)
        current = self.env
        if current is None:
            raise ValueError("Deployment must exist before its environment can be changed.")

        names = set(new_env)
        if partial:
            kept = [entry for entry in current if entry.env_var_name not in names]
        else:
            kept = [entry for entry in current if isinstance(entry, Secret) and entry.env_var_name not in names]
        self.update(env=[*kept, *(Env(name=name, value=value) for name, value in new_env.items())])

    def delete_env(self, key: str) -> None:
        """Delete one directly configured literal Deployment environment variable.

        Args:
            key: Name of the literal environment variable to delete.

        Raises:
            ValueError: If the name is invalid, the Deployment does not exist, or the
                literal variable is not directly configured.
        """
        _validate_env_names({key: ""})
        current = self.env
        if current is None:
            raise ValueError("Deployment must exist before its environment can be changed.")
        if not any(isinstance(entry, Env) and entry.name == key for entry in current):
            raise ValueError(f"Deployment environment variable {key!r} was not found.")
        self.update(env=[entry for entry in current if not (isinstance(entry, Env) and entry.name == key)])

    def set_secret(self, secret_name: str, env_name: Optional[str] = None) -> None:
        """Reference a teamspace secret as a Deployment environment variable.

        Args:
            secret_name: Name of the teamspace secret to inject.
            env_name: Environment variable name to inject the secret under. Defaults to
                the secret name.

        Raises:
            ValueError: If a name is invalid or the Deployment does not exist.
        """
        injected = env_name or secret_name
        _validate_env_names({secret_name: "", injected: ""})
        current = self.env
        if current is None:
            raise ValueError("Deployment must exist before its environment can be changed.")
        kept = [entry for entry in current if entry.env_var_name != injected]
        self.update(env=[*kept, Secret(name=secret_name, env_name=env_name)])

    def delete_secret(self, env_name: str) -> None:
        """Remove a teamspace secret reference from the Deployment.

        Args:
            env_name: Injected environment variable name of the secret reference to remove.

        Raises:
            ValueError: If the name is invalid, the Deployment does not exist, or no secret
                is injected under that name.
        """
        _validate_env_names({env_name: ""})
        current = self.env
        if current is None:
            raise ValueError("Deployment must exist before its environment can be changed.")
        if not any(isinstance(entry, Secret) and entry.env_var_name == env_name for entry in current):
            raise ValueError(f"Deployment secret {env_name!r} was not found.")
        kept = [e for e in current if not (isinstance(e, Secret) and e.env_var_name == env_name)]
        self.update(env=kept)

    @property
    def urls(self) -> Optional[List[str]]:
        """The urls to reach the deployment.

        Returns:
            Optional[List[str]]: The list of URLs for the deployment, or ``None`` if not started.
        """
        if self._deployment:
            self._deployment = self._deployment_api.get_deployment_by_name(self._name, self._teamspace.id)
            return self._deployment.status.urls
        return None

    @property
    def pending_replicas(self) -> Optional[int]:
        """The number of pending replicas.

        Returns:
            Optional[int]: The count of replicas currently pending, or ``None`` if not started.
        """
        if self._deployment:
            self._deployment = self._deployment_api.get_deployment_by_name(self._name, self._teamspace.id)
            return int(self._deployment.status.pending_replicas)
        return None

    @property
    def running_replicas(self) -> Optional[int]:
        """The number of failing replicas.

        Returns:
            Optional[int]: The count of replicas currently running and ready, or ``None`` if not started.
        """
        if self._deployment:
            self._deployment = self._deployment_api.get_deployment_by_name(self._name, self._teamspace.id)
            return int(self._deployment.status.ready_replicas)
        return None

    @property
    def failing_replicas(self) -> Optional[int]:
        """The number of failing replicas."""
        if self._deployment:
            self._deployment = self._deployment_api.get_deployment_by_name(self._name, self._teamspace.id)
            return int(self._deployment.status.failing_replicas)
        return None

    @property
    def deleting_replicas(self) -> Optional[int]:
        """The number of deleting replicas."""
        if self._deployment:
            self._deployment = self._deployment_api.get_deployment_by_name(self._name, self._teamspace.id)
            return int(self._deployment.status.deleting_replicas)
        return None

    @property
    def cloud_account(self) -> Optional[str]:
        """The cloud_account of the replicas."""
        if self._deployment:
            self._deployment = self._deployment_api.get_deployment_by_name(self._name, self._teamspace.id)
            return self._deployment.spec.cluster_id
        return None

    @property
    def release_id(self) -> Optional[str]:
        """The release id of the deployment."""
        if self._deployment:
            self._deployment = self._deployment_api.get_deployment_by_name(self._name, self._teamspace.id)
            return self._deployment.release_id
        return None

    @property
    def quantity(self) -> Optional[str]:
        """The number of machines per replica."""
        if self._deployment:
            self._deployment = self._deployment_api.get_deployment_by_name(self._name, self._teamspace.id)
            return self._deployment.spec.quantity
        return None

    @property
    def include_credentials(self) -> Optional[bool]:
        """The number of machines per replica."""
        if self._deployment:
            self._deployment = self._deployment_api.get_deployment_by_name(self._name, self._teamspace.id)
            return self._deployment.spec.include_credentials
        return None

    @property
    def org(self) -> Optional[Organization]:
        return self._org

    @property
    def user(self) -> Optional[User]:
        """The user of the deployment."""
        return self._user

    @property
    def teamspace(self) -> Teamspace:
        """The teamspace of the deployment."""
        return self._teamspace

    @property
    def is_started(self) -> bool:
        return self._is_created

    @property
    def is_stopped(self) -> Optional[bool]:
        if self._deployment:
            self._deployment = self._deployment_api.get_deployment_by_name(self._name, self._teamspace.id)
            return self._deployment.autoscaling.max_replicas == 0
        return None

    @property
    def image(self) -> Optional[str]:
        if self._deployment:
            self._deployment = self._deployment_api.get_deployment_by_name(self._name, self._teamspace.id)
            return self._deployment.spec.image
        return None

    @property
    def entrypoint(self) -> Optional[str]:
        if self._deployment:
            self._deployment = self._deployment_api.get_deployment_by_name(self._name, self._teamspace.id)
            return self._deployment.spec.entrypoint
        return None

    @property
    def command(self) -> Optional[str]:
        if self._deployment:
            self._deployment = self._deployment_api.get_deployment_by_name(self._name, self._teamspace.id)
            return self._deployment.spec.command
        return None

    @property
    def _session(self) -> Any:
        if self._request_session is None:
            session = requests.Session()
            session.headers.update(**self._get_auth_headers())
            self._request_session = session
        return self._request_session

    def _get_auth_headers(self) -> Dict:
        if self._deployment:
            self._deployment = self._deployment_api.get_deployment_by_name(self._name, self._teamspace.id)

        # TODO: Add support for all auth
        return {}

    def _get_url(self, port: Optional[int] = None) -> Any:
        urls = self.urls
        if urls is None:
            return None

        if port is None:
            return urls[0]

        return None

    def _prepare_url(self, path: str = "", port: Optional[int] = None) -> str:
        url = self._get_url(port)
        if url is None:
            raise ValueError("The url wasn't properly defined")

        if path.startswith("/"):
            path = path[1:]

        return f"{url}/{path}"

    def get(self, path: str = "", port: Optional[int] = None, **kwargs: Any) -> Any:
        return self._session.get(self._prepare_url(path, port), verify=False, **kwargs)

    def post(self, path: str = "", port: Optional[int] = None, **kwargs: Any) -> Any:
        return self._session.post(self._prepare_url(path, port), verify=False, **kwargs)

    def put(self, path: str = "", port: Optional[int] = None, **kwargs: Any) -> Any:
        return self._session.put(self._prepare_url(path, port), verify=False, **kwargs)

    def delete(self, path: str = "", port: Optional[int] = None, **kwargs: Any) -> Any:
        return self._session.delete(self._prepare_url(path, port), verify=False, **kwargs)