CODEWITHSUNDEEP
pythonfastapipydanticaws-parameter-storepydantic-v2
LearnToGrow
LearnToGrow

Reputation: 1778

pydantic-settings custom_resources does not work as expected

class DummyConfig(BaseSettings):
    PARAM1: str 
    PARAM2: str 
    API_KEY:  str
    model_config = SettingsConfigDict(
        env_prefix="dummy_", env_file=".env", env_file_encoding="utf-8"
    )

    @classmethod
    def customise_sources(
            cls,
            init_settings,
            env_settings,
            file_secret_settings):
        return (
            init_settings,
            env_settings,
            get_aws_prams_store,
            file_secret_settings,
        )


def get_aws_prams_store(param_name: str = "/dev/dummy-service/API_KEY"):
    session = boto3.session.Session(
        profile_name='my-account', region_name="us-east-1")
    client = session.client("ssm")
    response = client.get_parameters(
        Names=[param_name],
        WithDecryption=True
    )
    params = {param["Name"].split("/")[-1]: param["Value"]
              for param in response["Parameters"]}
    return params

.env content:

dummy_PARAM1=value1
dummy_PARAM2=value1

I tested get_aws_prams_store out of config and it's working well

config= DummyConfig()

I got an error API_KEY Field required [type=missing, .....]

I set API_KEY=None just to pass the test.

print("PARAM1", config.PARAM1) #value1
print("PARAM2", config.PARAM2) # value2
print("API_KEY", config.API_KEY) # None

I am not sure I am doing wrong ?

Upvotes: 0

Views: 34

Answers (1)

timothy jeong
timothy jeong

Reputation: 164

I think the problem is Custom Source Function Signature,

Pydantic’s settings system calls each custom source (that you list in the tuple returned by the source-customisation method) as a callable without any arguments.

In your original code, get_aws_prams_store is defined to accept a parameter param_name with a default value. However, when Pydantic calls your custom source, it may try to supply its own arguments (or none at all), which does not match your function’s expected signature.

In pydantic 2, you can implement DummyConfig like: (if your not in Pydantic 2 version, just let me know)

# settings.py
from typing import Tuple, Type
from pydantic_settings import (
    BaseSettings,
    PydanticBaseSettingsSource,
)
from pydantic_settings import SettingsConfigDict
import boto3

def get_aws_prams_store(settings_cls) -> dict:
    param_name = "/dev/dummy-service/API_KEY"
    session = boto3.session.Session(profile_name='default', region_name="us-east-1")
    client = session.client("ssm")
    response = client.get_parameters(
        Names=[param_name],
        WithDecryption=True
    )
    return {param["Name"].split("/")[-1]: param["Value"] for param in response["Parameters"]}

class DummyConfig(BaseSettings):
    PARAM1: str 
    PARAM2: str 
    API_KEY:  str

    model_config = SettingsConfigDict(
        env_prefix="dummy_", 
        env_file=".env", 
        env_file_encoding="utf-8"
    )

    @classmethod
    def settings_customise_sources(
        cls,
        settings_cls: Type[BaseSettings],
        init_settings: PydanticBaseSettingsSource,
        env_settings: PydanticBaseSettingsSource,
        dotenv_settings: PydanticBaseSettingsSource,
        file_secret_settings: PydanticBaseSettingsSource,
    ) -> Tuple[PydanticBaseSettingsSource, ...]:
        # wrapping 
        aws_source = lambda: get_aws_prams_store(settings_cls)
        return (
            init_settings,
            env_settings,
            dotenv_settings,
            aws_source,
            file_secret_settings,
        )

and i check that config in this code

# main.py
from app.settings import DummyConfig

def main():
    try:
        config = DummyConfig()
        print("PARAM1:", config.PARAM1)
        print("PARAM2:", config.PARAM2)
        print("API_KEY:", config.API_KEY)
    except Exception as e:
        print("Error during configuration:", e)

if __name__ == "__main__":
    main()

I refer Pydactic Document:Customise settings sources

Upvotes: 0

Related Questions