建站
咨詢
售后
建站咨詢
新聞中心
這里有您想知道的互聯(lián)網(wǎng)營銷解決方案
創(chuàng)新互聯(lián)Python教程：python如何制作探針模塊
1、涉及aiomysql模塊，在MetaPathFinder.find_module中只需要處理aiomysql模塊。
成都創(chuàng)新互聯(lián)科技有限公司專業(yè)互聯(lián)網(wǎng)基礎(chǔ)服務(wù)商，為您提供川西大數(shù)據(jù)中心，高防物理服務(wù)器租用，成都IDC機(jī)房托管，成都主機(jī)托管等互聯(lián)網(wǎng)服務(wù)。
其他先忽略，然后確定需要替換aiomysql的功能。從業(yè)務(wù)上來說，一般我們只需要cursor.execute、cursor.fetchone、cursor.fetchall、cursor.executemany這些主要操作。
2、先cursor.execute的源代碼(其他同理)，調(diào)用self.nextset的方法。
完成上一個(gè)請(qǐng)求的數(shù)據(jù)，然后合并sql語句，最后通過self._query查詢。
實(shí)例
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()
# 根據(jù)_query可以知道, 第一格參數(shù)是self, 第二個(gè)參數(shù)是sql
self: aiomysql.Cursor = args[0]
sql: str = args[1]
# 通過self,我們可以拿到其他的數(shù)據(jù)
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
# 可以根據(jù)自己定義的agent把數(shù)據(jù)發(fā)送到指定的平臺(tái), 然后我們就可以在平臺(tái)上看到對(duì)應(yīng)的數(shù)據(jù)或進(jìn)行監(jiān)控了,
# 這里只是打印一部分?jǐn)?shù)據(jù)出來
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才進(jìn)行hook
return MetaPathLoader()
else:
return None
class MetaPathLoader:
@staticmethod
def load_module(fullname: str):
if fullname in sys.modules:
return sys.modules[fullname]
# 防止遞歸調(diào)用
finder: "MetaPathFinder" = sys.meta_path.pop(0)
# 導(dǎo)入 module
module: ModuleType = importlib.import_module(fullname)
# 針對(duì)_query進(jìn)行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等參數(shù)也是, 還能知道執(zhí)行的結(jié)果和運(yùn)行時(shí)間
# {'sql': 'SELECT 42;', 'db': 'mysql', 'user': 'root', 'host': '127.0.0.1', 'port': 3306, 'result': ((42,),), 'speed time': 0.00045609474182128906}以上就是python制作探針模塊的方法，希望對(duì)大家有所幫助。更多Python學(xué)習(xí)指路：創(chuàng)新互聯(lián)python教程
本文教程操作環(huán)境：windows7系統(tǒng)、Python 3.9.1，DELL G3電腦。
分享文章：創(chuàng)新互聯(lián)Python教程：python如何制作探針模塊
URL分享：http://www.fisionsoft.com.cn/article/dhssgsd.html
