Reputation: 300
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