1、涉及aiomysql模块,在MetaPathFinder.find_module中只需要处理aiomysql模块。
其他先忽略,然后确定需要替换aiomysql的功能。从业务上来说,一般我们只需要cursor.execute、cursor.fetchone、cursor.fetchall、cursor.executemany这些主要操作。
2、先cursor.execute的源代码(其他同理),调用self.nextset的方法。
完成上一个请求的数据,然后合并sql语句,最后通过self._query查询。
实例
import importlib import time import sys from functools import wraps from typing import cast, Any, Callable, Optional, Tuple, TYPE_CHECKING from types import ModuleType if TYPE_CHECKING: import aiomysql def func_wrapper(func: Callable): @wraps(func) async def wrapper(*args, **kwargs) -> Any: start: float = time.time() func_result: Any = await func(*args, **kwargs) end: float = time.time() # 根据_query可以知道, 第一格参数是self, 第二个参数是sql self: aiomysql.Cursor = args[0] sql: str = args[1] # 通过self,我们可以拿到其他的数据 db: str = self._connection.db user: str = self._connection.user host: str = self._connection.host port: str = self._connection.port execute_result: Tuple[Tuple] = self._rows # 可以根据自己定义的agent把数据发送到指定的平台, 然后我们就可以在平台上看到对应的数据或进行监控了, # 这里只是打印一部分数据出来 print({ "sql": sql, "db": db, "user": user, "host": host, "port": port, "result": execute_result, "speed time": end - start }) return func_result return cast(Callable, wrapper) class MetaPathFinder: @staticmethod def find_module(fullname: str, path: Optional[str] = None) -> Optional["MetaPathLoader"]: if fullname == 'aiomysql': # 只有aiomysql才进行hook return MetaPathLoader() else: return None class MetaPathLoader: @staticmethod def load_module(fullname: str): if fullname in sys.modules: return sys.modules[fullname] # 防止递归调用 finder: "MetaPathFinder" = sys.meta_path.pop(0) # 导入 module module: ModuleType = importlib.import_module(fullname) # 针对_query进行hook module.Cursor._query = func_wrapper(module.Cursor._query) sys.meta_path.insert(0, finder) return module async def test_mysql() -> None: import aiomysql pool: aiomysql.Pool = await aiomysql.create_pool( host='127.0.0.1', port=3306, user='root', password='123123', db='mysql' ) async with pool.acquire() as conn: async with conn.cursor() as cur: await cur.execute("SELECT 42;") (r,) = await cur.fetchone() assert r == 42 pool.close() await pool.wait_closed() if __name__ == '__main__': sys.meta_path.insert(0, MetaPathFinder()) import asyncio asyncio.run(test_mysql()) # 输出示例: # 可以看出sql语句与我们输入的一样, db, user, host, port等参数也是, 还能知道执行的结果和运行时间 # {'sql': 'SELECT 42;', 'db': 'mysql', 'user': 'root', 'host': '127.0.0.1', 'port': 3306, 'result': ((42,),), 'speed time': 0.00045609474182128906}
以上就是python制作探针模块的方法,希望对大家有所帮助。更多Python学习指路:python基础教程
本文教程操作环境:windows7系统、Python 3.9.1,DELL G3电脑。
神龙|纯净稳定代理IP免费测试>>>>>>>>天启|企业级代理IP免费测试>>>>>>>>IPIPGO|全球住宅代理IP免费测试