From ff9205a9f2ad02a78ca751471d3a39832aa74ea3 Mon Sep 17 00:00:00 2001 From: Katsuyuki Takahashi Date: Sun, 28 Sep 2025 21:11:50 +0900 Subject: [PATCH] fix(batch): replace asyncio.get_event_loop() with future-compatible implementation for Python 3.12+ --- aws_lambda_powertools/utilities/batch/base.py | 11 ++++++++++- 1 file changed, 10 insertions(+), 1 deletion(-) diff --git a/aws_lambda_powertools/utilities/batch/base.py b/aws_lambda_powertools/utilities/batch/base.py index d21a329e7c9..f09d464089b 100644 --- a/aws_lambda_powertools/utilities/batch/base.py +++ b/aws_lambda_powertools/utilities/batch/base.py @@ -132,7 +132,16 @@ async def async_process_closure(): # whether we create an event loop (Lambda) or schedule it as usual (non-Lambda) coro = async_process_closure() if os.getenv(constants.LAMBDA_TASK_ROOT_ENV): - loop = asyncio.get_event_loop() # NOTE: this might return an error starting in Python 3.12 in a few years + # In Python 3.12+, asyncio.get_event_loop() raises a warning when called without a running loop + # In Python 3.14+, it will raise an exception if no event loop is running + # Use try/except to handle both the current behavior and future Python versions + try: + loop = asyncio.get_running_loop() + except RuntimeError: + # No running loop, create a new one for Lambda environment + loop = asyncio.new_event_loop() + asyncio.set_event_loop(loop) + task_instance = loop.create_task(coro) return loop.run_until_complete(task_instance)