async recipe function for singleton decorator

[Amol-Gupta]

Hi
How can I use an async function as recipe. I am trying to inject a authenticated client in my code and authentication is an async function.

[remimd]

Hi, this is a topic I've already tried to implement and it's really not simple to execute an asynchronous function, in a synchronous context.
I'll try to look into it, but I can't guarantee anything. If anyone has any implementation ideas, I'd be interested.

[remimd]

In most cases, the problem can be solved in another way. And in the meantime, I recommend using another strategy.
For example, with a setup method:

from injection import inject, singleton

@singleton
class Client:
    async def setup(self) -> None:
        """ Async setup implementation """

@inject
async def main(client: Client) -> None:
    await client.setup()
    # ...

[Amol-Gupta]

Thanks for the response
So you are suggesting that I do the setup after the object is injected. The Idea is to have a singleton object which is only setup once and then shared across multiple places in an application. I am not sure which place will fetch the object first so at each place I will inject the object, I will check if not authenticated then authenticate()

@inject
async def main(client: Client) -> None:
    if( not client.is_authenticated):
        await client.setup()

I am still learning about asyncio and the issues boils down to mixing blocking and nonblocking code. Constuctor/___init__ are supposed to be blocking code. They may be called from asyncio loop or from a blocking code i.e. loop could be running or there may be no loop. In my case __init__, a blocking function needs to call a async function. This is something discouraged in a well designed asyncio application ref. . Finally I discovered nest_asyncio package ref. Using these 2 resources I was able to make the following hack work.

import asyncio
import nest_asyncio

def call(coro):
    try:
        loop = asyncio.get_running_loop()
        if loop.is_running():
            nest_asyncio.apply(loop)
    except RuntimeError:
        return asyncio.run(coro)
    else:
        return loop.run_until_complete(coro)

This allows me to call auth function only once in the constructor.

[remimd]

If you use the setup method. It must be called in the main function, which is normally called only once. If you're using a framework like FastAPI, the setup method must be called in the function that manages the lifespan.

Coming back to asyncio, as far as I know, the best way to call an asynchronous function is to do it inside another asynchronous function.

Another solution that may be of interest to you is to execute the asynchronous code in a thread. But you should bear in mind that recovering a thread is quite costly.

import asyncio
from concurrent.futures import Future, ThreadPoolExecutor
from injection import singleton

thread_pool = ThreadPoolExecutor()

class Client:
    ...

@singleton
def client_recipe() -> Client:
    async def async_recipe() -> Client:
        client = Client()
        # ...
        return client

    coro = async_recipe()
    future: Future[Client] = thread_pool.submit(asyncio.run, coro)
    return future.result()

(nest-asyncio repository has been archived I'm not sure it's a good idea to use it.)

[remimd]

python-injection has natively supported asynchronous recipes since version 0.12.0.