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.
  """Models representing service connectors."""
  
  import json
  from datetime import datetime
  from typing import Any, ClassVar, Dict, List, Optional, Union
  
  from pydantic import (
      Field,
      GetCoreSchemaHandler,
      SecretStr,
      ValidationError,
      model_validator,
  )
  from pydantic_core import CoreSchema, core_schema
  
  from zenml.constants import STR_FIELD_MAX_LENGTH
  from zenml.logger import get_logger
  from zenml.models.v2.base.base import BaseUpdate
  from zenml.models.v2.base.scoped import (
      UserScopedFilter,
      UserScopedRequest,
      UserScopedResponse,
      UserScopedResponseBody,
      UserScopedResponseMetadata,
      UserScopedResponseResources,
  )
  from zenml.models.v2.misc.service_connector_type import (
      ServiceConnectorTypeModel,
  )
  from zenml.utils.secret_utils import PlainSerializedSecretStr
  
  logger = get_logger(__name__)
  
  # ------------------ Configuration Model ------------------
  
  
  class ServiceConnectorConfiguration(Dict[str, Any]):
      """Model for service connector configuration."""
  
      @classmethod
      def from_dict(
          cls, data: Dict[str, Any]
      ) -> "ServiceConnectorConfiguration":
          """Create a configuration model from a dictionary.
  
          Args:
              data: The dictionary to create the configuration model from.
  
          Returns:
              A configuration model.
          """
          return cls(**data)
  
      @property
      def secrets(self) -> Dict[str, PlainSerializedSecretStr]:
          """Get the secrets from the configuration.
  
          Returns:
              A dictionary of secrets.
          """
          return {k: v for k, v in self.items() if isinstance(v, SecretStr)}
  
      @property
      def plain_secrets(self) -> Dict[str, str]:
          """Get the plain secrets from the configuration.
  
          Returns:
              A dictionary of secrets.
          """
          return {
              k: v.get_secret_value()
              for k, v in self.items()
              if isinstance(v, SecretStr)
          }
  
      @property
      def non_secrets(self) -> Dict[str, Any]:
          """Get the non-secrets from the configuration.
  
          Returns:
              A dictionary of non-secrets.
          """
          return {k: v for k, v in self.items() if not isinstance(v, SecretStr)}
  
      @property
      def plain(self) -> Dict[str, Any]:
          """Get the configuration with secrets unpacked.
  
          Returns:
              A dictionary of configuration with secrets unpacked.
          """
          return {
              k: v.get_secret_value() if isinstance(v, SecretStr) else v
              for k, v in self.items()
          }
  
      def get_plain(self, key: str, default: Any = None) -> Any:
          """Get the plain value for the given key.
  
          Args:
              key: The key to get the value for.
              default: The default value to return if the key is not found.
  
          Returns:
              The plain value for the given key.
          """
          result = self.get(key, default)
          if isinstance(result, SecretStr):
              return result.get_secret_value()
          return result
  
      def add_secrets(self, secrets: Dict[str, str]) -> None:
          """Add the secrets to the configuration.
  
          Args:
              secrets: The secrets to add to the configuration.
          """
          self.update({k: SecretStr(v) for k, v in secrets.items()})
  
      @classmethod
      def __get_pydantic_core_schema__(
          cls, source_type: Any, handler: GetCoreSchemaHandler
      ) -> CoreSchema:
          """Additional method for pydantic to recognize it as a valid type.
  
          Args:
              source_type: the source type
              handler: the handler
  
          Returns:
              the schema for the custom type.
          """
          return core_schema.no_info_after_validator_function(
              cls,
              handler(
                  core_schema.dict_schema(
                      keys_schema=core_schema.str_schema(),
                      values_schema=core_schema.any_schema(),
                  )
              ),
              serialization=core_schema.plain_serializer_function_ser_schema(
                  lambda v: v.plain,
                  when_used="json",
              ),
          )
  
  
  # ------------------ Request Model ------------------
  
  
  class ServiceConnectorRequest(UserScopedRequest):
      """Request model for service connectors."""
  
      name: str = Field(
          title="The service connector name.",
          max_length=STR_FIELD_MAX_LENGTH,
      )
      connector_type: Union[str, "ServiceConnectorTypeModel"] = Field(
          title="The type of service connector.",
          union_mode="left_to_right",
      )
      description: str = Field(
          default="",
          title="The service connector instance description.",
      )
      auth_method: str = Field(
          title="The authentication method that the connector instance uses to "
          "access the resources.",
          max_length=STR_FIELD_MAX_LENGTH,
      )
      resource_types: List[str] = Field(
          default_factory=list,
          title="The type(s) of resource that the connector instance can be used "
          "to gain access to.",
      )
      resource_id: Optional[str] = Field(
          default=None,
          title="Uniquely identifies a specific resource instance that the "
          "connector instance can be used to access. If omitted, the connector "
          "instance can be used to access any and all resource instances that "
          "the authentication method and resource type(s) allow.",
          max_length=STR_FIELD_MAX_LENGTH,
      )
      supports_instances: bool = Field(
          default=False,
          title="Indicates whether the connector instance can be used to access "
          "multiple instances of the configured resource type.",
      )
      expires_at: Optional[datetime] = Field(
          default=None,
          title="Time when the authentication credentials configured for the "
          "connector expire. If omitted, the credentials do not expire.",
      )
      expires_skew_tolerance: Optional[int] = Field(
          default=None,
          title="The number of seconds of tolerance to apply when checking "
          "whether the authentication credentials configured for the connector "
          "have expired. If omitted, no tolerance is applied.",
      )
      expiration_seconds: Optional[int] = Field(
          default=None,
          title="The duration, in seconds, that the temporary credentials "
          "generated by this connector should remain valid. Only applicable for "
          "connectors and authentication methods that involve generating "
          "temporary credentials from the ones configured in the connector.",
      )
      configuration: ServiceConnectorConfiguration = Field(
          default_factory=ServiceConnectorConfiguration,
          title="The service connector configuration.",
      )
      labels: Dict[str, str] = Field(
          default_factory=dict,
          title="Service connector labels.",
      )
  
      # Analytics
      ANALYTICS_FIELDS: ClassVar[List[str]] = [
          "connector_type",
          "auth_method",
          "resource_types",
      ]
  
      def get_analytics_metadata(self) -> Dict[str, Any]:
          """Format the resource types in the analytics metadata.
  
          Returns:
              Dict of analytics metadata.
          """
          metadata = super().get_analytics_metadata()
          if len(self.resource_types) == 1:
              metadata["resource_types"] = self.resource_types[0]
          else:
              metadata["resource_types"] = ", ".join(self.resource_types)
          metadata["connector_type"] = self.type
          return metadata
  
      # Helper methods
      @property
      def type(self) -> str:
          """Get the connector type.
  
          Returns:
              The connector type.
          """
          if isinstance(self.connector_type, str):
              return self.connector_type
          return self.connector_type.connector_type
  
      @property
      def emojified_connector_type(self) -> str:
          """Get the emojified connector type.
  
          Returns:
              The emojified connector type.
          """
          if not isinstance(self.connector_type, str):
              return self.connector_type.emojified_connector_type
  
          return self.connector_type
  
      @property
      def emojified_resource_types(self) -> List[str]:
          """Get the emojified connector type.
  
          Returns:
              The emojified connector type.
          """
          if not isinstance(self.connector_type, str):
              return [
                  self.connector_type.resource_type_dict[
                      resource_type
                  ].emojified_resource_type
                  for resource_type in self.resource_types
              ]
  
          return self.resource_types
  
      def validate_and_configure_resources(
          self,
          connector_type: "ServiceConnectorTypeModel",
          resource_types: Optional[Union[str, List[str]]] = None,
          resource_id: Optional[str] = None,
          configuration: Optional[Dict[str, Any]] = None,
      ) -> None:
          """Validate and configure the resources that the connector can be used to access.
  
          Args:
              connector_type: The connector type specification used to validate
                  the connector configuration.
              resource_types: The type(s) of resource that the connector instance
                  can be used to access. If omitted, a multi-type connector is
                  configured.
              resource_id: Uniquely identifies a specific resource instance that
                  the connector instance can be used to access.
              configuration: The connector configuration.
          """
          _validate_and_configure_resources(
              connector=self,
              connector_type=connector_type,
              resource_types=resource_types,
              resource_id=resource_id,
              configuration=configuration,
          )
  
  
  # ------------------ Update Model ------------------
  
  
  class ServiceConnectorUpdate(BaseUpdate):
      """Model used for service connector updates.
  
      Most fields in the update model are optional and will not be updated if
      omitted. However, the following fields are "special" and leaving them out
      will also cause the corresponding value to be removed from the service
      connector in the database:
  
      * the `resource_id` field
      * the `expiration_seconds` field
  
      In addition to the above exceptions, the following rules apply:
  
      * the `configuration` field represents a full valid configuration update,
      not just a partial update. If it is set (i.e. not None) in the update,
      its values will replace the existing configuration values.
      * the `labels` field is also a full labels update: if set (i.e. not
      `None`), all existing labels are removed and replaced by the new labels
      in the update.
  
      NOTE: the attributes here override the ones in the base class, so they
      have a None default value.
      """
  
      name: Optional[str] = Field(
          title="The service connector name.",
          max_length=STR_FIELD_MAX_LENGTH,
          default=None,
      )
      connector_type: Optional[Union[str, "ServiceConnectorTypeModel"]] = Field(
          title="The type of service connector.",
          default=None,
          union_mode="left_to_right",
      )
      description: Optional[str] = Field(
          title="The service connector instance description.",
          default=None,
      )
      auth_method: Optional[str] = Field(
          title="The authentication method that the connector instance uses to "
          "access the resources.",
          max_length=STR_FIELD_MAX_LENGTH,
          default=None,
      )
      resource_types: Optional[List[str]] = Field(
          title="The type(s) of resource that the connector instance can be used "
          "to gain access to.",
          default=None,
      )
      resource_id: Optional[str] = Field(
          title="Uniquely identifies a specific resource instance that the "
          "connector instance can be used to access. If omitted, the "
          "connector instance can be used to access any and all resource "
          "instances that the authentication method and resource type(s) "
          "allow.",
          max_length=STR_FIELD_MAX_LENGTH,
          default=None,
      )
      supports_instances: Optional[bool] = Field(
          title="Indicates whether the connector instance can be used to access "
          "multiple instances of the configured resource type.",
          default=None,
      )
      expires_at: Optional[datetime] = Field(
          title="Time when the authentication credentials configured for the "
          "connector expire. If omitted, the credentials do not expire.",
          default=None,
      )
      expires_skew_tolerance: Optional[int] = Field(
          title="The number of seconds of tolerance to apply when checking "
          "whether the authentication credentials configured for the "
          "connector have expired. If omitted, no tolerance is applied.",
          default=None,
      )
      expiration_seconds: Optional[int] = Field(
          title="The duration, in seconds, that the temporary credentials "
          "generated by this connector should remain valid. Only "
          "applicable for connectors and authentication methods that "
          "involve generating temporary credentials from the ones "
          "configured in the connector.",
          default=None,
      )
      configuration: Optional[ServiceConnectorConfiguration] = Field(
          title="The service connector full configuration replacement.",
          default=None,
      )
      labels: Optional[Dict[str, str]] = Field(
          title="Service connector labels.",
          default=None,
      )
  
      # Analytics
      ANALYTICS_FIELDS: ClassVar[List[str]] = [
          "connector_type",
          "auth_method",
          "resource_types",
      ]
  
      def get_analytics_metadata(self) -> Dict[str, Any]:
          """Format the resource types in the analytics metadata.
  
          Returns:
              Dict of analytics metadata.
          """
          metadata = super().get_analytics_metadata()
  
          if self.resource_types is not None:
              if len(self.resource_types) == 1:
                  metadata["resource_types"] = self.resource_types[0]
              else:
                  metadata["resource_types"] = ", ".join(self.resource_types)
  
          if self.connector_type is not None:
              metadata["connector_type"] = self.type
  
          return metadata
  
      # Helper methods
      @property
      def type(self) -> Optional[str]:
          """Get the connector type.
  
          Returns:
              The connector type.
          """
          if self.connector_type is not None:
              if isinstance(self.connector_type, str):
                  return self.connector_type
              return self.connector_type.connector_type
          return None
  
      def validate_and_configure_resources(
          self,
          connector_type: "ServiceConnectorTypeModel",
          resource_types: Optional[Union[str, List[str]]] = None,
          resource_id: Optional[str] = None,
          configuration: Optional[Dict[str, Any]] = None,
      ) -> None:
          """Validate and configure the resources that the connector can be used to access.
  
          Args:
              connector_type: The connector type specification used to validate
                  the connector configuration.
              resource_types: The type(s) of resource that the connector instance
                  can be used to access. If omitted, a multi-type connector is
                  configured.
              resource_id: Uniquely identifies a specific resource instance that
                  the connector instance can be used to access.
              configuration: The connector configuration.
          """
          _validate_and_configure_resources(
              connector=self,
              connector_type=connector_type,
              resource_types=resource_types,
              resource_id=resource_id,
              configuration=configuration,
          )
  
      def convert_to_request(self) -> "ServiceConnectorRequest":
          """Method to generate a service connector request object from self.
  
          For certain operations, the service connector update model need to
          adhere to the limitations set by the request model. In order to use
          update models in such situations, we need to be able to convert an
          update model into a request model.
  
          Returns:
              The equivalent request model
  
          Raises:
              RuntimeError: if the model can not be converted to a request model.
          """
          try:
              return ServiceConnectorRequest.model_validate(self.model_dump())
          except ValidationError as e:
              raise RuntimeError(
                  "The service connector update model can not be converted into "
                  f"an equivalent request model: {e}"
              )
  
  
  # ------------------ Response Model ------------------
  
  
  class ServiceConnectorResponseBody(UserScopedResponseBody):
      """Response body for service connectors."""
  
      description: str = Field(
          default="",
          title="The service connector instance description.",
      )
      connector_type: Union[str, "ServiceConnectorTypeModel"] = Field(
          title="The type of service connector.", union_mode="left_to_right"
      )
      auth_method: str = Field(
          title="The authentication method that the connector instance uses to "
          "access the resources.",
          max_length=STR_FIELD_MAX_LENGTH,
      )
      resource_types: List[str] = Field(
          default_factory=list,
          title="The type(s) of resource that the connector instance can be used "
          "to gain access to.",
      )
      resource_id: Optional[str] = Field(
          default=None,
          title="Uniquely identifies a specific resource instance that the "
          "connector instance can be used to access. If omitted, the connector "
          "instance can be used to access any and all resource instances that "
          "the authentication method and resource type(s) allow.",
          max_length=STR_FIELD_MAX_LENGTH,
      )
      supports_instances: bool = Field(
          default=False,
          title="Indicates whether the connector instance can be used to access "
          "multiple instances of the configured resource type.",
      )
      expires_at: Optional[datetime] = Field(
          default=None,
          title="Time when the authentication credentials configured for the "
          "connector expire. If omitted, the credentials do not expire.",
      )
      expires_skew_tolerance: Optional[int] = Field(
          default=None,
          title="The number of seconds of tolerance to apply when checking "
          "whether the authentication credentials configured for the connector "
          "have expired. If omitted, no tolerance is applied.",
      )
  
  
  class ServiceConnectorResponseMetadata(UserScopedResponseMetadata):
      """Response metadata for service connectors."""
  
      configuration: ServiceConnectorConfiguration = Field(
          default_factory=ServiceConnectorConfiguration,
          title="The service connector configuration.",
      )
      expiration_seconds: Optional[int] = Field(
          default=None,
          title="The duration, in seconds, that the temporary credentials "
          "generated by this connector should remain valid. Only applicable for "
          "connectors and authentication methods that involve generating "
          "temporary credentials from the ones configured in the connector.",
      )
      labels: Dict[str, str] = Field(
          default_factory=dict,
          title="Service connector labels.",
      )
  
  
  class ServiceConnectorResponseResources(UserScopedResponseResources):
      """Class for all resource models associated with the service connector entity."""
  
  
  class ServiceConnectorResponse(
      UserScopedResponse[
          ServiceConnectorResponseBody,
          ServiceConnectorResponseMetadata,
          ServiceConnectorResponseResources,
      ]
  ):
      """Response model for service connectors."""
  
      # Disable the warning for updating responses, because we update the
      # service connector type in place
      _warn_on_response_updates: bool = False
  
      name: str = Field(
          title="The service connector name.",
          max_length=STR_FIELD_MAX_LENGTH,
      )
  
      def get_analytics_metadata(self) -> Dict[str, Any]:
          """Add the service connector labels to analytics metadata.
  
          Returns:
              Dict of analytics metadata.
          """
          metadata = super().get_analytics_metadata()
  
          metadata.update(
              {
                  label[6:]: value
                  for label, value in self.labels.items()
                  if label.startswith("zenml:")
              }
          )
          return metadata
  
      def get_hydrated_version(self) -> "ServiceConnectorResponse":
          """Get the hydrated version of this service connector.
  
          Returns:
              an instance of the same entity with the metadata field attached.
          """
          from zenml.client import Client
  
          return Client().zen_store.get_service_connector(self.id)
  
      # Helper methods
      @property
      def type(self) -> str:
          """Get the connector type.
  
          Returns:
              The connector type.
          """
          if isinstance(self.connector_type, str):
              return self.connector_type
          return self.connector_type.connector_type
  
      @property
      def emojified_connector_type(self) -> str:
          """Get the emojified connector type.
  
          Returns:
              The emojified connector type.
          """
          if not isinstance(self.connector_type, str):
              return self.connector_type.emojified_connector_type
  
          return self.connector_type
  
      @property
      def emojified_resource_types(self) -> List[str]:
          """Get the emojified connector type.
  
          Returns:
              The emojified connector type.
          """
          if not isinstance(self.connector_type, str):
              return [
                  self.connector_type.resource_type_dict[
                      resource_type
                  ].emojified_resource_type
                  for resource_type in self.resource_types
              ]
  
          return self.resource_types
  
      @property
      def is_multi_type(self) -> bool:
          """Checks if the connector is multi-type.
  
          A multi-type connector can be used to access multiple types of
          resources.
  
          Returns:
              True if the connector is multi-type, False otherwise.
          """
          return len(self.resource_types) > 1
  
      @property
      def is_multi_instance(self) -> bool:
          """Checks if the connector is multi-instance.
  
          A multi-instance connector is configured to access multiple instances
          of the configured resource type.
  
          Returns:
              True if the connector is multi-instance, False otherwise.
          """
          return (
              not self.is_multi_type
              and self.supports_instances
              and not self.resource_id
          )
  
      @property
      def is_single_instance(self) -> bool:
          """Checks if the connector is single-instance.
  
          A single-instance connector is configured to access only a single
          instance of the configured resource type or does not support multiple
          resource instances.
  
          Returns:
              True if the connector is single-instance, False otherwise.
          """
          return not self.is_multi_type and not self.is_multi_instance
  
      def set_connector_type(
          self, value: Union[str, "ServiceConnectorTypeModel"]
      ) -> None:
          """Auxiliary method to set the connector type.
  
          Args:
              value: the new value for the connector type.
          """
          self.get_body().connector_type = value
  
      def validate_configuration(self) -> None:
          """Validate the configuration of the connector."""
          if isinstance(self.connector_type, ServiceConnectorTypeModel):
              self.validate_and_configure_resources(
                  connector_type=self.connector_type,
                  resource_types=self.resource_types,
                  resource_id=self.resource_id,
                  configuration=self.configuration,
              )
  
      def validate_and_configure_resources(
          self,
          connector_type: "ServiceConnectorTypeModel",
          resource_types: Optional[Union[str, List[str]]] = None,
          resource_id: Optional[str] = None,
          configuration: Optional[Dict[str, Any]] = None,
      ) -> None:
          """Validate and configure the resources that the connector can be used to access.
  
          Args:
              connector_type: The connector type specification used to validate
                  the connector configuration.
              resource_types: The type(s) of resource that the connector instance
                  can be used to access. If omitted, a multi-type connector is
                  configured.
              resource_id: Uniquely identifies a specific resource instance that
                  the connector instance can be used to access.
              configuration: The connector configuration.
          """
          _validate_and_configure_resources(
              connector=self,
              connector_type=connector_type,
              resource_types=resource_types,
              resource_id=resource_id,
              configuration=configuration,
          )
  
      # Body and metadata properties
      @property
      def description(self) -> str:
          """The `description` property.
  
          Returns:
              the value of the property.
          """
          return self.get_body().description
  
      @property
      def connector_type(self) -> Union[str, "ServiceConnectorTypeModel"]:
          """The `connector_type` property.
  
          Returns:
              the value of the property.
          """
          return self.get_body().connector_type
  
      @property
      def auth_method(self) -> str:
          """The `auth_method` property.
  
          Returns:
              the value of the property.
          """
          return self.get_body().auth_method
  
      @property
      def resource_types(self) -> List[str]:
          """The `resource_types` property.
  
          Returns:
              the value of the property.
          """
          return self.get_body().resource_types
  
      @property
      def resource_id(self) -> Optional[str]:
          """The `resource_id` property.
  
          Returns:
              the value of the property.
          """
          return self.get_body().resource_id
  
      @property
      def supports_instances(self) -> bool:
          """The `supports_instances` property.
  
          Returns:
              the value of the property.
          """
          return self.get_body().supports_instances
  
      @property
      def expires_at(self) -> Optional[datetime]:
          """The `expires_at` property.
  
          Returns:
              the value of the property.
          """
          return self.get_body().expires_at
  
      @property
      def expires_skew_tolerance(self) -> Optional[int]:
          """The `expires_skew_tolerance` property.
  
          Returns:
              the value of the property.
          """
          return self.get_body().expires_skew_tolerance
  
      @property
      def configuration(self) -> ServiceConnectorConfiguration:
          """The `configuration` property.
  
          Returns:
              the value of the property.
          """
          return self.get_metadata().configuration
  
      def remove_secrets(self) -> None:
          """Remove the secrets from the configuration."""
          metadata = self.get_metadata()
          metadata.configuration = ServiceConnectorConfiguration(
              **metadata.configuration.non_secrets
          )
  
      def add_secrets(self, secrets: Dict[str, str]) -> None:
          """Add the secrets to the configuration.
  
          Args:
              secrets: The secrets to add to the configuration.
          """
          self.get_metadata().configuration.add_secrets(secrets)
  
      @property
      def expiration_seconds(self) -> Optional[int]:
          """The `expiration_seconds` property.
  
          Returns:
              the value of the property.
          """
          return self.get_metadata().expiration_seconds
  
      @property
      def labels(self) -> Dict[str, str]:
          """The `labels` property.
  
          Returns:
              the value of the property.
          """
          return self.get_metadata().labels
  
  
  # ------------------ Filter Model ------------------
  
  
  class ServiceConnectorFilter(UserScopedFilter):
      """Model to enable advanced filtering of service connectors."""
  
      FILTER_EXCLUDE_FIELDS: ClassVar[List[str]] = [
          *UserScopedFilter.FILTER_EXCLUDE_FIELDS,
          "resource_type",
          "labels_str",
          "labels",
      ]
      CLI_EXCLUDE_FIELDS: ClassVar[List[str]] = [
          *UserScopedFilter.CLI_EXCLUDE_FIELDS,
          "labels_str",
          "labels",
      ]
      name: Optional[str] = Field(
          default=None,
          description="The name to filter by",
      )
      connector_type: Optional[str] = Field(
          default=None,
          description="The type of service connector to filter by",
      )
      auth_method: Optional[str] = Field(
          default=None,
          title="Filter by the authentication method configured for the "
          "connector",
      )
      resource_type: Optional[str] = Field(
          default=None,
          title="Filter by the type of resource that the connector can be used "
          "to access",
      )
      resource_id: Optional[str] = Field(
          default=None,
          title="Filter by the ID of the resource instance that the connector "
          "is configured to access",
      )
      labels_str: Optional[str] = Field(
          default=None,
          title="Filter by one or more labels. This field can be either a JSON "
          "formatted dictionary of label names and values, where the values are "
          'optional and can be set to None (e.g. `{"label1":"value1", "label2": '
          "null}` ), or a comma-separated list of label names and values (e.g "
          "`label1=value1,label2=`. If a label name is specified without a "
          "value, the filter will match all service connectors that have that "
          "label present, regardless of value.",
      )
  
      # Use this internally to configure and access the labels as a dictionary
      labels: Optional[Dict[str, Optional[str]]] = Field(
          default=None,
          title="The labels to filter by, as a dictionary",
          exclude=True,
      )
  
      @model_validator(mode="after")
      def validate_labels(self) -> "ServiceConnectorFilter":
          """Parse the labels string into a label dictionary and vice-versa.
  
          Returns:
              The validated values.
          """
          if self.labels_str is not None:
              try:
                  self.labels = json.loads(self.labels_str)
              except json.JSONDecodeError:
                  # Interpret as comma-separated values instead
                  self.labels = {
                      label.split("=", 1)[0]: label.split("=", 1)[1]
                      if "=" in label
                      else None
                      for label in self.labels_str.split(",")
                  }
          elif self.labels is not None:
              self.labels_str = json.dumps(self.labels)
  
          return self
  
  
  # ------------------ Helper Functions ------------------
  
  
  def _validate_and_configure_resources(
      connector: Union[
          ServiceConnectorRequest,
          ServiceConnectorUpdate,
          ServiceConnectorResponse,
      ],
      connector_type: "ServiceConnectorTypeModel",
      resource_types: Optional[Union[str, List[str]]] = None,
      resource_id: Optional[str] = None,
      configuration: Optional[Dict[str, Any]] = None,
  ) -> None:
      """Validate and configure the resources that a connector can be used to access.
  
      Args:
          connector: The connector model to validate and configure.
          connector_type: The connector type specification used to validate
              the connector configuration.
          resource_types: The type(s) of resource that the connector instance
              can be used to access. If omitted, a multi-type connector is
              configured.
          resource_id: Uniquely identifies a specific resource instance that
              the connector instance can be used to access.
          configuration: The connector configuration.
  
      Raises:
  