这篇文章主要介绍python怎么制作探针模块，文中介绍的非常详细，具有一定的参考价值，感兴趣的小伙伴们一定要看完！
成都创新互联一直在为企业提供服务，多年的磨炼，使我们在创意设计，全网整合营销推广到技术研发拥有了开发经验。我们擅长倾听企业需求，挖掘用户对产品需求服务价值，为企业制作有用的创意设计体验。核心团队拥有超过10余年以上行业经验，涵盖创意，策化，开发等专业领域，公司涉及领域有基础互联网服务资阳移动机房、成都app开发、手机移动建站、网页设计、网络整合营销。
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怎么制作探针模块”这篇文章的所有内容，感谢各位的阅读！希望分享的内容对大家有帮助，更多相关知识，欢迎关注创新互联行业资讯频道！
