如何通过Python实时监控本地与SQL Server的SQL请求并提取参数？

方案一：修改Python应用的PYODBC连接，嵌入请求拦截逻辑 #

这是最直接且服务器负载最低的方案，完全在本地Python应用层面处理，无需SQL Server端额外配置：

• 封装PYODBC的execute/executemany方法，执行SQL前先检查语句内容
• 针对匹配exec StoredProc1格式的语句，用正则提取参数并触发后续逻辑：

  import re
  import pyodbc
  
  class MonitoredConnection:
      def __init__(self, conn_str):
          self.conn = pyodbc.connect(conn_str)
      
      def execute(self, sql, *args):
          # 匹配目标存储过程调用格式
          match = re.match(r'^\s*exec\s+StoredProc1\s+(\d+)\s*,\s*(\d+)\s*,\s*(\d+)\s*$', sql, re.IGNORECASE)
          if match:
              param1 = int(match.group(1))
              param2 = int(match.group(2))
              param3 = int(match.group(3))
              # 触发额外数据展示
              self._show_extra_data(param1, param2, param3)
          return self.conn.execute(sql, *args)
      
      def _show_extra_data(self, p1, p2, p3):
          # 这里实现你的额外数据展示功能
          print(f"触发额外数据：参数1={p1}, 参数2={p2}, 参数3={p3}")
  
  # 使用示例
  conn = MonitoredConnection("你的PYODBC连接字符串")
  cursor = conn.execute("exec StoredProc1 4, 500, 100")
  

• 优势：完全不占用SQL Server资源，仅监控自身应用请求，实时性拉满
• 注意：若存储过程调用有命名参数等其他格式，需调整正则表达式覆盖所有场景

方案二：扩展事件（Extended Events）写入内存目标，本地Python实时读取 #

若必须从SQL Server端获取事件，推荐用内存目标替代文件或表，避免IO开销且支持实时读取：

1. 创建扩展事件会话：

   CREATE EVENT SESSION [TrackSP1Calls] ON SERVER 
   ADD EVENT sqlserver.rpc_completed(
       ACTION(sqlserver.session_id, sqlserver.client_app_name)
       WHERE (sqlserver.like_i_sql_unicode_string(sqlserver.sql_text, N'%exec StoredProc1%'))
   ),
   ADD EVENT sqlserver.sql_batch_completed(
       ACTION(sqlserver.session_id, sqlserver.client_app_name)
       WHERE (sqlserver.like_i_sql_unicode_string(sqlserver.sql_text, N'%exec StoredProc1%'))
   )
   ADD TARGET package0.ring_buffer(SET max_memory=4096)
   WITH (STARTUP_STATE=OFF)
   

   • ring_buffer为内存目标，数据存在内存中，无IO负载
   • 过滤条件仅跟踪目标存储过程，减少事件量
   • 捕获session_id和client_app_name用于区分不同用户请求
2. 本地Python定时查询扩展事件数据：

   import pyodbc
   import xml.etree.ElementTree as ET
   import time
   import re
   
   def get_sp_events():
       conn = pyodbc.connect("你的SQL Server连接字符串")
       cursor = conn.execute("""
           SELECT CAST(target_data AS XML) 
           FROM sys.dm_xe_session_targets st
           JOIN sys.dm_xe_sessions s ON s.address = st.event_session_address
           WHERE s.name = 'TrackSP1Calls' AND st.target_name = 'ring_buffer'
       """)
       xml_data = cursor.fetchone()[0]
       conn.close()
       
       events = []
       if xml_data is not None:
           for event in xml_data.findall('.//event'):
               sql_text = event.find('.//data[@name="sql_text"]/value').text
               session_id = event.find('.//action[@name="session_id"]/value').text
               app_name = event.find('.//action[@name="client_app_name"]/value').text
               # 提取参数
               match = re.match(r'^\s*exec\s+StoredProc1\s+(\d+)\s*,\s*(\d+)\s*,\s*(\d+)\s*$', sql_text, re.IGNORECASE)
               if match:
                   events.append({
                       'session_id': session_id,
                       'app_name': app_name,
                       'params': (int(match.group(1)), int(match.group(2)), int(match.group(3)))
                   })
       return events
   
   # 定时轮询（可根据实时性调整间隔）
   while True:
       events = get_sp_events()
       for event in events:
           # 校验是否为当前应用的请求（需在连接时设置client_app_name）
           if event['app_name'] == 'MyLocalPythonApp':
               p1, p2, p3 = event['params']
               print(f"触发额外数据：参数1={p1}, 参数2={p2}, 参数3={p3}")
       time.sleep(1)
   

   • 连接时设置client_app_name：在PYODBC连接字符串中加入APP=MyLocalPythonApp
   • 优势：服务器负载极低，内存目标资源占用少；能准确捕获所有目标存储过程调用

方案三：扩展事件写入表的配置方法（备选） #

若一定要用表存储事件，可按以下步骤配置（实时性较差，有IO开销）：

1. 创建事件会话使用event_file目标：

   CREATE EVENT SESSION [TrackSP1CallsToFile] ON SERVER 
   ADD EVENT sqlserver.rpc_completed(
       ACTION(sqlserver.session_id, sqlserver.client_app_name)
       WHERE (sqlserver.like_i_sql_unicode_string(sqlserver.sql_text, N'%exec StoredProc1%'))
   ),
   ADD EVENT sqlserver.sql_batch_completed(
       ACTION(sqlserver.session_id, sqlserver.client_app_name)
       WHERE (sqlserver.like_i_sql_unicode_string(sqlserver.sql_text, N'%exec StoredProc1%'))
   )
   ADD TARGET package0.event_file(SET filename=N'C:\SQLServerEvents\TrackSP1Calls.xel', max_file_size=10)
   WITH (STARTUP_STATE=OFF)
   

2. 创建存储事件的表：

   CREATE TABLE SP1EventLog (
       event_time DATETIME,
       session_id INT,
       app_name NVARCHAR(128),
       sql_text NVARCHAR(MAX)
   )
   

3. 定期导入xel文件数据到表中（可通过SQL Agent作业或Python脚本）：

   INSERT INTO SP1EventLog (event_time, session_id, app_name, sql_text)
   SELECT 
       DATEADD(mi, DATEDIFF(mi, GETUTCDATE(), GETDATE()), xevent.value('(@timestamp)[1]', 'DATETIME')) AS event_time,
       xevent.value('(action[@name="session_id"]/value)[1]', 'INT') AS session_id,
       xevent.value('(action[@name="client_app_name"]/value)[1]', 'NVARCHAR(128)') AS app_name,
       xevent.value('(data[@name="sql_text"]/value)[1]', 'NVARCHAR(MAX)') AS sql_text
   FROM (
       SELECT CAST(event_data AS XML) AS xevent
       FROM sys.fn_xe_file_target_read_file('C:\SQLServerEvents\TrackSP1Calls*.xel', NULL, NULL, NULL)
   ) AS events
   WHERE NOT EXISTS (
       SELECT 1 FROM SP1EventLog 
       WHERE event_time = DATEADD(mi, DATEDIFF(mi, GETUTCDATE(), GETDATE()), xevent.value('(@timestamp)[1]', 'DATETIME'))
       AND session_id = xevent.value('(action[@name="session_id"]/value)[1]', 'INT')
   )
   

不推荐方案：监控网络数据包 #

• 需解析SQL Server的TDS协议，实现复杂，易受加密、版本兼容问题影响
• 无法准确区分自身应用与其他用户的请求，误判率高，且占用本地系统资源
• 完全没有必要，前两个方案已能满足需求且更可靠

内容的提问来源于stack exchange，提问作者Lzypenguin