"""Config file for lambda function.""" import os import subprocess from cryptography.hazmat.backends import default_backend from cryptography.hazmat.primitives import serialization from fangare import catchers from secrets_manager.lambda_ext import LambdaSecretsManager APPLICATION_NAME = LAMBDA_NAME = 'lambda-gda-tiktok-scraper' TIKTOK_LINK = 'https://www.tiktok.com/@' # AWS region configuration DEFAULT_AWS_REGION = 'us-east-1' AWS_REGION = os.environ.get('AWS_REGION', DEFAULT_AWS_REGION) # Environment setup QA_ENVIRONMENT = 'qa' PROD_ENVIRONMENT = 'prod' DEV_ENVIRONMENT = 'dev' TEST_ENVIRONMENT = 'test' ENVIRONMENT = os.environ.get('Environment', DEV_ENVIRONMENT) force_remote_secret = ENVIRONMENT != TEST_ENVIRONMENT secrets = LambdaSecretsManager( service_name=APPLICATION_NAME, aws_region=AWS_REGION, environment=ENVIRONMENT, force_remote=force_remote_secret) STATS_NAME = 'tiktok-statistics' TIKTOK_METRICS = ['followers', 'following', 'likes'] if ENVIRONMENT == 'prod': try: GIT_HASH_OF_LAST_COMMIT = ( subprocess.check_output( 'git rev-parse HEAD', shell=True).decode().strip()) except subprocess.CalledProcessError: GIT_HASH_OF_LAST_COMMIT = 'tiktok_commit' else: GIT_HASH_OF_LAST_COMMIT = 'tiktok_{}_commit'.format(ENVIRONMENT) # get Snowflake private_key SNOWFLAKE_PRIVATE_KEY_PATH = os.getenv( 'SNOWFLAKE_PRIVATE_KEY_PATH', '{}/.ssh/snowflake/rsa_key.p8'.format(os.getenv('HOME'))) SNOWFLAKE_KEY_PASSPHRASE = os.getenv('SNOWFLAKE_KEY_PASSPHRASE') if SNOWFLAKE_KEY_PASSPHRASE: # Snowflake key with open(SNOWFLAKE_PRIVATE_KEY_PATH, 'rb') as key: p_key = serialization.load_pem_private_key( key.read(), password=SNOWFLAKE_KEY_PASSPHRASE.encode(), backend=default_backend() ) PRIVATE_KEY = p_key.private_bytes( encoding=serialization.Encoding.DER, format=serialization.PrivateFormat.PKCS8, encryption_algorithm=serialization.NoEncryption()) else: PRIVATE_KEY = None SCRAPER_API_KEY = catchers.get_secret('TIKTOK_PASSWORD', 'tiktok-statistics', ENVIRONMENT) SF_CONFIG = { 'role': os.environ.get('SNOWFLAKE_ROLE'), 'warehouse': os.environ.get('SNOWFLAKE_WAREHOUSE'), 'db': os.environ.get('SNOWFLAKE_DATABASE'), 'schema': os.environ.get('SNOWFLAKE_SCHEMA'), 'user': os.environ.get('SNOWFLAKE_USER'), 'password': catchers.get_secret('SNOWFLAKE_PASSWORD', 'youtube-statistics', ENVIRONMENT), 'account': os.environ.get('SNOWFLAKE_ACCOUNT'), 'private_key': PRIVATE_KEY } snowflake_table_names = { 'artist_to_track': 'artist_to_track', 'tiktok_statistics': 'tiktok_statistics', 'staging_raw': 'staging_raw_tiktok_statistics' }