Sevy
Sevy

Reputation: 300

passing values from child to another child

I want to pass the SSM Client from AWS Config to AMQP Config. Unfortunately my code throws a KeyError: 'ssm_client'.

I tried to override the __init__ method but there were multiple other errors.

I think I need to find a way how to order the validators properly but no idea how.

class Config(BaseSettings):
    env: str = os.environ["ENV"]

    aws: AWSConfig = AWSConfig()
    db: DBConfig = DBConfig()

    @computed_field
    def amqp(self) -> AMQPConfig:
        return AMQPConfig(ssm_client=self.aws.ssm_client)

class AMQPConfig(BaseModel):
    ssm_client: SSMClient

    protocol: str = "amqps"
    host: str = os.environ["BROKER_HOST"]
    port: str | int = os.environ["BROKER_PORT"]
    queue: str = "test1234567"
    username: SecretStr | None = None
    password: SecretStr | None = None
    url: str | None = None

    @validator("url", pre=True, always=True)
    def set_url(cls, v, values) -> str:
        if not values.get("username") or not values.get("password"):
            username, password = retrieve_ssm_parameter(
                client=values["ssm_client"],
                parameter_name="/OEMDataRetrieve/brokerCredentials",
            ).split(" ")
            values["username"], values["password"] = SecretStr(username), SecretStr(
                password
            )
        if not v:
            v = create_connection_url(
                protocol=values["protocol"],
                host=values["host"],
                port=str(values["port"]),
                path=values["queue"],
                username=values["username"].get_secret_value(),
                password=values["password"].get_secret_value(),
            )
        return v

    class Config:
        arbitrary_types_allowed = True

Upvotes: 0

Views: 21

Answers (0)

Related Questions