content: #  Copyright (c) ZenML GmbH 2023. 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.
  """Service connector CLI commands."""
  
  from datetime import datetime
  from functools import partial
  from typing import Any, Dict, List, Optional, Union, cast
  from uuid import UUID
  
  import click
  
  from zenml.cli import utils as cli_utils
  from zenml.cli.cli import TagGroup, cli
  from zenml.cli.utils import (
      OutputFormat,
      is_sorted_or_filtered,
      list_options,
  )
  from zenml.client import Client
  from zenml.console import console
  from zenml.enums import CliCategories
  from zenml.exceptions import AuthorizationException, IllegalOperationError
  from zenml.models import (
      ServiceConnectorFilter,
      ServiceConnectorRequest,
      ServiceConnectorResourcesModel,
      ServiceConnectorResponse,
  )
  from zenml.utils.time_utils import seconds_to_human_readable, utc_now
  
  
  # Service connectors
  @cli.group(
      cls=TagGroup,
      tag=CliCategories.IDENTITY_AND_SECURITY,
  )
  def service_connector() -> None:
      """Configure and manage service connectors."""
  
  
  def prompt_connector_name(
      default_name: Optional[str] = None, connector: Optional[UUID] = None
  ) -> str:
      """Prompt the user for a service connector name.
  
      Args:
          default_name: The default name to use if the user doesn't provide one.
          connector: The UUID of a service connector being renamed.
  
      Returns:
          The name provided by the user.
      """
      client = Client()
  
      while True:
          # Ask for a name
          title = "Please enter a name for the service connector"
          if connector:
              title += " or press Enter to keep the current name"
  
          name = click.prompt(
              title,
              type=str,
              default=default_name,
          )
          if not name:
              cli_utils.warning("The name cannot be empty")
              continue
          assert isinstance(name, str)
  
          # Check if the name is taken
          try:
              existing_connector = client.get_service_connector(
                  name_id_or_prefix=name, allow_name_prefix_match=False
              )
          except KeyError:
              break
          else:
              if existing_connector.id == connector:
                  break
              cli_utils.warning(
                  f"A service connector with the name '{name}' already "
                  "exists. Please choose a different name."
              )
  
      return name
  
  
  def prompt_resource_type(available_resource_types: List[str]) -> Optional[str]:
      """Prompt the user for a resource type.
  
      Args:
          available_resource_types: The list of available resource types.
  
      Returns:
          The resource type provided by the user.
      """
      resource_type = None
      if len(available_resource_types) == 1:
          # Default to the first resource type if only one type is available
          click.echo(
              "Only one resource type is available for this connector"
              f" ({available_resource_types[0]})."
          )
          resource_type = available_resource_types[0]
      else:
          # Ask the user to select a resource type
          while True:
              resource_type = click.prompt(
                  "Please select a resource type or leave it empty to create "
                  "a connector that can be used to access any of the "
                  "supported resource types "
                  f"({', '.join(available_resource_types)}).",
                  type=str,
                  default="",
              )
              if resource_type and resource_type not in available_resource_types:
                  cli_utils.warning(
                      f"The entered resource type '{resource_type}' is not "
                      "one of the listed values. Please try again."
                  )
                  continue
              break
  
          if resource_type == "":
              resource_type = None
  
      return resource_type
  
  
  def prompt_resource_id(
      resource_name: str, resource_ids: List[str]
  ) -> Optional[str]:
      """Prompt the user for a resource ID.
  
      Args:
          resource_name: The name of the resource.
          resource_ids: The list of available resource IDs.
  
      Returns:
          The resource ID provided by the user.
      """
      resource_id: Optional[str] = None
      if resource_ids:
          resource_ids_list = "\n - " + "\n - ".join(resource_ids)
          prompt = (
              f"The following {resource_name} instances "
              "are reachable through this connector:"
              f"{resource_ids_list}\n"
              "Please select one or leave it empty to create a "
              "connector that can be used to access any of them"
          )
          while True:
              # Ask the user to enter an optional resource ID
              resource_id = click.prompt(
                  prompt,
                  default="",
                  type=str,
              )
              if (
                  not resource_ids
                  or not resource_id
                  or resource_id in resource_ids
              ):
                  break
  
              cli_utils.warning(
                  f"The selected '{resource_id}' value is not one of "
                  "the listed values. Please try again."
              )
      else:
          prompt = (
              "The connector configuration can be used to access "
              f"multiple {resource_name} instances. If you "
              "would like to limit the scope of the connector to one "
              "instance, please enter the name of a particular "
              f"{resource_name} instance. Or leave it "
              "empty to create a multi-instance connector that can "
              f"be used to access any {resource_name}"
          )
          resource_id = click.prompt(
              prompt,
              default="",
              type=str,
          )
  
      if resource_id == "":
          resource_id = None
  
      return resource_id
  
  
  def prompt_expiration_time(
      min: Optional[int] = None,
      max: Optional[int] = None,
      default: Optional[int] = None,
  ) -> int:
      """Prompt the user for an expiration time.
  
      Args:
          min: The minimum allowed expiration time.
          max: The maximum allowed expiration time.
          default: The default expiration time.
  
      Returns:
          The expiration time provided by the user.
      """
      if min is None:
          min = 0
      min_str = f"min: {min} = {seconds_to_human_readable(min)}; "
      if max is not None:
          max_str = str(max)
          max_str = f"max: {max} = {seconds_to_human_readable(max)}"
      else:
          max = -1
          max_str = "max: unlimited"
      if default:
          default_str = (
              f"; default: {default} = {seconds_to_human_readable(default)}"
          )
      else:
          default_str = ""
  
      while True:
          expiration_seconds = click.prompt(
              "The authentication method involves generating "
              "temporary credentials. Please enter the time that "
              "the credentials should be valid for, in seconds "
              f"({min_str}{max_str}{default_str})",
              type=int,
              default=default,
          )
  
          assert expiration_seconds is not None
          assert isinstance(expiration_seconds, int)
          if expiration_seconds < min:
              cli_utils.warning(
                  f"The expiration time must be at least "
                  f"{min} seconds. Please enter a larger value."
              )
              continue
          if max > 0 and expiration_seconds > max:
              cli_utils.warning(
                  f"The expiration time must not exceed "
                  f"{max} seconds. Please enter a smaller value."
              )
              continue
  
          confirm = click.confirm(
              f"Credentials will be valid for "
              f"{seconds_to_human_readable(expiration_seconds)}. Keep this "
              "value?",
              default=True,
          )
          if confirm:
              break
  
      return expiration_seconds
  
  
  def prompt_expires_at(
      default: Optional[datetime] = None,
  ) -> Optional[datetime]:
      """Prompt the user for an expiration timestamp.
  
      Args:
          default: The default expiration time.
  
      Returns:
          The expiration time provided by the user.
      """
      if default is None:
          confirm = click.confirm(
              "Are the credentials you configured temporary? If so, you'll be asked "
              "to provide an expiration time in the next step.",
              default=False,
          )
          if not confirm:
              return None
  
      while True:
          default_str = ""
          if default is not None:
              seconds = int(
                  (default - utc_now(tz_aware=default)).total_seconds()
              )
              default_str = (
                  f" [{str(default)} i.e. in "
                  f"{seconds_to_human_readable(seconds)}]"
              )
  
          expires_at = click.prompt(
              "Please enter the exact UTC date and time when the credentials "
              f"will expire e.g. '2023-12-31 23:59:59'{default_str}",
              type=click.DateTime(),
              default=default,
              show_default=False,
          )
  
          assert expires_at is not None
          assert isinstance(expires_at, datetime)
          if expires_at < utc_now(tz_aware=expires_at):
              cli_utils.warning(
                  "The expiration time must be in the future. Please enter a "
                  "later date and time."
              )
              continue
  
          seconds = int(
              (expires_at - utc_now(tz_aware=expires_at)).total_seconds()
          )
  
          confirm = click.confirm(
              f"Credentials will be valid until {str(expires_at)} UTC (i.e. "
              f"in {seconds_to_human_readable(seconds)}. Keep this value?",
              default=True,
          )
          if confirm:
              break
  
      return expires_at
  
  
  @service_connector.command(
      "register",
      context_settings={"ignore_unknown_options": True},
      help="""Configure, validate and register a service connector.
  
  This command can be used to configure and register a ZenML service connector and
  to optionally verify that the service connector configuration and credentials
  are valid and can be used to access the specified resource(s).
  
  If the `-i|--interactive` flag is set, it will prompt the user for all the
  information required to configure a service connector in a wizard-like fashion:
  
      $ zenml service-connector register -i
  
  To trim down the amount of information displayed in interactive mode, pass the
  `-n|--no-docs` flag:
  
      $ zenml service-connector register -ni
  
  Secret configuration attributes are not shown by default. Use the
  `-x|--show-secrets` flag to show them:
  
      $ zenml service-connector register -ix
  
  Non-interactive examples:
  
  - register a multi-purpose AWS service connector capable of accessing
  any of the resource types that it supports (e.g. S3 buckets, EKS Kubernetes
  clusters) using auto-configured credentials (i.e. extracted from the environment
  variables or AWS CLI configuration files):
  
      $ zenml service-connector register aws-auto-multi --description \\
  "Multi-purpose AWS connector" --type aws --auto-configure \\
  --label auto=true --label purpose=multi
  
  - register a Docker service connector providing access to a single DockerHub
  repository named `dockerhub-hyppo` using explicit credentials:
  
      $ zenml service-connector register dockerhub-hyppo --description \\
  "Hyppo's DockerHub repo" --type docker --resource-id dockerhub-hyppo \\
  --username=hyppo --password=mypassword
  
  - register an AWS service connector providing access to all the S3 buckets
  that it's authorized to access using IAM role credentials:
  
      $ zenml service-connector register aws-s3-multi --description \\   
  "Multi-bucket S3 connector" --type aws --resource-type s3-bucket \\    
  --auth_method iam-role --role_arn=arn:aws:iam::<account>:role/<role> \\
  --aws_region=us-east-1 --aws-access-key-id=<aws-key-id> \\            
  --aws_secret_access_key=<aws-secret-key> --expiration-seconds 3600
  
  All registered service connectors are validated before being registered. To
  skip validation, pass the `--no-verify` flag.
  """,
  )
  @click.argument(
      "name",
      type=str,
      required=False,
  )
  @click.option(
      "--description",
      "description",
      help="Short description for the connector instance.",
      required=False,
      type=str,
  )
  @click.option(
      "--type",
      "-t",
      "connector_type",
      help="The service connector type.",
      required=False,
      type=str,
  )
  @click.option(
      "--resource-type",
      "-r",
      "resource_type",
      help="The type of resource to connect to.",
      required=False,
      type=str,
  )
  @click.option(
      "--resource-id",
      "-ri",
      "resource_id",
      help="The ID of the resource to connect to.",
      required=False,
      type=str,
  )
  @click.option(
      "--auth-method",
      "-a",
      "auth_method",
      help="The authentication method to use.",
      required=False,
      type=str,
  )
  @click.option(
      "--expires-at",
      "expires_at",
      help="The exact UTC date and time when the credentials configured for this "
      "connector will expire. Takes the form 'YYYY-MM-DD HH:MM:SS'. This is only "
      "required if you are configuring a service connector with expiring "
      "credentials.",
      required=False,
      type=click.DateTime(),
  )
  @click.option(
      "--expires-skew-tolerance",
      "expires_skew_tolerance",
      help="The tolerance, in seconds, allowed when determining when the "
      "credentials configured for or generated by this connector will expire.",
      required=False,
      type=int,
  )
  @click.option(
      "--expiration-seconds",
      "expiration_seconds",
      help="The duration, in seconds, that the temporary credentials "
      "generated by this connector should remain valid.",
      required=False,
      type=int,
  )
  @click.option(
      "--label",
      "-l",
      "labels",
      help="Labels to be associated with the service connector. Takes the form "
      "-l key1=value1 and can be used multiple times.",
      multiple=True,
  )
  @click.option(
      "--no-verify",
      "no_verify",
      is_flag=True,
      default=False,
      help="Do not verify the service connector before registering.",
      type=click.BOOL,
  )
  @click.option(
      "--interactive",
      "-i",
      "interactive",
      is_flag=True,
      default=False,
      help="Register a new service connector interactively.",
      type=click.BOOL,
  )
  @click.option(
      "--no-docs",
      "-n",
      "no_docs",
      is_flag=True,
      default=False,
      help="Don't show documentation details during the interactive "
      "configuration.",
      type=click.BOOL,
  )
  @click.option(
      "--show-secrets",
      "-x",
      "show_secrets",
      is_flag=True,
      default=False,
      help="Show security sensitive configuration attributes in the terminal.",
      type=click.BOOL,
  )
  @click.option(
      "--auto-configure",
      "auto_configure",
      is_flag=True,
      default=False,
      help="Auto configure the service connector.",
      type=click.BOOL,
  )
  @click.argument("args", nargs=-1, type=click.UNPROCESSED)
  def register_service_connector(
      name: Optional[str],
      args: List[str],
      description: Optional[str] = None,
      connector_type: Optional[str] = None,
      resource_type: Optional[str] = None,
      resource_id: Optional[str] = None,
      auth_method: Optional[str] = None,
      expires_at: Optional[datetime] = None,
      expires_skew_tolerance: Optional[int] = None,
      expiration_seconds: Optional[int] = None,
      no_verify: bool = False,
      labels: Optional[List[str]] = None,
      interactive: bool = False,
      no_docs: bool = False,
      show_secrets: bool = False,
      auto_configure: bool = False,
  ) -> None:
      """Registers a service connector.
  
      Args:
          name: The name to use for the service connector.
          args: Configuration arguments for the service connector.
          description: Short description for the service connector.
          connector_type: The service connector type.
          resource_type: The type of resource to connect to.
          resource_id: The ID of the resource to connect to.
          auth_method: The authentication method to use.
          expires_at: The exact UTC date and time when the credentials configured
              for this connector will expire.
          expires_skew_tolerance: The tolerance, in seconds, allowed when
              determining when the credentials configured for or generated by
              this connector will expire.
          expiration_seconds: The duration, in seconds, that the temporary
              credentials generated by this connector should remain valid.
          no_verify: Do not verify the service connector before
              registering.
          labels: Labels to be associated with the service connector.
          interactive: Register a new service connector interactively.
          no_docs: Don't show documentation details during the interactive
              configuration.
          show_secrets: Show security sensitive configuration attributes in
              the terminal.
          auto_configure: Auto configure the service connector.
  
      Raises:
          TypeError: If the connector_model does not have the correct type.
      """
      from rich.markdown import Markdown
  
      client = Client()
  
      # Parse the given args
      name, parsed_args = cli_utils.parse_name_and_extra_arguments(
          list(args) + [name or ""],
          expand_args=True,
          name_mandatory=not interactive,
      )
  
      # Parse the given labels
      parsed_labels = cast(Dict[str, str], cli_utils.get_parsed_labels(labels))
  
      if interactive:
          # Get the list of available service connector types
          connector_types = client.list_service_connector_types(
              connector_type=connector_type,
              resource_type=resource_type,
              auth_method=auth_method,
          )
          if not connector_types:
              cli_utils.error(
                  "No service connectors found with the given parameters: "
                  + f"type={connector_type} "
                  if connector_type
                  else "" + f"resource_type={resource_type} "
                  if resource_type
                  else "" + f"auth_method={auth_method} "
                  if auth_method
                  else "",
              )
  
          # Ask for a connector name
          name = prompt_connector_name(name)
  
          # Ask for a description
          description = click.prompt(
              "Please enter a description for the service connector",
              type=str,
              default="",
          )
  
          available_types = {c.connector_type: c for c in connector_types}
          if len(available_types) == 1:
              # Default to the first connector type if not supplied and if
              # only one type is available
              connector_type = connector_type or list(available_types.keys())[0]
  
          # Print the name, type and description of all available service
          # connectors
          if not no_docs:
              message = "# Available service connector types\n"
              for spec in connector_types:
                  message += cli_utils.print_service_connector_type(
                      connector_type=spec,
                      heading="##",
                      footer="",
                      include_auth_methods=False,
                      include_resource_types=False,
                      print=False,
                  )
              console.print(Markdown(f"{message}---"), justify="left", width=80)
  
          # Ask the user to select a service connector type
          connector_type = click.prompt(
              "Please select a service connector type",
              type=click.Choice(list(available_types.keys())),
              default=connector_type,
          )
  
          assert connector_type is not None
          connector_type_spec = available_types[connector_type]
  
          available_resource_types = [
              t.resource_type for t in connector_type_spec.resource_types
          ]
  
          if not no_docs:
              # Print the name, resource type identifiers and description of all
              # available resource types
              message = "# Available resource types\n"
              for r in connector_type_spec.resource_types:
                  message += cli_utils.print_service_connector_resource_type(
                      resource_type=r,
                      heading="##",
                      footer="",
                      print=False,
                  )
              console.print(Markdown(f"{message}---"), justify="left", width=80)
  
          # Ask the user to select a resource type
          resource_type = prompt_resource_type(
              available_resource_types=available_resource_types
          )
  
          # Ask the user whether to use autoconfiguration, if the connector
          # implementation is locally available and if autoconfiguration is
          # supported
          if (
              connector_type_spec.supports_auto_configuration
              and connector_type_spec.local
          ):
              auto_configure = click.confirm(
                  "Would you like to attempt auto-configuration to extract the "
                  "authentication configuration from your local environment ?",
                  default=False,
              )
          else:
              auto_configure = False
  
          connector_model: Optional[
              Union[ServiceConnectorRequest, ServiceConnectorResponse]
          ] = None
          connector_resources: Optional[ServiceConnectorResourcesModel] = None
          if auto_configure:
              # Try to autoconfigure the service connector
              try:
                  with console.status("Auto-configuring service connector...\n"):
                      (
                          connector_model,
                          connector_resources,
                      ) = client.create_service_connector(
                          name=name,
                          description=description or "",
                          connector_type=connector_type,
                          resource_type=resource_type,
                          auth_method=auth_method,
                          expires_skew_tolerance=expires_skew_tolerance,
                          auto_configure=True,
                          verify=True,
                          register=False,
                      )
  
                  assert connector_model is not None
                  assert connector_resources is not None
              except (
                  KeyError,
                  ValueError,
                  IllegalOperationError,
                  NotImplementedError,
                  AuthorizationException,
              ) as e:
                  cli_utils.warning(
                      f"Auto-configuration was not successful: {e} "
                  )
                  # Ask the user whether to continue with manual configuration
                  manual = click.confirm(
                      "Would you like to continue with manual configuration ?",
                      default=True,
                  )
                  if not manual:
                      return
              else:
                  auth_method = connector_model.auth_method
                  expiration_seconds = connector_model.expiration_seconds
                  expires_at = connector_model.expires_at
                  cli_utils.declare(
                      "Service connector auto-configured successfully with the "
                      "following configuration:"
                  )
  
                  # Print the configuration detected by the autoconfiguration
                  # process
                  # TODO: Normally, this could have been handled with setter
                  #   functions over the connector type property in the response
                  #   model. However, pydantic breaks property setter functions.
                  #   We can find a more elegant solution here.
                  if isinstance(connector_model, ServiceConnectorResponse):
                      connector_model.set_connector_type(connector_type_spec)
                  elif isinstance(connector_model, ServiceConnectorRequest):
                      connector_model.connector_type = connector_type_spec
                  else:
                      raise TypeError(
                          "The service connector must be an instance of either"
                          "`ServiceConnectorResponse` or "
                          "`ServiceConnectorRequest`."
                      )
  
                  cli_utils.print_service_connector_configuration(
                      connector_model,
                      active_status=False,
                      show_secrets=show_secrets,
                  )
                  cli_utils.declare(
                      "The service connector configuration has access to the "
                      "following resources:"
                  )
                  cli_utils.print_service_connector_resource_table(
                      [connector_resources],
                      show_resources_only=True,
                  )
  
                  # Ask the user whether to continue with the autoconfiguration
                  choice = click.prompt(
                      "Would you like to continue with the auto-discovered "
                      "configuration or switch to manual ?",
                      type=click.Choice(["auto", "manual"]),
                      default="auto",
                  )
                  if choice == "manual":
                      # Reset the connector configuration to default to let the
                      # manual configuration kick in the next step
                      connector_model = None
                      connector_resources = None
                      expires_at = None
  
          if connector_model is not None and connector_resources is not None:
              assert auth_method is not None
              auth_method_spec = connector_type_spec.auth_method_dict[
                  auth_method
              ]
          else:
              # In this branch, we are either not using autoconfiguration or the
              # autoconfiguration failed or was dismissed. In all cases, we need
              # to ask the user for the authentication method to use and then
              # prompt for the configuration
  
              auth_methods = list(connector_type_spec.auth_method_dict.keys())
  
              if not no_docs:
                  # Print the name, identifier and description of all available
                  # auth methods
                  message = "# Available authentication methods\n"
                  for a in auth_methods:
                      message += cli_utils.print_service_connector_auth_method(
                          auth_method=connector_type_spec.auth_method_dict[a],
                          heading="##",
                          footer="",
                          print=False,
                      )
                  console.print(
                      Markdown(f"{message}---"), justify="left", width=80
                  )
  
              if len(auth_methods) == 1:
                  # Default to the first auth method if only one method is
                  # available
                  confirm = click.confirm(
                      "Only one authentication method is available for this "
                      f"connector ({auth_methods[0]}). Would you like to use it?",
                      default=True,
                  )
                  if not confirm:
                      return
  
                  auth_method = auth_methods[0]
              else:
                  # Ask the user to select an authentication method
                  auth_method = click.prompt(
                      "Please select an authentication method",
                      type=click.Choice(auth_methods),
                      default=auth_method,
                  )
  
              assert auth_method is not None
              auth_method_spec = connector_type_spec.auth_method_dict[
                  auth_method
              ]
  
              cli_utils.declare(
                  f"Please enter the configuration for the {auth_method_spec.name} "
                  "authentication method."
              )
  
              # Prompt for the configuration of the selected authentication method
              # field by field
              config_schema = auth_method_spec.config_schema or {}
              config_dict = cli_utils.prompt_configuration(
                  config_schema=config_schema,
                  show_secrets=show_secrets,
              )
  
              # Prompt for an expiration time if the auth method supports it
              if auth_method_spec.supports_temporary_credentials():
                  expiration_seconds = prompt_expiration_time(
                      min=auth_method_spec.min_expiration_seconds,
                      max=auth_method_spec.max_expiration_seconds,
                      default=auth_method_spec.default_expiration_seconds,
                  )
  
              # Prompt for the time when the credentials will expire
              expires_at = prompt_expires_at(expires_at)
  
              try:
                  # Validate the connector configuration and fetch all available
                  # resources that are accessible with the provided configuration
                  # in the process
                  with console.status(
                      "Validating service connector configuration...\n"
                  ):
                      (
                          connector_model,
                          connector_resources,
                      ) = client.create_service_connector(
                          name=name,
                          description=description or "",
                          connector_type=connector_type,
                          auth_method=auth_method,
                          resource_type=resource_type,
                          configuration=config_dict,
                          expires_at=expires_at,
                          expires_skew_tolerance=expires_skew_tolerance,
                          expiration_seconds=expiration_seconds,
                          auto_configure=False,
                          verify=True,
                          register=False,
                      )
                  assert connector_model is not None
                  assert connector_resources is not None
              except (
                  KeyError,
                  ValueError,
                  IllegalOperationError,
                  NotImplementedError,
                  AuthorizationException,
              ) as e:
                  cli_utils.error(f"Failed to configure service connector: {e}")
  
          if resource_type:
              # Finally, for connectors that are configured with a particular
              # resource type, prompt the user to select one of the available
              # resources that can be accessed with the connector. We don't do
              # need to do this for resource types that don't support instances.
              resource_type_spec = connector_type_spec.resource_type_dict[
                  resource_type
              ]
              if resource_type_spec.supports_instances:
                  assert len(connector_resources.resources) == 1
                  resource_ids = connector_resources.resources[0].resource_ids
                  assert resource_ids is not None
                  resource_id = prompt_resource_id(
                      resource_name=resource_type_spec.name,
                      resource_ids=resource_ids,
                  )
              else:
                  resource_id = None
          else:
              resource_id = None
  
          # Prepare the rest of the variables to fall through to the
          # non-interactive configuration case
          parsed_args = connector_model.configuration.plain
          auto_configure = False
          no_verify = False
          expiration_seconds = connector_model.expiration_seconds
  
      if not connector_type:
          cli_utils.error(
              "The connector type must be specified when using non-interactive "
              "configuration."
          )
  
      with console.status(f"Registering service connector '{name}'...\n"):
          try:
              # Create a new service connector
              assert name is not None
              (
                  connector_model,
                  connector_resources,
              ) = client.create_service_connector(
                  name=name,
                  connector_type=connector_type,
                  auth_method=auth_method,
                  resource_type=resource_type,
                  configuration=parsed_args,
                  resource_id=resource_id,
  