`Resource` subclass and its `init` return value

[pansen]

Hi there,

thanks a lot for this library and its great documentation! I could manage to port important parts of my application to using this, without any bigger issues.

One thing I had to spend some time for was the following.

Consider I subsclass a Resource, which creates a asyncio.Task, which should be cancelled on shutdown. I could not get this to work, except with this workaround:

class DispatcherQueueConsumerResource(resources.Resource):
    async def init(self, loop: AbstractEventLoop, disp, queue, **kwargs) -> Optional[Task]:
        t: Task = loop.create_task(disp.queue_consumer(queue))

        # If this is an async method, we store the task to have that usable in `shutdown`.
        #
        # The `shutdown` method receives this method as an awaited coroutine and *not* the task we
        # just created and may return. Ergo does the cancellation not work, if we don't store our own
        # copy here.
        setattr(self, '__t', t)
        
        return

    async def shutdown(self, t: Task) -> None:
        to_cancel = None
        t_ = getattr(self, '__t', None)

        if t_:
            # Take `self.__t` with preference
            to_cancel = t_
        elif t:
            to_cancel = t
        else:
            log.error("Nothing to shutdown: %s, %s", self, t)
            return 

        try:
            if isinstance(to_cancel, Task):
                to_cancel_task = to_cancel
            elif inspect.iscoroutine(to_cancel):
                to_cancel_task = asyncio.create_task(to_cancel)
            else:
                raise RuntimeError(f"Don't know what I got here: {t}")
            was_cancelled = to_cancel_task.cancel()
            if was_cancelled:
                wait_for_seconds = 0.1
                await asyncio.wait_for(to_cancel_task, wait_for_seconds)
                foo = 1
        except AttributeError as e:
            log.error('Could not call `%s.cancel`: %s', to_cancel, e)

As described in the init comment, we will receive a coroutine of DispatcherQueueConsumerResource.init as parameter in DispatcherQueueConsumerResource.shutdown, no matter what is returned from DispatcherQueueConsumerResource.init. It is hackable with this subclass solution, but not using the generator approach.

If I would be using the parameter t in shutdown, my code stalls forever when calling shutdown_resources. Therefore using the generator approach for my Resource is not working as is.

Just in case this is relevant, cause I saw this bug, I'm also using FastAPI in my application.

Please let me know if I should provide more information :)

Cheers!

[pansen]

Update, sorry. The generator approach also works, this was broken due to some other issue in my code 🙈

async def connect_dispatcher_to_queue(loop, disp, queue):
    t:Task = loop.create_task(disp.queue_consumer(queue))
    yield t
    t.cancel()

[ZipFile]

In v4.45.0 we've introduced ASGI Lifespan Protocol support and in v4.48.0 you can use context managers (async too!) in Resource providers. Give it a try, I think it should resolve all the issues you've had so far. Closing the issue.