import os
import json
import platform
import logging
import pymysql as pm
import boto3
from botocore.exceptions import ClientError
import logging
class LogOutput:
env=None
config=None
def __init__(self, env_filename):
self.env=env_filename
self.config=self.get_config()
def get_config(self):
with open(self.env) as file_in:
return json.load(file_in)
def is_Windows():
if "win" in (platform.system().lower()):
return True
else:
return False
def DB_connection(self):
logger = logging.getLogger()
logger.setLevel(logging.INFO)
connection = None
try:
config=LogOutput.get_config(self)
host=config["DB_conn_info_local"]["database-secrets"]["host"]
port=config["DB_conn_info_local"]["database-secrets"]["port"]
database=config["DB_conn_info_local"]["database-secrets"]["db"]
username=config["DB_conn_info_local"]["database-secrets"]["username"]
password=config["DB_conn_info_local"]["database-secrets"]["password"]
# Load the configuration from DB_conn_info_local.json
config_file_path = os.path.join(os.path.dirname(__file__), 'DB_conn_info_local.json')
with open(config_file_path) as config_file:
config = json.load(config_file)
connection = pm.connect(user=username,password=password,host=host,port=port,database=database)
logger.info("Successfully connected to database")
except Exception as e:
logger.error("Unable to connect to database: %s", str(e))
return connection
def logging_sp(self):
logging_sp_query = """
stored procedure query here
"""
with self.DB_connection() as cnxn:
with cnxn.cursor() as cur:
try:
cur.execute(logging_sp_query)
except Exception as e:
logging.exception(e)
else:
cnxn.commit()
def main():
cwd=os.getcwd()
if "win" in (platform.system().lower()):
vfc=(cwd+"\DB_conn_info_local"+".json")
else:
vfc=(cwd+"/DB_conn_info_local"+".json")
ve=LogOutput(vfc)
ve.logging_sp()
if __name__ == "__main__":
main()
I keep getting this error:
with self.DB_connection() as cnxn:
TypeError: 'NoneType' object does not support the context manager protocol
Is it because I am using “with”? Is there another way to execute it? It uses a json config file that it reads which contains the credentials. I didn’t need to include that here. I saw other posts saying “with” return none so that’s why it’s executing basically nothing? I don’t know how to re-write this because I am using with
twice.
2
Ensure DB_connection() returns a valid connection:
Your DB_connection() method should return the connection object from pymysql when successful.
In case of an error, ensure that you either raise an exception or handle the None value appropriately.
Check for None in logging_sp():
Before proceeding with with self.DB_connection(), you should verify that the connection object (cnxn) is not None.
import os
import json
import platform
import pymysql as pm
import boto3
from botocore.exceptions import ClientError
import logging
class LogOutput:
env = None
config = None
def __init__(self, env_filename):
self.env = env_filename
self.config = self.get_config()
def get_config(self):
with open(self.env) as file_in:
return json.load(file_in)
def is_Windows():
return "win" in platform.system().lower()
def DB_connection(self):
logger = logging.getLogger()
logger.setLevel(logging.INFO)
connection = None
try:
config = self.config
host = config["DB_conn_info_local"]["database-secrets"]["host"]
port = config["DB_conn_info_local"]["database-secrets"]["port"]
database = config["DB_conn_info_local"]["database-secrets"]["db"]
username = config["DB_conn_info_local"]["database-secrets"]["username"]
password = config["DB_conn_info_local"]["database-secrets"]["password"]
# Establish the connection
connection = pm.connect(user=username, password=password, host=host, port=port, database=database)
logger.info("Successfully connected to database")
except Exception as e:
logger.error("Unable to connect to database: %s", str(e))
return connection
def logging_sp(self):
logging_sp_query = """
stored procedure query here
"""
cnxn = self.DB_connection()
if cnxn is None:
logging.error("Database connection is None. Cannot proceed.")
return # we exit here if there's no valid connection
with cnxn.cursor() as cur:
try:
cur.execute(logging_sp_query)
except Exception as e:
logging.exception(e)
else:
cnxn.commit()
cnxn.close() # we have to close the connection afterwards
def main():
cwd = os.getcwd()
if "win" in platform.system().lower():
vfc = os.path.join(cwd, "DB_conn_info_local.json")
else:
vfc = os.path.join(cwd, "DB_conn_info_local.json")
ve = LogOutput(vfc)
ve.logging_sp()
if __name__ == "__main__":
main()
8