diff --git a/templates/public_func.py b/templates/public_func.py deleted file mode 100644 index e7e000e..0000000 --- a/templates/public_func.py +++ /dev/null @@ -1,86 +0,0 @@ -# -*- 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) "