上传文件至 templates

templates/public_func.py

@@ -0,0 +1,86 @@
+# -*- 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) "