blob: ceccc3289e9fc31643c15331b24a81cc0470d43c (
plain)
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
|
"""Check the various connection used in the application"""
import sys
import traceback
import redis
import MySQLdb
from qc_app.db_utils import database_connection
def check_redis(uri: str):
"Check the redis connection"
try:
with redis.Redis.from_url(uri) as rconn:
rconn.ping()
except redis.exceptions.ConnectionError as conn_err:
print(conn_err, file=sys.stderr)
print(traceback.format_exc(), file=sys.stderr)
sys.exit(1)
def check_db(uri: str):
"Check the mysql connection"
try:
with database_connection(uri) as dbconn: # pylint: disable=[unused-variable]
pass
except MySQLdb.OperationalError as op_err:
print(op_err, file=sys.stderr)
print(traceback.format_exc(), file=sys.stderr)
sys.exit(1)
|
