Spaces:
Running
Running
File size: 2,044 Bytes
ee37dcf |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 |
import mysql.connector
from mysql.connector import pooling
from .utils import config
class Database:
__connection_pool = None
@classmethod
def init(cls):
try:
cls.__connection_pool = pooling.MySQLConnectionPool(
pool_name="v2boardPool",
pool_size=5,
host=config["host"],
database=config["databaseName"],
user=config["databaseUser"],
password=config["password"]
)
except mysql.connector.Error as e:
print("Error initializing connection pool: {}".format(str(e)))
@classmethod
def get_connection(cls):
if cls.__connection_pool is None:
cls.init()
conn = cls.__connection_pool.get_connection()
return conn
@classmethod
def release_connection(cls, connection):
connection.close()
@classmethod
def close_all_connections(cls):
cls.__connection_pool.close()
@classmethod
def status(cls):
# 获取使用了的连接状态
if cls.__connection_pool is None:
cls.init()
return cls.__connection_pool.get_pool_size()
if __name__ == '__main__':
email = 'hanryr.dios482992030@gmail.com'
with Database.get_connection() as conn:
with conn.cursor() as cur:
print("链接数据库成功")
query = "SELECT balance FROM `v2_user` WHERE email = %(email)s"
params = {'email': email}
cur.execute(query, params)
result = cur.fetchone()
if result is not None:
# 如果email已经存在于数据库中，那么从数据库中获取balance
balance = result[0]
print(balance)
else:
# 如果email不存在于数据库中，提示用户未注册(没必要，因为没法通知到app),所以不做处理
pass
print("数据库操作完成")
|