Shulou(Shulou.com)06/03 Report--

This article mainly introduces how to make the probe module of python, which is very detailed and has certain reference value. Friends who are interested must finish reading it!

1. The aiomysql module is involved, and only the aiomysql module needs to be dealt with in MetaPathFinder.find_module.

Others are ignored and then determined that the functionality of aiomysql needs to be replaced. From a business point of view, generally we only need cursor.execute, cursor.fetchone, cursor.fetchall, cursor.executemany these major operations.

2. First, the source code of cursor.execute (the same as others), call the method of self.nextset.

Complete the data of the previous request, then merge the sql statements, and finally query through self._query.

Example

Import importlibimport timeimport sysfrom functools import wraps from typing import cast, Any, Callable, Optional, Tuple, TYPE_CHECKINGfrom types import ModuleTypeif 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 () # according to _ query, the first parameter is self, and the second parameter is sql self: aiomysql.Cursor = args [0] sql: str = args [1] # through self We can get other data 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 # We can send the data to the specified platform according to the agent defined by ourselves Then we can see the corresponding data or monitor it on the platform. # here is just a part of the data printed out 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': # only aiomysql performs hook return MetaPathLoader () else: return None class MetaPathLoader: @ staticmethod def load_module (fullname: str): if fullname in sys.modules: return sys.modules [fullname] # prevent delivery Call finder: "MetaPathFinder" = sys.meta_path.pop (0) # Import module module: ModuleType = importlib.import_module (fullname) # hook module.Cursor._query = func_wrapper (module.Cursor._query) sys.meta_path.insert (0) for _ query 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 ()) # output example: # you can see that the sql statement is the same as what we entered, and so are db, user, host, port and other parameters You can also know the execution result and run time # {'sql':' SELECT 42 ',' db': 'mysql',' user': 'root',' host': '127.0.0.1,' port': 3306, 'result': ((42,),),' speed time': 0.00045609474182128906} above are all the contents of the article "how to make a probe module in python". Thank you for reading! Hope to share the content to help you, more related knowledge, welcome to follow the industry information channel!

Welcome to subscribe "Shulou Technology Information " to get latest news, interesting things and hot topics in the IT industry, and controls the hottest and latest Internet news, technology news and IT industry trends.