我有database.py其中包含数据库执行函数
import sqlachemy
class DB():
def __init__(self, **kwargs):
self.connection = None
self.connect(**kwargs)
def connect(self, **kwargs):
if 'url' in kwargs and kwargs.get('url') is not None:
return self.connectUrl(kwargs.get('url'), kwargs.get('username'),
kwargs.get('password'), kwargs.get('database'))
else :
return self.connectHost(kwargs.get('host'), kwargs.get('port'), kwargs.get('username'),
kwargs.get('password'), kwargs.get('database'))
def execute(self, query):
if self.connect is None:
raise Exception('No connection')
try:
with self.connection.connect() as conn:
return conn.execute(query)
except (pymysql.err.OperationalError, sqlalchemy.exc.OperationalError) as e:
print(e)
except (Exception) as e:
print(e)
else:
return None
Run Code Online (Sandbox Code Playgroud)
我还有一个 main.py. 我从上面的文件导入 DB 类并在此处使用它。
from database import db …Run Code Online (Sandbox Code Playgroud)