34ff075ba7
* Secure logger implementation minus still broken new tests * Secure logger implementation and unit tests * code review changes * filter text on uncaught exceptions * auto-formatting * Mistaken change left in code * filter text on uncaught exceptions * Simplify code, remove LoggingFilter and move obfuscation functionality to Formatter * Update airbyte-cdk/python/airbyte_cdk/entrypoint.py Co-authored-by: Eugene Kulak <widowmakerreborn@gmail.com> * Obfuscate Secrets in Logging, code review changes * Obfuscate Secrets in Logging, code review changes, unit test fixes * CHANGELOG.md * Format and flake8 * Fix build error/bump version/format * Add airbyte prefix on test logger Co-authored-by: Eugene Kulak <widowmakerreborn@gmail.com>
21 lines
835 B
Python
21 lines
835 B
Python
#
|
|
# Copyright (c) 2021 Airbyte, Inc., all rights reserved.
|
|
#
|
|
|
|
import logging
|
|
from typing import Any, List, Mapping
|
|
|
|
from airbyte_cdk.sources import Source
|
|
from airbyte_cdk.utils.mapping_utils import all_key_pairs_dot_notation, get_value_by_dot_notation
|
|
|
|
|
|
def get_secrets(source: Source, config: Mapping[str, Any], logger: logging.Logger) -> List[Any]:
|
|
"""
|
|
Get a list of secrets from the source config based on the source specification
|
|
"""
|
|
flattened_key_values = all_key_pairs_dot_notation(source.spec(logger).connectionSpecification.get("properties", {}))
|
|
secret_key_names = [
|
|
".".join(key.split(".")[:1]) for key, value in flattened_key_values.items() if value and key.endswith("airbyte_secret")
|
|
]
|
|
return [str(get_value_by_dot_notation(config, key)) for key in secret_key_names if config.get(key)]
|