I am experimenting with ways to record and replay SQL queries send by asyncpg (similar to VCR.py).

I do not know the internals of asyncpg well and therefore would appreciate feedback and pointers.

To record and replay, following steps have to be done (roughly):

1. Record the SQL query and the results
2. Next time the same SQL query is executed, do not interact with the database at all (e.g. no connection and no execution). Instead, replay the recorded results.

Step 1

After reading the source code of asyncpg a bit, I decided to wrap the Connection._execute function like this:

(As context: Below record function is intended to be used as decorator around a test function.)

def record(func):
 @wraps(func)
 async def wrapper(*args, **kwargs):
 execture_original = asyncpg.connection.Connection._execute
 @wraps(execture_original)
 async def execute_wrapper(self, *execute_args, **execute_kwargs):
 # TODO: Record (serialize to disk) execute parameters and result
 return await execture_original(self, *execute_args, **execute_kwargs)
 asyncpg.connection.Connection._execute = execute_wrapper
 return await func(*args, **kwargs)
 return wrapper

Do you find this sensible? Any scenario where this approach would not work?

Step 2

Here I mainly have the following questions:

Which is the best way to mock every possible interaction with the database and create a asyncpg record object from the recorded results?

Any thoughts about any of the points above are highly appreciated. Thanks!