"""Config for YouTube Statistics Workflow.""" import os from cryptography.hazmat.backends import default_backend from cryptography.hazmat.primitives import serialization YOUTUBE_LINK = 'https://www.youtube.com/channel/' VIDEO_LINK = 'https://www.youtube.com/watch?v=' env = os.environ.get('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 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': os.environ.get('SNOWFLAKE_PASSWORD'), 'account': os.environ.get('SNOWFLAKE_ACCOUNT'), 'private_key': PRIVATE_KEY } STATS_NAME = 'video-statistics' FLOW_NAME = 'youtube-statistics' snowflake_table_names = { 'video_statistics': 'video_statistics_101', } youtube = {}