syncasync.py

# Example of implementation supporting async/sync calls.
# As a user you don't have to worry about anything,
# you can either use `async cache.get` or `cache.get` and it will work
# as expected where expected means the following:
# - If you are running inside a loop, `cache.get` will return a coroutine
#   which can be awaited, passed to ensure_future, etc. Obviously
#   `await cache.get` then works
# - If you are running outside a loop context, `cache.get` will perform
#   the underlying calls inside a `loop.run_until_complete` call which
#   will result in a blocking call until the result is retrieved.

# For this to work correctly, there is only ONE must: The methods of the
# first class citizen MUST be synchronous (return a coroutine if its async
# or return the result if its not)

import asyncio


async def some_extra_stuff():
    print("Extra...")
    await asyncio.sleep(1)
    print("Extra done")


class AsyncConnector:
    """
    All methods where you want sync/async compatibility must be sync. Only
    do that for the ones that can block your loop
    
    This class can act both as first or second class citizen. If you don't
    want to expose sync API, you can leave all the methods async as if you would
    code a normal asyncio class
    
    This is done this way to prove that this way of doing sync/async compat
    is compatible between classes using the same approach
    """
    
    # If this is a second citizen class, you can leave this method async without all
    # the overhead of sync_or_async_run
    def _get(self):  
        print("Getting...")
        return sync_or_async_run(some_extra_stuff)
        print("Got you!")


def sync_or_async_run(fn):
    if not asyncio.iscoroutinefunction(fn):  # If what we receive its not a coroutine just return
        return fn()
    if asyncio.Task.current_task():  # If we are inside a task, it means we are in a loop
        return fn()                  # so we return a coroutine that will be awaited outside
    return asyncio.get_event_loop().run_until_complete(fn())


class Cache(AsyncConnector):

    def get(self):
        return sync_or_async_run(self._get)


async def asyncmain():
    c = Cache()
    await c.get()


def syncmain():
    c = Cache()
    c.get()


if __name__ == "__main__":
    print("Doing it asynchronously:")
    loop = asyncio.get_event_loop()
    loop.run_until_complete(asyncmain())
    print("Doing it synchronously:")
    syncmain()

thanks for this example! I might try something similar to your sync_or_async_run function. Here's a library where I'm try to do something similar and maintain both a sync and async interface. I found myself repeating quite a bit of code and was looking for a way to reduce it further but have been unsuccessful thus far: https://github.com/AlJohri/craigslist#api