2

I was wondering if someone knew of any magic function that could potentially tell me if the function running has been called synchronously or asynchronously. Long story short, I would like to do the following:

def fsync():
    was_called_async() # >>> False

async fasync():
    was_called_async() # >>> True

fsync()
await fasync()

This might sound a bit weird, and granted, I agree. The reason why is because I'm trying to implement a function similar to functools.singledispatchmethod but for sync and async methods. This requires a decorator to be utilized, and therefore the need for "await-introspection." My current working method here works as expected:

from toolbox.cofunc import cofunc
import asyncio
import time

@cofunc
def hello():
    time.sleep(0.01)
    return "hello sync world!"

@hello.register
async def _():
    await asyncio.sleep(0.01)
    return "hello async world!"

async def main():
    print(hello())        # >>> "hello sync world!"
    print(await hello())  # >>> "hello async world!"

asyncio.run(main())

However, it utilizes the inspect module, and is not something I would want to be dependent on (gets the previous frame and searches for await). If anyone knows of an alternative that would be lovely.

felipe
  • 7,324
  • 2
  • 28
  • 37
  • Your answer definitely deserves an upvote for the sheer creativity of your current solution. :) But unfortunately it is far too clever to work, not only because it uses inspect, but because it requires access to Python source, so it won't work with pyc-only code, and it won't work with Cython. Also, I'm not sure it correctly handles multiple invocations where some are awaited and others are not. – user4815162342 Dec 29 '20 at 19:48
  • And of course, the answer provides the more immediate mundane reasons why it won't work - in Python you await an awaitable object (such as the coroutine object returned by calling an async function, or in asyncio any kind of future), not necessarily a function call. In other words, it's entirely legal to do something like `x = foo(); await x` or stash the coroutine object in a dictionary only to await it later, and so on. – user4815162342 Dec 29 '20 at 19:49
  • A few days later and I think I found the solution: and that is, there is none. No proper solution to this problem. – felipe Jan 03 '21 at 08:03

2 Answers2

2

There just isn't a good way to do what you're looking for.

There's no actual difference between a function being "called synchronously" or being "called asynchronously". Either way, the function is just called.

The closest thing to a difference is what the caller does with the return value, but even looking for await is a bad idea, because things like

await asyncio.gather(yourfunc(), something_else())

or

asyncio.run(yourfunc())

would intuitively be considered "async".

The most reliable option by far is to just have two functions.

user2357112
  • 260,549
  • 28
  • 431
  • 505
  • Your point regarding other methods to call an async function are right on target, and _exactly_ why I was wondering about this question- parsing the ast tree to find every possible way to call an async function (`gather`, `run`, `run_until_complete`, etc.) would be a pain. I appreciate the reply nonetheless! – felipe Dec 29 '20 at 15:06
0

I think you can use asyncio.get_running_loop() to determine if a function is called in an async context or not as the function returns the running loop during async execution and raises a RuntimeError if the loop is not running.

import asyncio


def is_async():
    try:
        asyncio.get_running_loop()
        return True
    except RuntimeError:
        return False


def example():
    if is_async():
        print('I was called by an async function.')
    else:
        print('I was called by a sync function.')


async def async_function():
    print('I am async. is_async():', is_async())
    example()


asyncio.run(async_function())
print()
example()

Console Output:

I am async. is_async(): True
I was called by an async function.

I was called by a sync function.
Jonas Kopka
  • 1
  • 1