Overview

This repository shows how to extend the PydanticBaseSettingsSource class to allow you to get your settings from Google Secret Manager.

I have created a YouTube video that walks through the code in this repository. You can find it here: Settings management with Pydantic and Google Secret Manager

How to use this repo

The application located in the pydantic_google_secrets folder can be run in the following ways:

1. From the command line with python ./pydantic_google_secrets/app.py
2. Using Docker, made easy with the included Makefile.
   1. make build to build the image.
   2. make run-app-env-file to run a container with the environment variables configured using the .env file.
   3. make run-app-env-var to run a container with the environment variables configured as part of the docker run command.
   4. make run-app-adc to run a container using Application Default Credentials (ADC) to authenticate to Google Cloud.

Code that is the focus of this repository

The code that is the focus of this repository is in the pydantic_google_secrets/config.py file. It is the GoogleSecretManagerConfigSettingsSource class that extends the PydanticBaseSettingsSource class and is subsequently used in our Settings class.

class GoogleSecretManagerConfigSettingsSource(PydanticBaseSettingsSource):
    """
    A settings class that loads settings from Google Secret Manager.

    The account under which the application is executed should have the
    required access to Google Secret Manager.
    """

    def __init__(self, settings_cls: Type[BaseSettings]):
        super().__init__(settings_cls)

        self._client = None
        self._project_id = None

    def _get_gsm_value(self, field_name: str) -> Optional[str]:
        """
        Make the call to the Google Secret Manager API to get the value of the
        secret.
        """
        secret_name = self._client.secret_version_path(
            project=self._project_id, secret=field_name, secret_version="latest"
        )

        response = self._client.access_secret_version(name=secret_name)
        return response.payload.data.decode("UTF-8")

    def get_field_value(
        self, field: FieldInfo, field_name: str
    ) -> Tuple[Any, str, bool]:
        """
        Get the value of a field from Google Secret Manager.
        """
        try:
            field_name = field.alias or field_name
            field_value = self._get_gsm_value(field_name)
        except (NotFound, PermissionDenied) as e:
            logger.debug(e)
            field_value = None

        return field_value, field_name, False

    def __call__(self) -> Dict[str, Any]:
        d: Dict[str, Any] = {}

        try:
            # Set the credentials and project ID from the application default
            # credentials
            _credentials, project_id = gc_auth.default()
            self._project_id = project_id
            self._client = secretmanager.SecretManagerServiceClient(
                credentials=_credentials
            )

            for field_name, field in self.settings_cls.model_fields.items():
                field_value, field_key, value_is_complex = self.get_field_value(
                    field, field_name
                )
                field_value = self.prepare_field_value(
                    field_name, field, field_value, value_is_complex
                )
                if field_value is not None:
                    d[field_key] = field_value

        except DefaultCredentialsError as e:
            logger.debug(e)

        return d