AsyncBatchProcessor with async tests (pytest-asyncio) -- has anyone been able to make this work?

[hughgrigg]

My project uses Lambda Powertools AsyncBatchProcessor, and pytest-asyncio for tests.

This causes problems during test runs, as AsyncBatchProcessor calls either loop.run_until_complete or asyncio.run, depending on the LAMBDA_TASK_ROOT_ENV env var. https://github.com/aws-powertools/powertools-lambda-python/blob/develop/aws_lambda_powertools/utilities/batch/base.py#L126

This clashes with the event loop that pytest-asyncio is using to run the async tests. When a test calls lambda_handler, it leads to one of these two errors due to the above:

E           RuntimeError: This event loop is already running

E           RuntimeError: asyncio.run() cannot be called from a running event loop

I can workaround this in a single test case by running lambda_handler in a thread pool executor from the test like this:

await asyncio.get_event_loop().run_in_executor(
    None,
    lambda: lambda_handler(
        lambda_event,
        lambda_context,
    ),
)

But when running multiple test cases in the same test process, the first cases passes, but all subsequent test cases hit this error:

RuntimeError: Task <Task pending name='Task-43' coro=<AsyncBatchProcessor._async_process_record()
running at .../python3.11/site-packages/aws_lambda_powertools/utilities/batch/base.py:653>
cb=[gather.<locals>._done_callback() at .../python3.11/asyncio/tasks.py:764]>
got Future <Future pending cb=[Protocol._on_waiter_completed()]>
attached to a different loop

Has anyone else been able to get pytest-asyncio to work with AsyncBatchProcessor?

[leandrodamascena]

Hi @hughgrigg! We currently do not use pytest-asyncio for our tests and this kind of problem can take a few days until we can simulate it.

I will keep this discussion open to update as soon as we have answers and I am tagging @heitorlessa who has more experience in these async tests and the batch utility.

Thanks

[hughgrigg]

Apologies, after some more investigation, it turned out that this was due to how sqlalchemy.ext.asyncio was set up inside the Lambda. As far as I can tell, an async_scoped_session needs to be created and closed inside the async_record_handler, i.e. per record, otherwise various async race condition errors can occur.

TLDR this is not a problem with AsyncBatchProcessor, it's to do with how async database sessions need to be managed inside it.

[heitorlessa]

hey @hughgrigg would be up to make a PR updating the docs to help other customers in the future? :)

It looks like the async race condition is similar to asyncio.gather() that we reported to X-Ray SDK team back then. I'm unsure if you're running Python 3.12 which that code section warned it could cause troubles in getting the running event loop.

context: The issue was that threading.local wasn't asyncio friendly - contextvars is - and that would need some refactoring there to make it work, there are however always caveats.

If you don't have the bandwidth, feel free to post the code snippet and we can get your thoughts on a written text once we have one for this :)

Edit: missed the thread that it mentions it's Batch, not Tracer although sqlalchemy.ext.asyncio is X-Ray SDK IIRC.