Python探針完成調(diào)用庫的數(shù)據(jù)提取

更新時(shí)間：2022年05月11日 17:26:17 作者：??編程學(xué)習(xí)網(wǎng)????

這篇文章主要介紹了Python探針完成調(diào)用庫的數(shù)據(jù)提取，Python中可以通過sys.meta_path來實(shí)現(xiàn)import?hook的功能，下文詳細(xì)資料介紹，需要的小伙伴可以參考一下

1.簡單粗暴的方法--對mysql庫進(jìn)行封裝

要統(tǒng)計(jì)一個(gè)執(zhí)行過程，就需要知道這個(gè)執(zhí)行過程的開始位置和結(jié)束位置，所以最簡單粗暴的方法就是基于要調(diào)用的方法進(jìn)行封裝，在框架調(diào)用MySQL庫和MySQL庫中間實(shí)現(xiàn)一個(gè)中間層, 在中間層完成耗時(shí)統(tǒng)計(jì),如:

# 偽代碼
def my_execute(conn, sql, param):
 # 針對MySql庫的統(tǒng)計(jì)封裝組件
 with MyTracer(conn, sql, param):
     # 以下為正常使用MySql庫的代碼
with conn.cursor as cursor:
?cursor.execute(sql, param)
...

看樣子實(shí)現(xiàn)起來非常不錯(cuò), 而且更改非常方便, 但由于是在最頂層的API上進(jìn)行修改, 其實(shí)是非常不靈活的, 同時(shí)在cursor.execute里會進(jìn)行一些預(yù)操作, 如把sql和param進(jìn)行拼接, 調(diào)用nextset清除當(dāng)前游標(biāo)的數(shù)據(jù)等等。我們最后拿到的數(shù)據(jù)如時(shí)間耗時(shí)也是不準(zhǔn)確的, 同時(shí)也沒辦法得到一些詳細(xì)的元數(shù)據(jù)，如錯(cuò)誤碼等等.

如果要拿到最直接有用的數(shù)據(jù)，就只能去改源代碼, 然后再調(diào)用源代碼了, 但是如果每個(gè)庫都需要改源代碼才能統(tǒng)計(jì), 那也太麻煩了, 好在Python也提供了一些類似探針的接口, 可以通過探針把庫的源碼進(jìn)行替換完成我們的代碼.

2.Python的探針

在Python中可以通過sys.meta_path來實(shí)現(xiàn)import hook的功能, 當(dāng)執(zhí)行 import 相關(guān)操作時(shí), 會根據(jù)sys.meta_path定義的對象對import相關(guān)庫進(jìn)行更改.sys.meta_path中的對象需要實(shí)現(xiàn)一個(gè)find_module方法，這個(gè)find_module方法返回None或一個(gè)實(shí)現(xiàn)了load_module方法的對象, 我們可以通過這個(gè)對象, 針對一些庫在import時(shí), 把相關(guān)的方法進(jìn)行替換, 簡單用法如下,通過hooktime.sleep讓他在sleep的時(shí)候能打印消耗的時(shí)間.

import importlib
import sys
from functools import wraps
def func_wrapper(func):
    """這里通過一個(gè)裝飾器來達(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í)可以看出來在import哪些模塊
        print(f'find module:{path}:{fullname}')
        return MetaPathLoader()
class MetaPathLoader:
    def load_module(self, fullname):
        # import的模塊都會存放在sys.modules里面， 通過判斷可以減少重復(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

3.制作探針模塊

了解完了主要流程后，可以開始制作自己的探針模塊了, 由于示例只涉及到aiomysql模塊, 那么在MetaPathFinder.find_module中需要只對aiomysql模塊進(jìn)行處理, 其他的先忽略. 然后我們需要確定我們要把a(bǔ)iomysql的哪個(gè)功能給替換, 從業(yè)務(wù)上來說, 一般情況下我們只要cursor.execute, cursor.fetchone, cursor.fetchall, cursor.executemany這幾個(gè)主要的操作,所以需要深入cursor看看如何去更改代碼, 后者重載哪個(gè)函數(shù).

先cursor.execute的源碼(cursor.executemanay也類似), 發(fā)現(xiàn)會先調(diào)用self.nextset的方法, 把上個(gè)請求的數(shù)據(jù)先拿完, 再合并sql語句, 最后通過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

綜合上面的分析, 我們只要對核心的方法self._query進(jìn)行重載即可拿到我們要的數(shù)據(jù), 從源碼中我們可以知道, 我們能獲取到傳入self._query的self和sql參數(shù), 根據(jù)self又能獲取到查詢的結(jié)果, 同時(shí)我們通過裝飾器能獲取到運(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]
        # 通過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ā)送到指定的平臺, 然后我們就可以在平臺上看到對應(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)
        # 針對_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}

這個(gè)例子看來很不錯(cuò), 但是需要在調(diào)用的入口處顯式調(diào)用該邏輯, 通常一個(gè)項(xiàng)目可能有幾個(gè)入口, 每個(gè)入口都顯示調(diào)用該邏輯會非常麻煩，而且必須先調(diào)用我們的hook邏輯后才能import，這樣就得訂好引入規(guī)范，不然就可能出現(xiàn)部分地方hook不成功, 如果能把引入hook這個(gè)邏輯安排在解析器啟動后馬上執(zhí)行，就可以完美地解決這個(gè)問題了. 查閱了一翻資料后發(fā)現(xiàn),python解釋器初始化的時(shí)候會自動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存放的代碼如下, 非常簡單，就是引入我們的探針代碼, 并插入到sys.meta_path：

import sys
from hook_aiomysql import MetaPathFinder
sys.meta_path.insert(0, MetaPathFinder())

test_auto_hook.py則是測試代碼：

import asyncio
from hook_aiomysql import test_mysql
asyncio.run(test_mysql())

接下來只要設(shè)置PYTHONPATH并運(yùn)行我們的代碼即可（如果是項(xiàng)目的話一般交由superisor啟動，則可以在配置文件中設(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}

4.直接替換方法

可以看到上面的方法很好的運(yùn)行了，而且可以很方便的嵌入到我們的項(xiàng)目中，但是依賴與sitecustomize.py文件很難讓他抽離成一個(gè)第三方的庫，如果要抽離成第三方的庫就得考慮看看有沒有其他的方法。在上面介紹MetaPathLoader時(shí)說到了sys.module, 在里面通過sys.modules來減少重復(fù)引入：

class MetaPathLoader:
    def load_module(self, fullname):
        # import的模塊都會存放在sys.modules里面， 通過判斷可以減少重復(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è)模塊后，他就會存放在sys.modules，如果是重復(fù)引入，就會直接刷新成最新引入的模塊。上面之所以會考慮到減少重復(fù)import是因?yàn)槲覀儾粫诔绦蜻\(yùn)行時(shí)升級第三方庫的依賴。利用到我們可以不考慮重復(fù)引入同名不同實(shí)現(xiàn)的模塊，以及sys.modules會緩存引入模塊的特點(diǎ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ù), 這里簡單略過"""
# 判斷是否hook過
_IS_HOOK: bool = False
# 存放原來的_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
# 還原到原來的函數(shù)方法
def reset_hook() -> None:
    aiomysql.Cursor._query = _query
    _IS_HOOK = False

代碼簡單明了，接下來跑一跑剛才的測試：

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")

通過測試輸出可以發(fā)現(xiàn)我們的邏輯的正確的, install hook后能出現(xiàn)我們提取的元信息，而reset后則不會打印原信息

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

5.總結(jié)

得益于Python動態(tài)語言的特性，我們可以很容易的為第三方庫實(shí)現(xiàn)鉤子方法，上面說的兩種方法中，第二種方法非常簡單，但在自己項(xiàng)目中最好還是采用第一種方法，因?yàn)镻ython是通過一行一行代碼進(jìn)行掃描執(zhí)行的，第二種方法只能放在入口代碼中，并且要在被hook的對象實(shí)例化之前執(zhí)行，不然就會實(shí)現(xiàn)hook失敗的現(xiàn)象, 而第一種方法除了麻煩外，基本上能躲避所有坑。

到此這篇關(guān)于Python探針完成調(diào)用庫的數(shù)據(jù)提取的文章就介紹到這了,更多相關(guān) Python探針內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: