删除 templates/public_func.py

templates/public_func.py

@@ -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) "