tiararodney/httpaste
2
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

fix(backend/mysql): isolate third-party module imports

Browse source 
currently there is a problem with loading the backends through
the httpaste.Config class, since they aren't being lazily loaded, will have to
rework this sometime later in the future.
This commit is contained in:
Tiara Rodney 2022-04-03 02:45:01 +02:00
parent 4993a5a3ea
commit 9c31f044ce
1 changed files with 42 additions and 28 deletions
Download patch file Download diff file Expand all files Collapse all files

70
src/httpaste/backend/mysql/__init__.py
View file

@ -1,19 +1,6 @@
"""MySQL backend """MySQL backend
""" """
try: from typing import NamedTuple, Optional, Callable
from mysql.connector import connect
from mysql.connector.connection import MySQLConnection
except ImportError as e:
raise ImportError(' '.join((
'\'mysql-connector-python\' is not installed.',
'Install it by running',
'\'python3 -m pip install mysql-connector-python\'.'
))) from e
from typing import NamedTuple, Optional
from . import user
from . import paste
class Parameters(NamedTuple): class Parameters(NamedTuple):
@ -29,65 +16,77 @@ class Parameters(NamedTuple):
#: database identifier #: database identifier
database: str database: str
#: a mysql.connection.MySQLConnection object (does not apply to config) #: a mysql.connection.MySQLConnection object (does not apply to config)
connection: Optional[MySQLConnection] = None connection: Optional[object] = None
class User(object): class User(object):
"""MySQL user model backend """MySQL user model backend
""" """
connection: MySQLConnection connection: object
def __init__(self, parameters: Parameters, model_class: type) -> None: def __init__(self, parameters: Parameters, model_class: type) -> None:
from . import user
connect = get_mysql_connect_callee()
self.interface = user
self.model_class = model_class self.model_class = model_class
self.connection = get_connection(parameters) self.connection = get_connection(parameters)
def load(self, proto: object) -> object: def load(self, proto: object) -> object:
return user.load(proto, self.connection, self.model_class) return self.interface.load(proto, self.connection, self.model_class)
def dump(self, model: object) -> None: def dump(self, model: object) -> None:
return user.dump(model, self.connection) return self.interface.dump(model, self.connection)
def delete(self, proto: object) -> None: def delete(self, proto: object) -> None:
return user.delete(proto, self.connection) return self.interface.delete(proto, self.connection)
def init(self) -> None: def init(self) -> None:
return user.init(self.connection) return self.interface.init(self.connection)
def sanitize(self) -> None: def sanitize(self) -> None:
return user.sanitize(self.connection, self.model_class) return self.interface.sanitize(self.connection, self.model_class)
class Paste(object): class Paste(object):
"""MySQL paste model backend """MySQL paste model backend
""" """
connection: MySQLConnection connection: object
def __init__(self, parameters: Parameters, model_class: type) -> None: def __init__(self, parameters: Parameters, model_class: type) -> None:
from . import paste
connect = get_mysql_connect_callee()
self.interface = paste
self.model_class = model_class self.model_class = model_class
self.connection = get_connection(parameters) self.connection = get_connection(parameters, connect)
def load(self, proto: object) -> object: def load(self, proto: object) -> object:
return paste.load(proto, self.connection, self.model_class) return self.interface.load(proto, self.connection, self.model_class)
def dump(self, model: object) -> None: def dump(self, model: object) -> None:
return paste.dump(model, self.connection) return self.interface.dump(model, self.connection)
def delete(self, proto: object) -> None: def delete(self, proto: object) -> None:
return paste.delete(proto, self.connection) return self.interface.delete(proto, self.connection)
def init(self) -> None: def init(self) -> None:
@ -95,10 +94,25 @@ class Paste(object):
def sanitize(self) -> None: def sanitize(self) -> None:
return paste.sanitize(self.connection, self.model_class) return self.interface.sanitize(self.connection, self.model_class)
def get_connection(parameters: Parameters) -> MySQLConnection: def get_mysql_connect_callee() -> object:
try:
from mysql.connector import connect
from mysql.connector.connection import MySQLConnection
except ImportError as e:
raise ImportError(' '.join((
'\'mysql-connector-python\' is not installed.',
'Install it by running',
'\'python3 -m pip install mysql-connector-python\'.'
))) from e
return connect
def get_connection(parameters: Parameters, connect_callee: Callable) -> object:
"""get a mysql.connection.MySQLConnection object """get a mysql.connection.MySQLConnection object
""" """