87 lines
2.5 KiB
Python
87 lines
2.5 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""
|
|
@Project : data_upload_download
|
|
@File : public_func.py
|
|
@IDE : PyCharm
|
|
@Author : liu jian jiang
|
|
@Date : 2025/6/17 09:31
|
|
"""
|
|
import sys
|
|
import pymysql
|
|
from typing import Optional
|
|
from dbutils.pooled_db import PooledDB
|
|
|
|
|
|
class MySQLPool:
|
|
def __init__(self):
|
|
self.config = {
|
|
'host': '47.238.96.231',
|
|
'user': 'root',
|
|
'password': 'gin_demo',
|
|
'database': 'gin_demo',
|
|
'charset': 'utf8mb4'
|
|
}
|
|
# if sys.platform == 'win32':
|
|
# self.config = {
|
|
# 'host': '127.0.0.1',
|
|
# 'user': 'root',
|
|
# 'password': '123456',
|
|
# 'database': 'person_test',
|
|
# 'charset': 'utf8mb4'
|
|
# }
|
|
|
|
self.pool = PooledDB(
|
|
creator=pymysql,
|
|
mincached=2, # 初始空闲连接数
|
|
maxcached=5, # 最大空闲连接数
|
|
maxconnections=20, # 最大连接数
|
|
blocking=True, # 连接数满时阻塞等待
|
|
**self.config
|
|
)
|
|
|
|
def get_connection(self):
|
|
return self.pool.connection()
|
|
|
|
def execute_query(self, sql: str, params: Optional[tuple] = None):
|
|
conn = self.get_connection()
|
|
try:
|
|
with conn.cursor() as cursor:
|
|
cursor.execute(sql, params or ())
|
|
return cursor.fetchall()
|
|
finally:
|
|
conn.close()
|
|
|
|
def execute_single(self, sql: str, params: tuple = None) -> int:
|
|
"""执行单条SQL并返回影响行数"""
|
|
conn = self.get_connection()
|
|
try:
|
|
with conn.cursor() as cursor:
|
|
affected = cursor.execute(sql, params or ())
|
|
conn.commit()
|
|
return affected
|
|
except Exception as e:
|
|
conn.rollback()
|
|
raise e
|
|
finally:
|
|
conn.close()
|
|
|
|
def execute_update(self, sql: str, params: Optional[tuple] = None):
|
|
conn = self.get_connection()
|
|
try:
|
|
with conn.cursor() as cursor:
|
|
affected = cursor.execute(sql, params or ())
|
|
conn.commit()
|
|
return affected
|
|
except Exception as e:
|
|
conn.rollback()
|
|
raise e
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
# 使用示例
|
|
if __name__ == "__main__":
|
|
pool = MySQLPool()
|
|
# 查询示例
|
|
# sql_str = "INSERT INTO device_info(android_id, device_info) "
|