Get connection URI with AWS Secrets Manager

[JavierLopezT]

One of the main advantages of using AWS Secrets Manager is its ability to automatically create secrets of RDS databases and Redshift databases. Those secrets consist of several keys with their values, i.e user, pass, etc. Also, it is normal to store API Keys, sftp, or whatever using different values, as shown in the picture below:

With the current code, all the keys and values obtained from a secret are stored in the schema attribute of the conn object, unless you have just one key with the conn_uri in the value. Thus, the current situation is forcing to use Secrets Manager in a way it is not intended to.

With this proposed modification, you can use AWS Secrets Manager using keys and values and have some kind of freedom to choose different words for each key to make the get_conn work.

I need help and guidance with tests, please. Thanks

[github-actions]

The Workflow run is cancelling this PR. It has some failed jobs matching ^Pylint$,^Static checks,^Build docs$,^Spell check docs$,^Provider packages,^Checks: Helm tests$,^Test OpenAPI*.

[xinbinhuang]

Suggested change

	counter = 0
	for key, value in extra_dict.items():
	if counter == 0:
	conn_string += f'?{key}={value}'
	else:
	conn_string += f'&{key}={value}'
	counter += 1
	return conn_string
	kvs = "&".join([f"{key}={value} for key, value in extra_dict.items()])
	return f"{conn_string}?{kvs}"

[xinbinhuang]

I am not convinced that we need ast.literal_eval here. Can you explain a bit more on why we need this?

[xinbinhuang]

Another note, the signature for is _get_secret(self, path_prefix: str, secret_id: str).

airflow/airflow/providers/amazon/aws/secrets/secrets_manager.py

Lines 190 to 200 in c651ab7

	def _get_secret(self, path_prefix: str, secret_id: str) -> Optional[str]:
	"""
	Get secret value from Secrets Manager
	:param path_prefix: Prefix for the Path to get Secret
	:type path_prefix: str
	:param secret_id: Secret Key
	:type secret_id: str
	"""
	secrets_path = self.build_path(path_prefix, secret_id, self.sep)
	try:

[xinbinhuang]

Suggested change

	conn_type = conn_d['conn_type']
	user = conn_d['user']
	password = conn_d['password']
	host = conn_d['host']
	port = conn_d['port']
	schema = conn_d['schema']
	conn_string = f'{conn_type}://{user}:{password}@{host}:{port}/{schema}'
	connection = self._get_extra(secret, conn_string)
	from airflow.models import Connection
	connection_uri = Connection(**conn_d, extra=extra).get_uri()

[xinbinhuang]

Also, I would recommend separating this part of logic (if secret: ) into another method for easier testing

[xinbinhuang]

I feel like this PR is also related to #15146, #15146, and #15100 . cc: @dstandish @davlum

[dstandish]

may i suggest using _create_connection function here
https://github.com/apache/airflow/pull/15013/files#diff-258df8f49a97540963b45ae4ac0e0f71720101134f73b2981b8ab3dd1a4cf5fcR24

i think we ultimately ought to provide a shared way of parsing creds from json and this function could be a good starting point.

PR #15013 is doing the same thing for Vault and is re-using that function.

[github-actions]

This pull request has been automatically marked as stale because it has not had recent activity. It will be closed in 5 days if no further activity occurs. Thank you for your contributions.