-
-
Notifications
You must be signed in to change notification settings - Fork 444
Description
I often work with Tortoise ORM in environments where I need to run queries synchronously, such as in scripts or synchronous functions. Currently, QuerySetSingle and AwaitableQuery only support await, which makes integrating with sync code cumbersome. And seeing as all the other query types are subclass's of AwaitableQuery it would make it implement a sync query system quite simply
I propose adding an execute method to both classes that allows running queries in a blocking (sync) or non-blocking (fire-and-forget) manner. Here's a sample implementation:
def execute(self, blocking: bool = True) -> Any | Future[Any] | None:
"""
Execute the QuerySet in a sync context or fire-and-forget.
Parameters
----------
blocking : bool
If True (default), waits for the result and returns it.
If False, schedules execution and returns a Future immediately.
Returns
-------
If blocking=True: the query result.
If blocking=False: concurrent.futures.Future
"""
import asyncio
import nest_asyncio
try:
loop = asyncio.get_event_loop()
except RuntimeError:
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
nest_asyncio.apply(loop=loop)
if blocking:
return loop.run_until_complete(self)
else:
async def _run() -> Coroutine[Any, Any, None]:
return await self
return asyncio.run_coroutine_threadsafe(_run(), loop)This addition would allow seamless use of Tortoise ORM in synchronous code without requiring major refactoring or the use of workarounds like asyncio.run().
Benefits:
- Easier integration with sync scripts or legacy code.
- Provides a simple fire-and-forget option with blocking=False.
- Avoids repetitive boilerplate for creating event loops in sync contexts.
I’d be happy to contribute a PR if the idea is acceptable.
Thanks for considering!