sh-edraft.de/sh_cpl
1
1
Fork 0
Code Issues 4 Pull Requests 1 Actions Packages Projects 1 Releases 12 Wiki Activity

Removed console from db connection

Browse Source
This commit is contained in:
Sven Heidemann 2021-11-30 11:15:41 +01:00
parent 127c67e566
commit de22e79b71
1 changed files with 24 additions and 34 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

58
src/cpl_core/database/connection/database_connection.py
View File

@ -1,8 +1,6 @@
from typing import Optional from typing import Optional
import mysql.connector as sql import mysql.connector as sql
from cpl_core.console.console import Console
from cpl_core.console.foreground_color_enum import ForegroundColorEnum
from cpl_core.database.connection.database_connection_abc import \ from cpl_core.database.connection.database_connection_abc import \
DatabaseConnectionABC DatabaseConnectionABC
from cpl_core.database.database_settings import DatabaseSettings from cpl_core.database.database_settings import DatabaseSettings
@ -20,43 +18,35 @@ class DatabaseConnection(DatabaseConnectionABC):
self._database: Optional[MySQLConnectionAbstract] = None self._database: Optional[MySQLConnectionAbstract] = None
self._cursor: Optional[MySQLCursorBuffered] = None self._cursor: Optional[MySQLCursorBuffered] = None
@property @property
def server(self) -> MySQLConnectionAbstract: def server(self) -> MySQLConnectionAbstract:
return self._database return self._database
@property @property
def cursor(self) -> MySQLCursorBuffered: def cursor(self) -> MySQLCursorBuffered:
return self._cursor return self._cursor
def connect(self, database_settings: DatabaseSettings): def connect(self, database_settings: DatabaseSettings):
try: connection = sql.connect(
connection = sql.connect( host=database_settings.host,
host=database_settings.host, user=database_settings.user,
user=database_settings.user, passwd=CredentialManager.decrypt(database_settings.password),
passwd=CredentialManager.decrypt(database_settings.password), charset=database_settings.charset,
charset=database_settings.charset, use_unicode=database_settings.use_unicode,
use_unicode=database_settings.use_unicode, buffered=database_settings.buffered,
buffered=database_settings.buffered, auth_plugin=database_settings.auth_plugin
auth_plugin=database_settings.auth_plugin )
) connection.cursor().execute(
connection.cursor().execute(f'CREATE DATABASE IF NOT EXISTS `{database_settings.database}`;') f'CREATE DATABASE IF NOT EXISTS `{database_settings.database}`;')
self._database = sql.connect( self._database = sql.connect(
host=database_settings.host, host=database_settings.host,
user=database_settings.user, user=database_settings.user,
passwd=CredentialManager.decrypt(database_settings.password), passwd=CredentialManager.decrypt(database_settings.password),
db=database_settings.database, db=database_settings.database,
charset=database_settings.charset, charset=database_settings.charset,
use_unicode=database_settings.use_unicode, use_unicode=database_settings.use_unicode,
buffered=database_settings.buffered, buffered=database_settings.buffered,
auth_plugin=database_settings.auth_plugin auth_plugin=database_settings.auth_plugin
) )
self._cursor = self._database.cursor() self._cursor = self._database.cursor()
Console.set_foreground_color(ForegroundColorEnum.green)
Console.write_line(f'[{__name__}] Connected to database')
Console.set_foreground_color(ForegroundColorEnum.default)
except Exception as e:
Console.set_foreground_color(ForegroundColorEnum.red)
Console.write_line(f'[{__name__}] Database connection failed -> {e}')
Console.set_foreground_color(ForegroundColorEnum.default)
exit()