Unable to access info.context['background']

[austincollinpena]

I'm hoping to avoid any use of Celery at the moment. In the docs there are two ways to add background tasks:

Via Graphene: https://www.starlette.io/graphql/

class Query(graphene.ObjectType):
    user_agent = graphene.String()

    def resolve_user_agent(self, info):
        """
        Return the User-Agent of the incoming request.
        """
        user_agent = request.headers.get("User-Agent", "<unknown>")
        background = info.context["background"]
        background.add_task(log_user_agent, user_agent=user_agent)
        return user_agent

Via a JSON response: https://www.starlette.io/background/

async def signup(request):
    data = await request.json()
    username = data['username']
    email = data['email']
    task = BackgroundTask(send_welcome_email, to_address=email)
    message = {'status': 'Signup successful'}
    return JSONResponse(message, background=task)

Does anyone know of a way to add tasks to Starlette's background with Ariadne? I am unable to return a JSONResponse in my resolver, and I do not have access to a info.context["background"]. The only thing I have attached to my context is my request object.

[bartek]

Do you by chance have an example how this would look with Ariadne? Or how you would expect it to function? I'm not particularly familiar with it and I imagine that might be the case for others.

[austincollinpena]

@bartek I don't see this being any different than with Graphene. If during the resolver of a mutation we want to execute a task in the background and return something, I'd like to be able to:

For example if I'm trying to send an authentication email it could look roughly like this:

@mutation.field("authenticateEmail")
async def send_authentication_email(user, obj, info):
    email = user["email"]
    tasks = BackgroundTasks()
    tasks.add_task(mail("authenticate"))
    return True


async def mail(type: string):
    if type === "authenticate":
        send_the_email()

[austincollinpena]

Got the first part thanks to Benwis on spectrum.

1. Middleware:

@app.middleware("http")
async def dispatch(request: Request, call_next: RequestResponseEndpoint) -> Response:
    # Available as info.context["request"].state.background in your resolver. 
    request.state.background = None
    response = await call_next(request)
    if request.state.background:
        background_task = BackgroundTask(request.state.background)    
        response.background = background_task
    return response

2. In the resolver:

@query.field("getUser")
@check_authentication
async def resolve_get_user(user, obj, info):
    request = info.context["request"]
    request.state.background = testFunc
    return True


async def testFunc():
    print('I did it!')

However, could you help with:

1. Passing through an argument to testFunc
2. Adding multiple tasks

When calling request.state.background = testFunc("foo"), (with the testFunc accepting an argument), I get: TypeError: the first argument must be callable

I'm assuming the solution to #1 will help with #2 as well.

[austincollinpena]

Solved

Middleware:

class BackgroundTaskMiddleware(BaseHTTPMiddleware):
    async def dispatch(
            self, request: Request, call_next: RequestResponseEndpoint
    ) -> Response:
        request.state.background = None
        response = await call_next(request)
        if request.state.background:
            response.background = request.state.background
        return response

Resolver:

@query.field("getUser")
@check_authentication
async def resolve_get_user(user, obj, info):
    task = BackgroundTasks()
    task.add_task(test_func)
    task.add_task(testing_func_two, "I work now")
    request = info.context["request"]
    request.state.background = task
    return True


async def test_func():
    await asyncio.sleep(10)
    print("once!!")


async def testing_func_two(message: str):
    print(message)

The functions still execute synchronously, but because they're background tasks I'm not too worried.