-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathsettings.py
More file actions
94 lines (81 loc) · 2.58 KB
/
Copy pathsettings.py
File metadata and controls
94 lines (81 loc) · 2.58 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
import boto3
import json
import os
import logging
import psycopg
logging.basicConfig(level=logging.ERROR)
log = logging.getLogger(__name__)
'''
init
Initialize global variables to use across our modules
'''
def init():
global db_secrets
global memcached_cluster_cfg_ep
global memcached_cache_ttl
global auth0_token
global auth0_domain
global auth0_client_ids
global auth0_be_audience
db_credentials_secrets_store_arn = os.environ.get('SR_DB_SECRET_ARN')
memcached_cluster_cfg_ep = os.environ.get('SR_MEMCACHED_CFG_EP')
memcached_cache_ttl = 3600
# Auth0
auth0_token = os.environ.get('SR_AUTH0_TOKEN')
auth0_domain = os.environ.get('SR_AUTH0_DOMAIN')
auth0_client_ids = json.loads(os.environ.get('SR_AUTH0_CLIENT_IDS'))
auth0_be_audience = os.environ.get('SR_AUTH0_BE_AUDIENCE')
try:
secrets_client = boto3.client('secretsmanager')
db_secret_value = secrets_client.get_secret_value(
SecretId = db_credentials_secrets_store_arn
)
except Exception as error:
raise error
if 'SecretString' in db_secret_value:
db_secrets = json.loads(db_secret_value['SecretString'])
'''
db_conn
Reusable database connection
'''
def db_conn():
database_name = os.environ.get('SR_DB_NAME')
try:
return psycopg.connect(
host = db_secrets['host'],
port = db_secrets['port'],
dbname = database_name,
user = db_secrets['username'],
password = db_secrets['password']
)
except Exception as error:
raise error
'''
logging_config
Sets global logging configuration. Set to INFO in testing. Set to WARN or ERROR otherwise
'''
def logging_config():
global log_level
log_level = logging.ERROR
if os.environ['SR_LOGGING_LEVEL']:
log_level = {
'DEBUG': logging.DEBUG,
'INFO': logging.INFO,
'WARN': logging.WARN,
'ERROR': logging.ERROR,
'CRITICAL': logging.CRITICAL
}.get(os.environ['SR_LOGGING_LEVEL'], logging.ERROR)
'''
return_error
'''
def return_error(status_code=500, message='An Unknown Error Occurred'):
log.info(':: return_error')
body = f'{{"status_code":"{status_code}","error":"{message}"}}'
print(f':: return_error: {body}')
return {
'statusCode': status_code,
'headers': {
'Content-Type': 'application/json'
},
'body': body
}