您好,登錄后才能下訂單哦!
今天小編給大家分享一下Python探針如何完成調(diào)用庫(kù)的數(shù)據(jù)提取的相關(guān)知識(shí)點(diǎn),內(nèi)容詳細(xì),邏輯清晰,相信大部分人都還太了解這方面的知識(shí),所以分享這篇文章給大家參考一下,希望大家閱讀完這篇文章后有所收獲,下面我們一起來(lái)了解一下吧。
要統(tǒng)計(jì)一個(gè)執(zhí)行過(guò)程, 就需要知道這個(gè)執(zhí)行過(guò)程的開始位置和結(jié)束位置, 所以最簡(jiǎn)單粗暴的方法就是基于要調(diào)用的方法進(jìn)行封裝,在框架調(diào)用MySQL庫(kù)和MySQL庫(kù)中間實(shí)現(xiàn)一個(gè)中間層, 在中間層完成耗時(shí)統(tǒng)計(jì),如:
# 偽代碼 def my_execute(conn, sql, param): # 針對(duì)MySql庫(kù)的統(tǒng)計(jì)封裝組件 with MyTracer(conn, sql, param): # 以下為正常使用MySql庫(kù)的代碼 with conn.cursor as cursor: cursor.execute(sql, param) ...
看樣子實(shí)現(xiàn)起來(lái)非常不錯(cuò), 而且更改非常方便, 但由于是在最頂層的API上進(jìn)行修改, 其實(shí)是非常不靈活的, 同時(shí)在cursor.execute里會(huì)進(jìn)行一些預(yù)操作, 如把sql和param進(jìn)行拼接, 調(diào)用nextset清除當(dāng)前游標(biāo)的數(shù)據(jù)等等。我們最后拿到的數(shù)據(jù)如時(shí)間耗時(shí)也是不準(zhǔn)確的, 同時(shí)也沒(méi)辦法得到一些詳細(xì)的元數(shù)據(jù), 如錯(cuò)誤碼等等.
如果要拿到最直接有用的數(shù)據(jù),就只能去改源代碼, 然后再調(diào)用源代碼了, 但是如果每個(gè)庫(kù)都需要改源代碼才能統(tǒng)計(jì), 那也太麻煩了, 好在Python也提供了一些類似探針的接口, 可以通過(guò)探針把庫(kù)的源碼進(jìn)行替換完成我們的代碼.
在Python中可以通過(guò)sys.meta_path來(lái)實(shí)現(xiàn)import hook的功能, 當(dāng)執(zhí)行 import 相關(guān)操作時(shí), 會(huì)根據(jù)sys.meta_path定義的對(duì)象對(duì)import相關(guān)庫(kù)進(jìn)行更改.sys.meta_path中的對(duì)象需要實(shí)現(xiàn)一個(gè)find_module方法, 這個(gè)find_module方法返回None或一個(gè)實(shí)現(xiàn)了load_module方法的對(duì)象, 我們可以通過(guò)這個(gè)對(duì)象, 針對(duì)一些庫(kù)在import時(shí), 把相關(guān)的方法進(jìn)行替換, 簡(jiǎn)單用法如下,通過(guò)hooktime.sleep讓他在sleep的時(shí)候能打印消耗的時(shí)間.
import importlib import sys from functools import wraps def func_wrapper(func): """這里通過(guò)一個(gè)裝飾器來(lái)達(dá)到貍貓換太子和獲取數(shù)據(jù)的效果""" @wraps(func) def wrapper(*args, **kwargs): # 記錄開始時(shí)間 start = time.time() result = func(*args, **kwargs) # 統(tǒng)計(jì)消耗時(shí)間 end = time.time() print(f"speed time:{end - start}") return result return wrapper class MetaPathFinder: def find_module(self, fullname, path=None): # 執(zhí)行時(shí)可以看出來(lái)在import哪些模塊 print(f'find module:{path}:{fullname}') return MetaPathLoader() class MetaPathLoader: def load_module(self, fullname): # import的模塊都會(huì)存放在sys.modules里面, 通過(guò)判斷可以減少重復(fù)import if fullname in sys.modules: return sys.modules[fullname] # 防止遞歸調(diào)用 finder = sys.meta_path.pop(0) # 導(dǎo)入 module module = importlib.import_module(fullname) if fullname == 'time': # 替換函數(shù) module.sleep = func_wrapper(module.sleep) sys.meta_path.insert(0, finder) return module sys.meta_path.insert(0, MetaPathFinder()) if __name__ == '__main__': import time time.sleep(1) # 輸出示例: # find module:datetime # find module:time # load module:time # find module:math # find module:_datetime # speed time:1.00073385238647468
了解完了主要流程后, 可以開始制作自己的探針模塊了, 由于示例只涉及到aiomysql模塊, 那么在MetaPathFinder.find_module中需要只對(duì)aiomysql模塊進(jìn)行處理, 其他的先忽略. 然后我們需要確定我們要把a(bǔ)iomysql的哪個(gè)功能給替換, 從業(yè)務(wù)上來(lái)說(shuō), 一般情況下我們只要cursor.execute, cursor.fetchone, cursor.fetchall, cursor.executemany這幾個(gè)主要的操作,所以需要深入cursor看看如何去更改代碼, 后者重載哪個(gè)函數(shù).
先cursor.execute的源碼(cursor.executemanay也類似), 發(fā)現(xiàn)會(huì)先調(diào)用self.nextset的方法, 把上個(gè)請(qǐng)求的數(shù)據(jù)先拿完, 再合并sql語(yǔ)句, 最后通過(guò)self._query進(jìn)行查詢:
async def execute(self, query, args=None): """Executes the given operation Executes the given operation substituting any markers with the given parameters. For example, getting all rows where id is 5: cursor.execute("SELECT * FROM t1 WHERE id = %s", (5,)) :param query: ``str`` sql statement :param args: ``tuple`` or ``list`` of arguments for sql query :returns: ``int``, number of rows that has been produced of affected """ conn = self._get_db() while (await self.nextset()): pass if args is not None: query = query % self._escape_args(args, conn) await self._query(query) self._executed = query if self._echo: logger.info(query) logger.info("%r", args) return self._rowcount
再看cursor.fetchone的源碼(cursor.fetchall也類似), 發(fā)現(xiàn)其實(shí)是從緩存中獲取數(shù)據(jù),
這些數(shù)據(jù)在執(zhí)行cursor.execute中就已經(jīng)獲取了:
def fetchone(self): """Fetch the next row """ self._check_executed() fut = self._loop.create_future() if self._rows is None or self._rownumber >= len(self._rows): fut.set_result(None) return fut result = self._rows[self._rownumber] self._rownumber += 1 fut = self._loop.create_future() fut.set_result(result) return fut
綜合上面的分析, 我們只要對(duì)核心的方法self._query進(jìn)行重載即可拿到我們要的數(shù)據(jù), 從源碼中我們可以知道, 我們能獲取到傳入self._query的self和sql參數(shù), 根據(jù)self又能獲取到查詢的結(jié)果, 同時(shí)我們通過(guò)裝飾器能獲取到運(yùn)行的時(shí)間, 要的數(shù)據(jù)基本都到齊了,
按照思路修改后的代碼如下:
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] # 通過(guò)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ù)出來(lái) 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語(yǔ)句與我們輸入的一樣, 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}
這個(gè)例子看來(lái)很不錯(cuò), 但是需要在調(diào)用的入口處顯式調(diào)用該邏輯, 通常一個(gè)項(xiàng)目可能有幾個(gè)入口, 每個(gè)入口都顯示調(diào)用該邏輯會(huì)非常麻煩, 而且必須先調(diào)用我們的hook邏輯后才能import, 這樣就得訂好引入規(guī)范, 不然就可能出現(xiàn)部分地方hook不成功, 如果能把引入hook這個(gè)邏輯安排在解析器啟動(dòng)后馬上執(zhí)行, 就可以完美地解決這個(gè)問(wèn)題了. 查閱了一翻資料后發(fā)現(xiàn),python解釋器初始化的時(shí)候會(huì)自動(dòng)import PYTHONPATH下存在的sitecustomize和usercustomize模塊, 我們只要?jiǎng)?chuàng)建該模塊, 并在模塊里面寫入我們的 替換函數(shù)即可。
. ├── __init__.py ├── hook_aiomysql.py ├── sitecustomize.py └── test_auto_hook.py
hook_aiomysql.py是我們制作探針的代碼為例子, 而sitecustomize.py存放的代碼如下, 非常簡(jiǎn)單, 就是引入我們的探針代碼, 并插入到sys.meta_path:
import sys from hook_aiomysql import MetaPathFinder sys.meta_path.insert(0, MetaPathFinder())
test_auto_hook.py則是測(cè)試代碼:
import asyncio from hook_aiomysql import test_mysql asyncio.run(test_mysql())
接下來(lái)只要設(shè)置PYTHONPATH并運(yùn)行我們的代碼即可(如果是項(xiàng)目的話一般交由superisor啟動(dòng),則可以在配置文件中設(shè)置好PYTHONPATH):
(.venv) ? python_hook git:(master) ? export PYTHONPATH=. (.venv) ? python_hook git:(master) ? python test_auto_hook.py {'sql': 'SELECT 42;', 'db': 'mysql', 'user': 'root', 'host': '127.0.0.1', 'port': 3306, 'result': ((42,),), 'speed time': 0.000213623046875}
可以看到上面的方法很好的運(yùn)行了, 而且可以很方便的嵌入到我們的項(xiàng)目中, 但是依賴與sitecustomize.py文件很難讓他抽離成一個(gè)第三方的庫(kù), 如果要抽離成第三方的庫(kù)就得考慮看看有沒(méi)有其他的方法。在上面介紹MetaPathLoader時(shí)說(shuō)到了sys.module, 在里面通過(guò)sys.modules來(lái)減少重復(fù)引入:
class MetaPathLoader: def load_module(self, fullname): # import的模塊都會(huì)存放在sys.modules里面, 通過(guò)判斷可以減少重復(fù)import if fullname in sys.modules: return sys.modules[fullname] # 防止遞歸調(diào)用 finder = sys.meta_path.pop(0) # 導(dǎo)入 module module = importlib.import_module(fullname) if fullname == 'time': # 替換函數(shù) module.sleep = func_wrapper(module.sleep) sys.meta_path.insert(0, finder) return module
這個(gè)減少重復(fù)引入的原理是, 每次引入一個(gè)模塊后, 他就會(huì)存放在sys.modules, 如果是重復(fù)引入, 就會(huì)直接刷新成最新引入的模塊。上面之所以會(huì)考慮到減少重復(fù)import是因?yàn)槲覀儾粫?huì)在程序運(yùn)行時(shí)升級(jí)第三方庫(kù)的依賴。利用到我們可以不考慮重復(fù)引入同名不同實(shí)現(xiàn)的模塊, 以及sys.modules會(huì)緩存引入模塊的特點(diǎn), 我們可以把上面的邏輯簡(jiǎn)化成引入模塊->替換當(dāng)前模塊方法為我們修改的hook方法。
import time from functools import wraps from typing import Any, Callable, Tuple, cast import aiomysql def func_wrapper(func: Callable): """和上面一樣的封裝函數(shù), 這里簡(jiǎn)單略過(guò)""" # 判斷是否hook過(guò) _IS_HOOK: bool = False # 存放原來(lái)的_query _query: Callable = aiomysql.Cursor._query # hook函數(shù) def install_hook() -> None: _IS_HOOK = False if _IS_HOOK: return aiomysql.Cursor._query = func_wrapper(aiomysql.Cursor._query) _IS_HOOK = True # 還原到原來(lái)的函數(shù)方法 def reset_hook() -> None: aiomysql.Cursor._query = _query _IS_HOOK = False
代碼簡(jiǎn)單明了,接下來(lái)跑一跑剛才的測(cè)試:
import asyncio import aiomysql from demo import install_hook, reset_hook async def test_mysql() -> None: pool: aiomysql.Pool = await aiomysql.create_pool( host='127.0.0.1', port=3306, user='root', password='', 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() print("install hook") install_hook() asyncio.run(test_mysql()) print("reset hook") reset_hook() asyncio.run(test_mysql()) print("end")
通過(guò)測(cè)試輸出可以發(fā)現(xiàn)我們的邏輯的正確的, install hook后能出現(xiàn)我們提取的元信息, 而reset后則不會(huì)打印原信息
install hook {'sql': 'SELECT 42;', 'db': 'mysql', 'user': 'root', 'host': '127.0.0.1', 'port': 3306, 'result': ((42,),), 'speed time': 0.000347137451171875} reset hook end
以上就是“Python探針如何完成調(diào)用庫(kù)的數(shù)據(jù)提取”這篇文章的所有內(nèi)容,感謝各位的閱讀!相信大家閱讀完這篇文章都有很大的收獲,小編每天都會(huì)為大家更新不同的知識(shí),如果還想學(xué)習(xí)更多的知識(shí),請(qǐng)關(guān)注億速云行業(yè)資訊頻道。
免責(zé)聲明:本站發(fā)布的內(nèi)容(圖片、視頻和文字)以原創(chuàng)、轉(zhuǎn)載和分享為主,文章觀點(diǎn)不代表本網(wǎng)站立場(chǎng),如果涉及侵權(quán)請(qǐng)聯(lián)系站長(zhǎng)郵箱:is@yisu.com進(jìn)行舉報(bào),并提供相關(guān)證據(jù),一經(jīng)查實(shí),將立刻刪除涉嫌侵權(quán)內(nèi)容。