pyceo/ceod/db/MySQLService.py

90 lines
3.2 KiB
Python
Raw Normal View History

2021-08-21 00:34:10 -04:00
from zope.interface import implementer
from zope import component
2021-08-26 15:42:18 -04:00
from contextlib import contextmanager
2021-08-24 02:14:28 -04:00
from ceo_common.interfaces import IDatabaseService, IConfig
2021-08-26 15:42:18 -04:00
from ceo_common.errors import DatabaseConnectionError, DatabasePermissionError, UserAlreadyExistsError, \
UserNotFoundError
2021-08-24 02:14:28 -04:00
from ceod.utils import gen_password
from ceod.db.utils import response_is_empty
2021-08-26 15:42:18 -04:00
2021-08-24 02:14:28 -04:00
from mysql.connector import connect
from mysql.connector.errors import InterfaceError, ProgrammingError
2021-08-21 00:34:10 -04:00
@implementer(IDatabaseService)
class MySQLService:
2021-08-26 02:02:47 -04:00
type = 'mysql'
2021-08-21 00:34:10 -04:00
def __init__(self):
config = component.getUtility(IConfig)
self.auth_username = config.get('mysql_username')
self.auth_password = config.get('mysql_password')
2021-08-26 15:42:18 -04:00
@contextmanager
def mysql_connection(self):
try:
with connect(
host='localhost',
user=self.auth_username,
password=self.auth_password,
) as con:
yield con
except InterfaceError:
raise DatabaseConnectionError()
except ProgrammingError:
raise DatabasePermissionError()
2021-08-21 00:34:10 -04:00
def create_db(self, username: str) -> str:
2021-08-24 02:14:28 -04:00
password = gen_password()
search_for_user = f"SELECT user FROM mysql.user WHERE user='{username}'"
search_for_db = f"SHOW DATABASES LIKE '{username}'"
2021-08-26 16:45:24 -04:00
create_user = f"""
CREATE USER '{username}'@'localhost' IDENTIFIED BY %(password)s;
CREATE USER '{username}'@'%' IDENTIFIED BY %(password)s;
"""
2021-08-24 02:14:28 -04:00
create_database = f"""
CREATE DATABASE {username};
GRANT ALL PRIVILEGES ON {username}.* TO '{username}'@'localhost';
2021-08-26 16:45:24 -04:00
GRANT ALL PRIVILEGES ON {username}.* TO '{username}'@'%';
2021-08-24 02:14:28 -04:00
"""
2021-08-26 15:42:18 -04:00
with self.mysql_connection() as con:
with con.cursor() as cursor:
if response_is_empty(search_for_user, con):
cursor.execute(create_user, {'password': password})
2021-08-24 02:14:28 -04:00
if response_is_empty(search_for_db, con):
cursor.execute(create_database)
2021-08-26 15:42:18 -04:00
else:
raise UserAlreadyExistsError()
return password
def reset_db_passwd(self, username: str) -> str:
password = gen_password()
search_for_user = f"SELECT user FROM mysql.user WHERE user='{username}'"
2021-08-26 16:45:24 -04:00
reset_password = f"""
ALTER USER '{username}'@'localhost' IDENTIFIED BY %(password)s
ALTER USER '{username}'@'%' IDENTIFIED BY %(password)s
"""
2021-08-26 15:42:18 -04:00
with self.mysql_connection() as con:
with con.cursor() as cursor:
if not response_is_empty(search_for_user, con):
cursor.execute(reset_password, {'password': password})
else:
raise UserNotFoundError(username)
return password
2021-08-24 22:31:50 -04:00
def delete_db(self, username: str):
drop_db = f"DROP DATABASE IF EXISTS {username}"
2021-08-26 16:45:24 -04:00
drop_user = f"""
DROP USER IF EXISTS '{username}'@'localhost';
DROP USER IF EXISTS '{username}'@'%';
"""
2021-08-26 15:42:18 -04:00
with self.mysql_connection() as con:
with con.cursor() as cursor:
cursor.execute(drop_db)
cursor.execute(drop_user)