content: #  Copyright (c) ZenML GmbH 2022. 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.
  """Client implementation."""
  
  import functools
  import json
  import os
  from abc import ABCMeta
  from datetime import datetime
  from pathlib import Path
  from typing import (
      TYPE_CHECKING,
      Any,
      Callable,
      Dict,
      Generator,
      List,
      Mapping,
      Optional,
      Sequence,
      Tuple,
      Type,
      TypeVar,
      Union,
      cast,
  )
  from uuid import UUID
  
  from pydantic import ConfigDict, SecretStr
  
  from zenml.client_lazy_loader import (
      client_lazy_loader,
      evaluate_all_lazy_load_args_in_client_methods,
  )
  from zenml.config.global_config import GlobalConfiguration
  from zenml.config.pipeline_run_configuration import PipelineRunConfiguration
  from zenml.config.source import Source
  from zenml.constants import (
      ENV_ZENML_ACTIVE_PROJECT_ID,
      ENV_ZENML_ACTIVE_STACK_ID,
      ENV_ZENML_ENABLE_REPO_INIT_WARNINGS,
      ENV_ZENML_REPOSITORY_PATH,
      ENV_ZENML_SERVER,
      PAGE_SIZE_DEFAULT,
      PAGINATION_STARTING_PAGE,
      REPOSITORY_DIRECTORY_NAME,
      TEXT_FIELD_MAX_LENGTH,
      handle_bool_env_var,
  )
  from zenml.enums import (
      ArtifactType,
      ColorVariants,
      CuratedVisualizationSize,
      DeploymentStatus,
      LogicalOperators,
      ModelStages,
      OAuthDeviceStatus,
      PluginSubType,
      PluginType,
      ServiceState,
      SorterOps,
      StackComponentType,
      StoreType,
      TaggableResourceTypes,
      VisualizationResourceTypes,
  )
  from zenml.exceptions import (
      AuthorizationException,
      EntityExistsError,
      IllegalOperationError,
      InitializationException,
      ValidationError,
      ZenKeyError,
  )
  from zenml.io import fileio
  from zenml.logger import get_logger
  from zenml.models import (
      ActionFilter,
      ActionRequest,
      ActionResponse,
      ActionUpdate,
      APIKeyFilter,
      APIKeyRequest,
      APIKeyResponse,
      APIKeyRotateRequest,
      APIKeyUpdate,
      ArtifactFilter,
      ArtifactResponse,
      ArtifactUpdate,
      ArtifactVersionFilter,
      ArtifactVersionResponse,
      ArtifactVersionUpdate,
      BaseIdentifiedResponse,
      CodeRepositoryFilter,
      CodeRepositoryRequest,
      CodeRepositoryResponse,
      CodeRepositoryUpdate,
      ComponentFilter,
      ComponentRequest,
      ComponentResponse,
      ComponentUpdate,
      CuratedVisualizationRequest,
      CuratedVisualizationResponse,
      CuratedVisualizationUpdate,
      DeploymentFilter,
      DeploymentResponse,
      EventSourceFilter,
      EventSourceRequest,
      EventSourceResponse,
      EventSourceUpdate,
      FlavorFilter,
      FlavorResponse,
      ModelFilter,
      ModelRequest,
      ModelResponse,
      ModelUpdate,
      ModelVersionArtifactFilter,
      ModelVersionArtifactResponse,
      ModelVersionFilter,
      ModelVersionPipelineRunFilter,
      ModelVersionPipelineRunResponse,
      ModelVersionRequest,
      ModelVersionResponse,
      ModelVersionUpdate,
      OAuthDeviceFilter,
      OAuthDeviceResponse,
      OAuthDeviceUpdate,
      Page,
      PipelineBuildFilter,
      PipelineBuildResponse,
      PipelineFilter,
      PipelineResponse,
      PipelineRunFilter,
      PipelineRunResponse,
      PipelineSnapshotFilter,
      PipelineSnapshotResponse,
      PipelineSnapshotRunRequest,
      PipelineSnapshotUpdate,
      ProjectFilter,
      ProjectRequest,
      ProjectResponse,
      ProjectUpdate,
      RunMetadataRequest,
      RunMetadataResource,
      RunTemplateFilter,
      RunTemplateRequest,
      RunTemplateResponse,
      RunTemplateUpdate,
      ScheduleFilter,
      ScheduleResponse,
      ScheduleUpdate,
      SecretFilter,
      SecretRequest,
      SecretResponse,
      SecretUpdate,
      ServerSettingsResponse,
      ServerSettingsUpdate,
      ServiceAccountFilter,
      ServiceAccountRequest,
      ServiceAccountResponse,
      ServiceAccountUpdate,
      ServiceConnectorFilter,
      ServiceConnectorRequest,
      ServiceConnectorResourcesModel,
      ServiceConnectorResponse,
      ServiceConnectorTypeModel,
      ServiceConnectorUpdate,
      ServiceFilter,
      ServiceRequest,
      ServiceResponse,
      ServiceType,
      ServiceUpdate,
      StackFilter,
      StackRequest,
      StackResponse,
      StackUpdate,
      StepRunFilter,
      StepRunResponse,
      StepRunUpdate,
      TagFilter,
      TagRequest,
      TagResource,
      TagResourceRequest,
      TagResponse,
      TagUpdate,
      TriggerExecutionFilter,
      TriggerExecutionResponse,
      TriggerFilter,
      TriggerRequest,
      TriggerResponse,
      TriggerUpdate,
      UserFilter,
      UserRequest,
      UserResponse,
      UserUpdate,
  )
  from zenml.utils import dict_utils, io_utils, source_utils, tag_utils
  from zenml.utils.dict_utils import dict_to_bytes
  from zenml.utils.filesync_model import FileSyncModel
  from zenml.utils.pagination_utils import depaginate
  from zenml.utils.uuid_utils import is_valid_uuid
  
  if TYPE_CHECKING:
      from zenml.metadata.metadata_types import MetadataType, MetadataTypeEnum
      from zenml.orchestrators import BaseOrchestrator
      from zenml.service_connectors.service_connector import ServiceConnector
      from zenml.services.service import ServiceConfig
      from zenml.stack import Stack
      from zenml.zen_stores.base_zen_store import BaseZenStore
  
  logger = get_logger(__name__)
  
  AnyResponse = TypeVar("AnyResponse", bound=BaseIdentifiedResponse)  # type: ignore[type-arg]
  F = TypeVar("F", bound=Callable[..., Any])
  
  
  class ClientConfiguration(FileSyncModel):
      """Pydantic object used for serializing client configuration options."""
  
      _active_project: Optional["ProjectResponse"] = None
      active_project_id: Optional[UUID] = None
      active_stack_id: Optional[UUID] = None
      _active_stack: Optional["StackResponse"] = None
  
      @property
      def active_project(self) -> "ProjectResponse":
          """Get the active project for the local client.
  
          Returns:
              The active project.
  
          Raises:
              RuntimeError: If no active project is set.
          """
          if self._active_project:
              return self._active_project
          else:
              raise RuntimeError(
                  "No active project is configured. Run "
                  "`zenml project set <NAME>` to set the active "
                  "project."
              )
  
      def set_active_project(self, project: "ProjectResponse") -> None:
          """Set the project for the local client.
  
          Args:
              project: The project to set active.
          """
          self._active_project = project
          self.active_project_id = project.id
  
      def set_active_stack(self, stack: "StackResponse") -> None:
          """Set the stack for the local client.
  
          Args:
              stack: The stack to set active.
          """
          self.active_stack_id = stack.id
          self._active_stack = stack
  
      model_config = ConfigDict(
          # Validate attributes when assigning them. We need to set this in order
          # to have a mix of mutable and immutable attributes
          validate_assignment=True,
          # Allow extra attributes from configs of previous ZenML versions to
          # permit downgrading
          extra="allow",
      )
  
  
  class ClientMetaClass(ABCMeta):
      """Client singleton metaclass.
  
      This metaclass is used to enforce a singleton instance of the Client
      class with the following additional properties:
  
      * the singleton Client instance is created on first access to reflect
      the global configuration and local client configuration.
      * the Client shouldn't be accessed from within pipeline steps (a warning
      is logged if this is attempted).
      """
  
      def __init__(cls, *args: Any, **kwargs: Any) -> None:
          """Initialize the Client class.
  
          Args:
              *args: Positional arguments.
              **kwargs: Keyword arguments.
          """
          super().__init__(*args, **kwargs)
          cls._global_client: Optional["Client"] = None
  
      def __call__(cls, *args: Any, **kwargs: Any) -> "Client":
          """Create or return the global Client instance.
  
          If the Client constructor is called with custom arguments,
          the singleton functionality of the metaclass is bypassed: a new
          Client instance is created and returned immediately and without
          saving it as the global Client singleton.
  
          Args:
              *args: Positional arguments.
              **kwargs: Keyword arguments.
  
          Returns:
              Client: The global Client instance.
          """
          if args or kwargs:
              return cast("Client", super().__call__(*args, **kwargs))
  
          if not cls._global_client:
              cls._global_client = cast(
                  "Client", super().__call__(*args, **kwargs)
              )
  
          return cls._global_client
  
  
  def _fail_for_sql_zen_store(method: F) -> F:
      """Decorator for methods that are not allowed with a SQLZenStore.
  
      Args:
          method: The method to decorate.
  
      Returns:
          The decorated method.
      """
  
      @functools.wraps(method)
      def wrapper(self: "Client", *args: Any, **kwargs: Any) -> Any:
          # No isinstance check to avoid importing ZenStore implementations
          if self.zen_store.__class__.__name__ == "SqlZenStore":
              raise TypeError(
                  "This method is not allowed when not connected "
                  "to a ZenML Server through the API interface."
              )
          return method(self, *args, **kwargs)
  
      return cast(F, wrapper)
  
  
  @evaluate_all_lazy_load_args_in_client_methods
  class Client(metaclass=ClientMetaClass):
      """ZenML client class.
  
      The ZenML client manages configuration options for ZenML stacks as well
      as their components.
      """
  
      _active_user: Optional["UserResponse"] = None
      _active_project: Optional["ProjectResponse"] = None
      _active_stack: Optional["StackResponse"] = None
  
      def __init__(
          self,
          root: Optional[Path] = None,
      ) -> None:
          """Initializes the global client instance.
  
          Client is a singleton class: only one instance can exist. Calling
          this constructor multiple times will always yield the same instance (see
          the exception below).
  
          The `root` argument is only meant for internal use and testing purposes.
          User code must never pass them to the constructor.
          When a custom `root` value is passed, an anonymous Client instance
          is created and returned independently of the Client singleton and
          that will have no effect as far as the rest of the ZenML core code is
          concerned.
  
          Instead of creating a new Client instance to reflect a different
          repository root, to change the active root in the global Client,
          call `Client().activate_root(<new-root>)`.
  
          Args:
              root: (internal use) custom root directory for the client. If
                  no path is given, the repository root is determined using the
                  environment variable `ZENML_REPOSITORY_PATH` (if set) and by
                  recursively searching in the parent directories of the
                  current working directory. Only used to initialize new
                  clients internally.
          """
          self._root: Optional[Path] = None
          self._config: Optional[ClientConfiguration] = None
  
          self._set_active_root(root)
  
      @classmethod
      def get_instance(cls) -> Optional["Client"]:
          """Return the Client singleton instance.
  
          Returns:
              The Client singleton instance or None, if the Client hasn't
              been initialized yet.
          """
          return cls._global_client
  
      @classmethod
      def _reset_instance(cls, client: Optional["Client"] = None) -> None:
          """Reset the Client singleton instance.
  
          This method is only meant for internal use and testing purposes.
  
          Args:
              client: The Client instance to set as the global singleton.
                  If None, the global Client singleton is reset to an empty
                  value.
          """
          cls._global_client = client
  
      def _set_active_root(self, root: Optional[Path] = None) -> None:
          """Set the supplied path as the repository root.
  
          If a client configuration is found at the given path or the
          path, it is loaded and used to initialize the client.
          If no client configuration is found, the global configuration is
          used instead to manage the active stack, project etc.
  
          Args:
              root: The path to set as the active repository root. If not set,
                  the repository root is determined using the environment
                  variable `ZENML_REPOSITORY_PATH` (if set) and by recursively
                  searching in the parent directories of the current working
                  directory.
          """
          enable_warnings = handle_bool_env_var(
              ENV_ZENML_ENABLE_REPO_INIT_WARNINGS, False
          )
          self._root = self.find_repository(
              root, enable_warnings=enable_warnings
          )
  
          if not self._root:
              self._config = None
              if enable_warnings:
                  logger.info("Running without an active repository root.")
          else:
              logger.debug("Using repository root %s.", self._root)
              self._config = self._load_config()
  
          # Sanitize the client configuration to reflect the current
          # settings
          self._sanitize_config()
  
      def _config_path(self) -> Optional[str]:
          """Path to the client configuration file.
  
          Returns:
              Path to the client configuration file or None if the client
              root has not been initialized yet.
          """
          if not self.config_directory:
              return None
          return str(self.config_directory / "config.yaml")
  
      def _sanitize_config(self) -> None:
          """Sanitize and save the client configuration.
  
          This method is called to ensure that the client configuration
          doesn't contain outdated information, such as an active stack or
          project that no longer exists.
          """
          if not self._config:
              return
  
          active_project, active_stack = self.zen_store.validate_active_config(
              self._config.active_project_id,
              self._config.active_stack_id,
              config_name="repo",
          )
          self._config.set_active_stack(active_stack)
          if active_project:
              self._config.set_active_project(active_project)
  
      def _load_config(self) -> Optional[ClientConfiguration]:
          """Loads the client configuration from disk.
  
          This happens if the client has an active root and the configuration
          file exists. If the configuration file doesn't exist, an empty
          configuration is returned.
  
          Returns:
              Loaded client configuration or None if the client does not
              have an active root.
          """
          config_path = self._config_path()
          if not config_path:
              return None
  
          # load the client configuration file if it exists, otherwise use
          # an empty configuration as default
          if fileio.exists(config_path):
              logger.debug(f"Loading client configuration from {config_path}.")
          else:
              logger.debug(
                  "No client configuration file found, creating default "
                  "configuration."
              )
  
          return ClientConfiguration(config_file=config_path)
  
      @staticmethod
      def initialize(
          root: Optional[Path] = None,
      ) -> None:
          """Initializes a new ZenML repository at the given path.
  
          Args:
              root: The root directory where the repository should be created.
                  If None, the current working directory is used.
  
          Raises:
              InitializationException: If the root directory already contains a
                  ZenML repository.
          """
          root = root or Path.cwd()
          logger.debug("Initializing new repository at path %s.", root)
          if Client.is_repository_directory(root):
              raise InitializationException(
                  f"Found existing ZenML repository at path '{root}'."
              )
  
          config_directory = str(root / REPOSITORY_DIRECTORY_NAME)
          io_utils.create_dir_recursive_if_not_exists(config_directory)
          # Initialize the repository configuration at the custom path
          Client(root=root)
  
      @property
      def uses_local_configuration(self) -> bool:
          """Check if the client is using a local configuration.
  
          Returns:
              True if the client is using a local configuration,
              False otherwise.
          """
          return self._config is not None
  
      @staticmethod
      def is_repository_directory(path: Path) -> bool:
          """Checks whether a ZenML client exists at the given path.
  
          Args:
              path: The path to check.
  
          Returns:
              True if a ZenML client exists at the given path,
              False otherwise.
          """
          config_dir = path / REPOSITORY_DIRECTORY_NAME
          return fileio.isdir(str(config_dir))
  
      @staticmethod
      def find_repository(
          path: Optional[Path] = None, enable_warnings: bool = False
      ) -> Optional[Path]:
          """Search for a ZenML repository directory.
  
          Args:
              path: Optional path to look for the repository. If no path is
                  given, this function tries to find the repository using the
                  environment variable `ZENML_REPOSITORY_PATH` (if set) and
                  recursively searching in the parent directories of the current
                  working directory.
              enable_warnings: If `True`, warnings are printed if the repository
                  root cannot be found.
  
          Returns:
              Absolute path to a ZenML repository directory or None if no
              repository directory was found.
          """
          if not path:
              # try to get path from the environment variable
              env_var_path = os.getenv(ENV_ZENML_REPOSITORY_PATH)
              if env_var_path:
                  path = Path(env_var_path)
  
          if path:
              # explicit path via parameter or environment variable, don't search
              # parent directories
              search_parent_directories = False
              warning_message = (
                  f"Unable to find ZenML repository at path '{path}'. Make sure "
                  f"to create a ZenML repository by calling `zenml init` when "
                  f"specifying an explicit repository path in code or via the "
                  f"environment variable '{ENV_ZENML_REPOSITORY_PATH}'."
              )
          else:
              # try to find the repository in the parent directories of the
              # current working directory
              path = Path.cwd()
              search_parent_directories = True
              warning_message = (
                  f"Unable to find ZenML repository in your current working "
                  f"directory ({path}) or any parent directories. If you "
                  f"want to use an existing repository which is in a different "
                  f"location, set the environment variable "
                  f"'{ENV_ZENML_REPOSITORY_PATH}'. If you want to create a new "
                  f"repository, run `zenml init`."
              )
  
          def _find_repository_helper(path_: Path) -> Optional[Path]:
              """Recursively search parent directories for a ZenML repository.
  
              Args:
                  path_: The path to search.
  
              Returns:
                  Absolute path to a ZenML repository directory or None if no
                  repository directory was found.
              """
              if Client.is_repository_directory(path_):
                  return path_
  
              if not search_parent_directories or io_utils.is_root(str(path_)):
                  return None
  
              return _find_repository_helper(path_.parent)
  
          repository_path = _find_repository_helper(path)
  
          if repository_path:
              return repository_path.resolve()
          if enable_warnings:
              logger.warning(warning_message)
          return None
  
      @staticmethod
      def is_inside_repository(file_path: str) -> bool:
          """Returns whether a file is inside the active ZenML repository.
  
          Args:
              file_path: A file path.
  
          Returns:
              True if the file is inside the active ZenML repository, False
              otherwise.
          """
          if repo_path := Client.find_repository():
              return repo_path in Path(file_path).resolve().parents
          return False
  
      @property
      def zen_store(self) -> "BaseZenStore":
          """Shortcut to return the global zen store.
  
          Returns:
              The global zen store.
          """
          return GlobalConfiguration().zen_store
  
      @property
      def root(self) -> Optional[Path]:
          """The root directory of this client.
  
          Returns:
              The root directory of this client, or None, if the client
              has not been initialized.
          """
          return self._root
  
      @property
      def config_directory(self) -> Optional[Path]:
          """The configuration directory of this client.
  
          Returns:
              The configuration directory of this client, or None, if the
              client doesn't have an active root.
          """
          return self.root / REPOSITORY_DIRECTORY_NAME if self.root else None
  
      def activate_root(self, root: Optional[Path] = None) -> None:
          """Set the active repository root directory.
  
          Args:
              root: The path to set as the active repository root. If not set,
                  the repository root is determined using the environment
                  variable `ZENML_REPOSITORY_PATH` (if set) and by recursively
                  searching in the parent directories of the current working
                  directory.
          """
          self._set_active_root(root)
  
      def set_active_project(
          self, project_name_or_id: Union[str, UUID]
      ) -> "ProjectResponse":
          """Set the project for the local client.
  
          Args:
              project_name_or_id: The name or ID of the project to set active.
  
          Returns:
              The model of the active project.
          """
          project = self.zen_store.get_project(
              project_name_or_id=project_name_or_id
          )  # raises KeyError
          if self._config:
              self._config.set_active_project(project)
              # Sanitize the client configuration to reflect the current
              # settings
              self._sanitize_config()
          else:
              # set the active project globally only if the client doesn't use
              # a local configuration
              GlobalConfiguration().set_active_project(project)
          return project
  
      # ----------------------------- Server Settings ----------------------------
  
      def get_settings(self, hydrate: bool = True) -> ServerSettingsResponse:
          """Get the server settings.
  
          Args:
              hydrate: Flag deciding whether to hydrate the output model(s)
                  by including metadata fields in the response.
  
          Returns:
              The server settings.
          """
          return self.zen_store.get_server_settings(hydrate=hydrate)
  
      def update_server_settings(
          self,
          updated_name: Optional[str] = None,
          updated_logo_url: Optional[str] = None,
          updated_enable_analytics: Optional[bool] = None,
          updated_enable_announcements: Optional[bool] = None,
          updated_enable_updates: Optional[bool] = None,
          updated_onboarding_state: Optional[Dict[str, Any]] = None,
      ) -> ServerSettingsResponse:
          """Update the server settings.
  
          Args:
              updated_name: Updated name for the server.
              updated_logo_url: Updated logo URL for the server.
              updated_enable_analytics: Updated value whether to enable
                  analytics for the server.
              updated_enable_announcements: Updated value whether to display
                  announcements about ZenML.
              updated_enable_updates: Updated value whether to display updates
                  about ZenML.
              updated_onboarding_state: Updated onboarding state for the server.
  
          Returns:
              The updated server settings.
          """
          update_model = ServerSettingsUpdate(
              server_name=updated_name,
              logo_url=updated_logo_url,
              enable_analytics=updated_enable_analytics,
              display_announcements=updated_enable_announcements,
              display_updates=updated_enable_updates,
              onboarding_state=updated_onboarding_state,
          )
          return self.zen_store.update_server_settings(update_model)
  
      # ---------------------------------- Users ---------------------------------
  
      def create_user(
          self,
          name: str,
          password: Optional[str] = None,
          is_admin: bool = False,
      ) -> UserResponse:
          """Create a new user.
  
          Args:
              name: The name of the user.
              password: The password of the user. If not provided, the user will
                  be created with empty password.
              is_admin: Whether the user should be an admin.
  
          Returns:
              The model of the created user.
          """
          user = UserRequest(
              name=name, password=password or None, is_admin=is_admin
          )
          user.active = (
              password != "" if self.zen_store.type != StoreType.REST else True
          )
          created_user = self.zen_store.create_user(user=user)
  
          return created_user
  
      def get_user(
          self,
          name_id_or_prefix: Union[str, UUID],
          allow_name_prefix_match: bool = True,
          hydrate: bool = True,
      ) -> UserResponse:
          """Gets a user.
  
          Args:
              name_id_or_prefix: The name or ID of the user.
              allow_name_prefix_match: If True, allow matching by name prefix.
              hydrate: Flag deciding whether to hydrate the output model(s)
                  by including metadata fields in the response.
  
          Returns:
              The User
          """
          return self._get_entity_by_id_or_name_or_prefix(
              get_method=self.zen_store.get_user,
              list_method=self.list_users,
              name_id_or_prefix=name_id_or_prefix,
              allow_name_prefix_match=allow_name_prefix_match,
              hydrate=hydrate,
          )
  
      def list_users(
          self,
          sort_by: str = "created",
          page: int = PAGINATION_STARTING_PAGE,
          size: int = PAGE_SIZE_DEFAULT,
          logical_operator: LogicalOperators = LogicalOperators.AND,
          id: Optional[Union[UUID, str]] = None,
          external_user_id: Optional[str] = None,
          created: Optional[Union[datetime, str]] = None,
          updated: Optional[Union[datetime, str]] = None,
          name: Optional[str] = None,
          full_name: Optional[str] = None,
          email: Optional[str] = None,
          active: Optional[bool] = None,
          email_opted_in: Optional[bool] = None,
          hydrate: bool = False,
      ) -> Page[UserResponse]:
          """List all users.
  
          Args:
              sort_by: The column to sort by
              page: The page of items
              size: The maximum size of all pages
              logical_operator: Which logical operator to use [and, or]
              id: Use the id of stacks to filter by.
              external_user_id: Use the external user id for filtering.
              created: Use to filter by time of creation
              updated: Use the last updated date for filtering
              name: Use the username for filtering
              full_name: Use the user full name for filtering
              email: Use the user email for filtering
              active: User the user active status for filtering
              email_opted_in: Use the user opt in status for filtering
              hydrate: Flag deciding whether to hydrate the output model(s)
                  by including metadata fields in the response.
  
          Returns:
              The User
          """
          return self.zen_store.list_users(
              UserFilter(
                  sort_by=sort_by,
                  page=page,
                  size=size,
                  logical_operator=logical_operator,
                  id=id,
                  external_user_id=external_user_id,
                  created=created,
                  updated=updated,
                  name=name,
                  full_name=full_name,
                  email=email,
                  active=active,
                  email_opted_in=email_opted_in,
              ),
              hydrate=hydrate,
          )
  
      def update_user(
          self,
          name_id_or_prefix: Union[str, UUID],
          updated_name: Optional[str] = None,
          updated_full_name: Optional[str] = None,
          updated_email: Optional[str] = None,
          updated_email_opt_in: Optional[bool] = None,
          updated_password: Optional[str] = None,
          old_password: Optional[str] = None,
          updated_is_admin: Optional[bool] = None,
          updated_metadata: Optional[Dict[str, Any]] = None,
          updated_default_project_id: Optional[UUID] = None,
          active: Optional[bool] = None,
      ) -> UserResponse:
          """Update a user.
  
          Args:
              name_id_or_prefix: The name or ID of the user to update.
              updated_name: The new name of the user.
              updated_full_name: The new full name of the user.
              updated_email: The new email of the user.
              updated_email_opt_in: The new email opt-in status of the user.
              updated_password: The new password of the user.
              old_password: The old password of the user. Required for password
                  update.
              updated_is_admin: Whether the user should be an admin.
              updated_metadata: The new metadata for the user.
              updated_default_project_id: The new default project ID for the user.
              active: Use to activate or deactivate the user.
  
          Returns:
              The updated user.
  
          Raises:
              ValidationError: If the old password is not provided when updating
                  the password.
          """
          user = self.get_user(
              name_id_or_prefix=name_id_or_prefix, allow_name_prefix_match=False
          )
          user_update = UserUpdate(name=updated_name or user.name)
          if updated_full_name:
              user_update.full_name = updated_full_name
          if updated_email is not None:
              user_update.email = updated_email
              user_update.email_opted_in = (
                  updated_email_opt_in or user.email_opted_in
              )
          if updated_email_opt_in is not None:
              user_update.email_opted_in = updated_email_opt_in
          if updated_password is not None:
              user_update.password = updated_password
              if old_password is None:
                  raise ValidationError(
                      "Old password is required to update the password."
                  )
              user_update.old_password = old_password
          if updated_is_admin is not None:
              user_update.is_admin = updated_is_admin
          if active is not None:
              user_update.active = active
  
          if updated_metadata is not None:
              user_update.user_metadata = updated_metadata
  
          if updated_default_project_id is not None:
              user_update.default_project_id = updated_default_project_id
  
          return self.zen_store.update_user(
              user_id=user.id, user_update=user_update
          )
  
      @_fail_for_sql_zen_store
      def deactivate_user(self, name_id_or_prefix: str) -> "UserResponse":
          """Deactivate a user and generate an activation token.
  
          Args:
              name_id_or_prefix: The name or ID of the user to reset.
  
          Returns:
              The deactivated user.
          """
          from zenml.zen_stores.rest_zen_store import RestZenStore
  
          user = self.get_user(name_id_or_prefix, allow_name_prefix_match=False)
          assert isinstance(self.zen_store, RestZenStore)
          return self.zen_store.deactivate_user(user_name_or_id=user.name)
  
      def delete_user(self, name_id_or_prefix: str) -> None:
          """Delete a user.
  
          Args:
  