compass.services.provider.RunningAsyncServices#

class RunningAsyncServices(services)[source]#

Bases: object

Async context manager for running services.

Parameters:

services (Iterable) – An iterable of async services to run during program execution.

Methods

run(services, coroutine)

Run an async function that relies on services.
classmethod run(services, coroutine)[source]#

Run an async function that relies on services.

You can treat this function like the asyncio.run function with an extra parameter:

openai_service = OpenAIService(...)
RunningAsyncServices.run(
    [openai_service], my_async_func(*args, **kwargs)
)
Parameters:

  • services (Iterable of Service) – An iterable (i.e. a list) of Services that are needed to run the asynchronous function.

  • coroutine (callable()) – A coroutine that should be run with the services.

Returns:

Any – Returns the output of the coroutine.