Pydantic - Google Cloud Secret

pydantic_settings_gcs.py

import logging
import os
from pathlib import Path
from typing import Any, Dict, Optional, Union, Mapping

from google.api_core.exceptions import GoogleAPIError
from google.auth.exceptions import GoogleAuthError
from google.cloud import secretmanager

from pydantic import BaseSettings, Field
from pydantic.env_settings import env_file_sentinel, read_env_file
from pydantic.utils import deep_update

LOGGER = logging.getLogger(__name__)

GCS_APP_KEY = "GOOGLE_APPLICATION_CREDENTIALS"


def get_google_cloud_secret(key) -> Optional[str]:
    if not os.getenv(GCS_APP_KEY):
        LOGGER.debug("Not fetching secret from GCS: %s not present", GCS_APP_KEY)
        return

    try:
        client = secretmanager.SecretManagerServiceClient()
        response = client.access_secret_version(key)
        return response.payload.data.decode()
    except (GoogleAuthError, GoogleAPIError) as e:
        LOGGER.error("Could not fetch (%s) from GCS: %s", key, e)


class GoogleCloudSecretSettings(BaseSettings):
    """
    Fetch setting value from Google Cloud Secret Manager.
    Local environment variables, env file, manually passed kwargs still take
    preference over values in GCS.
    Will fail silently if secret cannot be fetched for any reason
    """

    def _build_gcs_values(
            self, _env_file: Union[Path, str, None] = None, _env_file_encoding: Optional[str] = None
    ) -> Dict[str, Optional[str]]:
        if self.__config__.case_sensitive:
            env_vars: Mapping[str, Optional[str]] = os.environ
        else:
            env_vars = {k.lower(): v for k, v in os.environ.items()}

        env_file = _env_file if _env_file != env_file_sentinel else self.__config__.env_file
        env_file_encoding = _env_file_encoding if _env_file_encoding is not None else self.__config__.env_file_encoding
        if env_file is not None:
            env_path = Path(env_file).expanduser()
            if env_path.is_file():
                env_vars = {
                    **read_env_file(
                        env_path, encoding=env_file_encoding, case_sensitive=self.__config__.case_sensitive
                    ),
                    **env_vars,
                }

        env = dict()
        for field in self.__fields__.values():
            cloud_key = field.field_info.extra.get("cloud_key")
            cloud_key = cloud_key.lower() if not self.__config__.case_sensitive else cloud_key
            if not isinstance(cloud_key, str):
                continue

            if not env_vars.get(cloud_key):
                continue

            env[field.alias] = get_google_cloud_secret(env_vars[cloud_key])
        return env

    def _build_values(
            self,
            init_kwargs: Dict[str, Any],
            _env_file: Union[Path, str, None] = None,
            _env_file_encoding: Optional[str] = None,
            _secrets_dir: Union[Path, str, None] = None,
    ) -> Dict[str, Any]:
        return deep_update(
            self._build_gcs_values(_env_file, _env_file_encoding),
            self._build_secrets_files(_secrets_dir),
            self._build_environ(_env_file, _env_file_encoding),
            init_kwargs
        )


class Settings(GoogleCloudSecretSettings):
    # GCS_MY_KEY_RESOURCE_NAME=projects/<id>/secrets/<resource-name>/versions/latest
    # in environment variables or .env file
    MY_KEY: str = Field(cloud_key="GCS_MY_KEY_RESOURCE_NAME")


conf = Settings()


if __name__ == "__main__":
    print(f"MY_KEY: {conf.MY_KEY}")