1、在MetapathFinder中涉及aiomysql模块.find_module只需处理aiomysql模块。
首先忽略其他功能，然后确定需要替换aiomysql的功能。在业务方面，我们通常只需要cursor.execute、cursor.fetchone、cursor.fetchall、cursor.executemany这些主要操作。
2、先cursor.execute的源代码(其他同理)，调用self.nextset的方法。
完成上一个请求的数据，然后合并sql语句，最后通过self._query查询。
实例
importimportlib
importtime
importsys
fromfunctoolsimportwraps
fromtypingimportcast,Any,Callable,Optional,Tuple,TYPE_CHECKING
fromtypesimportModuleType
ifTYPE_CHECKING:
importaiomysql
deffunc_wrapper(func:Callable):
@wraps(func)
asyncdefwrapper(*args,**kwargs)->Any:
start:float=time.time()
func_result:Any=awaitfunc(*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,
"speedtime":end-start
})
returnfunc_result
returncast(Callable,wrapper)
classMetaPathFinder:
@staticmethod
deffind_module(fullname:str,path:Optional[str]=None)->Optional["MetaPathLoader"]:
iffullname=='aiomysql':
#hookk只有aiomysql才能进行
returnMetaPathLoader()
else:
returnNone
classMetaPathLoader:
@staticmethod
defload_module(fullname:str):
iffullnameinsys.modules:
returnsys.modules[fullname]
#防止递归调用
finder:"MetaPathFinder"=sys.meta_path.pop(0)
#导入module
module:ModuleType=importlib.import_module(fullname)
#hoook为_query
module.Cursor._query=func_wrapper(module.Cursor._query)
sys.meta_path.insert(0,finder)
returnmodule
asyncdeftest_mysql()->None:
importaiomysql
pool:aiomysql.Pool=awaitaiomysql.create_pool(
host='127.0.0.1',port=3306,user='root',password='123123',db='mysql'
)
asyncwithpool.acquire()asconn:
asyncwithconn.cursor()ascur:
awaitcur.execute("SELECT42;")
(r,)=awaitcur.fetchone()
assertr==42
pool.close()
awaitpool.wait_closed()
if__name__=='__main__':
sys.meta_path.insert(0,MetaPathFinder())
importasyncio
asyncio.run(test_mysql())
#输出示例:
#可见sql语句和我们输入的一样，db,user,host,port等参数也是如此，还可以知道执行结果和运行时间
#{'sql':'SELECT42;','db':'mysql','user':'root','host':'127.0.0.1','port':3306,'result':(42)，)，'，;speedtime':0.00045609474182128906}以上是python制作探针模块的方法，希望对大家有所帮助。更多Python学习指导：基础教程python基础教程
本文教程操作环境：windows7系统Python 3.9.1，DELL G3电脑。