counterweight/drogon
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Formatting.

Browse source
This commit is contained in:
pablo 2020-11-03 07:29:17 +01:00
parent cd9c3b6e39
commit a79fc533ee
11 changed files with 231 additions and 204 deletions
Download patch file Download diff file Expand all files Collapse all files

69
core/mysql_wrapper.py
View file

@ -1,31 +1,34 @@
# -*- coding: utf-8 -*-
import sys
sys.path.append('..')
sys.path.append("..")
import mysql.connector
from core.alerts import alert_master
from core.config import current_db_parameters
anuncios_db_parameters = {'database': 'anuncios', **current_db_parameters}
tasks_db_parameters = {'database': 'tasks', **current_db_parameters}
anuncios_db_parameters = {"database": "anuncios", **current_db_parameters}
tasks_db_parameters = {"database": "tasks", **current_db_parameters}
class DatabaseWrapper():
class DatabaseWrapper:
def __init__(self, connection_parameters):
self.host = connection_parameters['host']
self.database = connection_parameters['database']
self.user = connection_parameters['user']
self.password = connection_parameters['password']
self.host = connection_parameters["host"]
self.database = connection_parameters["database"]
self.user = connection_parameters["user"]
self.password = connection_parameters["password"]
self.connection = None
self.ping()
def connect(self):
self.connection = mysql.connector.connect(host = self.host,
database = self.database,
user = self.user,
password = self.password,
autocommit = False)
self.connection = mysql.connector.connect(
host=self.host,
database=self.database,
user=self.user,
password=self.password,
autocommit=False,
)
def disconnect(self):
if self.connection.is_connected():
self.connection.disconnect()
@ -33,41 +36,41 @@ class DatabaseWrapper():
def ping(self):
self.connect()
self.disconnect()
def query(self, query_statement, query_parameters=None, dictionary=False):
self.connect()
if self.connection.is_connected():
try:
execution_cursor = self.connection.cursor(dictionary=dictionary,
buffered=True)
execution_cursor = self.connection.cursor(
dictionary=dictionary, buffered=True
)
execution_cursor.execute(query_statement, query_parameters)
self.connection.commit()
self.disconnect()
return execution_cursor
except Exception as e:
alert_master("SQL ERROR", """Se ha producido un error ejecutando la
alert_master(
"SQL ERROR",
"""Se ha producido un error ejecutando la
siguiente query: {}.
Con los siguientes parametros: {}
{}
""".format(query_statement,
query_parameters,
e))
""".format(
query_statement, query_parameters, e
),
)
else:
raise Exception("Could not connect to the database.")
def query_dict(self, query_statement, query_parameters = None):
return self.query(query_statement, query_parameters, dictionary = True)
def query_dict(self, query_statement, query_parameters=None):
return self.query(query_statement, query_parameters, dictionary=True)
def get_anunciosdb():
return DatabaseWrapper(anuncios_db_parameters)
return DatabaseWrapper(anuncios_db_parameters)
def get_tasksdb():
return DatabaseWrapper(tasks_db_parameters)