mirror of
https://github.com/open-metadata/OpenMetadata.git
synced 2025-08-12 02:56:59 +00:00
41ea49d11e
* Fix: Secrets Manager not working with Airflow config * Address Python Checkstyle * Minor change
39 lines
1.3 KiB
Python
39 lines
1.3 KiB
Python
from typing import Optional
|
|
|
|
from airflow.configuration import conf
|
|
|
|
from metadata.generated.schema.security.credentials.awsCredentials import AWSCredentials
|
|
from metadata.generated.schema.security.secrets.secretsManagerProvider import (
|
|
SecretsManagerProvider,
|
|
)
|
|
from metadata.ingestion.models.custom_pydantic import CustomSecretStr
|
|
from metadata.utils.secrets.secrets_manager import SECRET_MANAGER_AIRFLOW_CONF
|
|
|
|
|
|
def build_aws_credentials() -> Optional[AWSCredentials]:
|
|
aws_region = conf.get(SECRET_MANAGER_AIRFLOW_CONF, "aws_region", fallback=None)
|
|
if aws_region:
|
|
credentials = AWSCredentials(awsRegion=aws_region)
|
|
credentials.awsAccessKeyId = conf.get(
|
|
SECRET_MANAGER_AIRFLOW_CONF, "aws_access_key_id", fallback=""
|
|
)
|
|
credentials.awsSecretAccessKey = CustomSecretStr(
|
|
conf.get(SECRET_MANAGER_AIRFLOW_CONF, "aws_secret_access_key", fallback="")
|
|
)
|
|
return credentials
|
|
return None
|
|
|
|
|
|
def build_secrets_manager_credentials(
|
|
secrets_manager: SecretsManagerProvider,
|
|
) -> Optional[AWSCredentials]:
|
|
if secrets_manager in {
|
|
SecretsManagerProvider.aws,
|
|
SecretsManagerProvider.managed_aws,
|
|
SecretsManagerProvider.aws_ssm,
|
|
SecretsManagerProvider.managed_aws_ssm,
|
|
}:
|
|
return build_aws_credentials()
|
|
|
|
return None
|