游标分批获取数据库数据方式单元测试

repositoryZN/utils/rdbmsUtil/unitTest.py

@@ -36,19 +36,41 @@ def access_sqlite():
 
 
 def access_tidb():
-    print('access TiDB')
-    DatabaseUtil(url='mysql+pymysql://root:123456@192.168.50.234:4000/energy',
-                        pool_size=5,
-                        max_overflow=3,
-                        connect_timeout=1800)
-    db = DatabaseUtil('mysql+pymysql://root:123456@192.168.50.234:4000/energy')
+    print('access MySQL/TiDB')
+    # DatabaseUtil(url='mysql+pymysql://root:123456@192.168.50.234:4000/energy',
+    #                     pool_size=5,
+    #                     max_overflow=3,
+    #                     connect_timeout=1800)
+    # db = DatabaseUtil('mysql+pymysql://root:123456@192.168.50.234:4000/energy')
 
+    db = DatabaseUtil('mysql+pymysql://root:admin123456@192.168.50.235:30306/energy_data_prod')
+
+    query_text = f"""SELECT * FROM analysis_result  """
+    query_text=f"""SELECT  wind_turbine_number, time_stamp, wind_velocity, active_power, yaw_error1 FROM `WOF055000062_second` WHERE wind_turbine_number IN ('WOG02325') AND time_stamp >= '2026-02-01 00:00:00' AND time_stamp <= '2026-04-26 00:00:00' AND lab in (0,1,2,3,4)"""
+
+    chunks = []
     with db.session_scope() as session:
-        query_text = f"""SELECT * FROM analysis_result  """
-        query_result = session.execute(text(query_text)).fetchall()
+        # query_result = session.execute(text(query_text)).fetchall()
+
+        # dataFrame = pd.DataFrame(query_result)
+
+        conn=session.connection()
+        
+        result=conn.execution_options(stream_results=True).execute(text(query_text))
 
-        dataFrame = pd.DataFrame(query_result)
+        # 每次取 100000 行，可根据实际情况调整
+        while True:
+            rows = result.fetchmany(100000)
+            if not rows:
+                break
+            chunks.append(pd.DataFrame(rows))
 
-        print(dataFrame.head(5))
+    if not chunks:
+        return pd.DataFrame()
+    
+    dataFrame=pd.concat(chunks, ignore_index=True)        
+    
+    print(len(dataFrame))
+    print(dataFrame.head(5))
 
 access_tidb()