Events are not received when dispatch() occurs in an asyncio.Task about fastapi-events

[smithk86]

@melvinkcx Setting in_req_res_cycle to False is how I am currently getting around this and, while a bit hacky, it does work quite well.

My wrapper for dispatch() looks something like this:

@contextmanager
def _force_fastapi_events_dispatch_as_task() -> Generator:
    token: ContextVarToken = in_req_res_cycle.set(False)
    yield
    in_req_res_cycle.reset(token)


def dispatch(
    event_name: Union[str, Enum],
    payload: Optional[Any] = None,
    validate_payload: bool = True,
    payload_schema_cls_dict_args: Optional[Dict[str, Any]] = None,
    payload_schema_registry: Optional[BaseEventPayloadSchemaRegistry] = None,
    middleware_id: Optional[int] = None
    as_task: bool = False,
) -> None:
    if as_task:
        with _force_fastapi_events_dispatch_as_task():
            fastapi_events_dispatch(
                event_name=event_name,
                payload=payload,
                validate_payload=validate_payload,
                payload_schema_cls_dict_args=payload_schema_cls_dict_args,
                payload_schema_registry=payload_schema_registry,
            )
    else:
        fastapi_events_dispatch(
            event_name=event_name,
            payload=payload,
            validate_payload=validate_payload,
            payload_schema_cls_dict_args=payload_schema_cls_dict_args,
            payload_schema_registry=payload_schema_registry,
        )

[melvinkcx]

Hi @smithk86, thank you for reporting the issue and for your patience. I've been thinking about this issue.

When you said the actual work being done in an asyncio.Task, do you mean you're create a task manually using asyncio.create_task()?

[melvinkcx]

Do you know if there is any good way to tell when contextvars are being copied? Would you mind sharing it or pointing me to the resource? 🙏 thanks

[smithk86]

Below is a brief example of the route handler I was working on when I found this issue. The goal was to have the request start a long-running process but not have the client wait on the work being done. The results of the work would be published as an event.

To answer your question, the context (with contextvars) are being copied when asyncio.create_task() is run. The context is used for the duration of the coroutine being called. Because of this, fastapi-events believes it is executing within the context of a request when the request is actually done.

Cheers
Kyle

def add_data(data: DataModel):
    """
    This function may take a while to complete
    """
    dispatch("add_data_start", payload={"message": "starting"})
    try:
        await database.insert(data)
        await app.reload_data()
    except:
        dispatch("add_data_error", payload={"message": "an error occurred"})
    else:
        dispatch("add_data_done", payload={"message": "complete"})


@app.post("/addData")
async def uuid(
    data: DataModel,
):
    """
    This request will return immediately for the client and then do the work in the background.
    Events will be dispatched within the function/task once the work is complete.
    """
    asyncio.create_task(add_data(data))
    return {"detail": "add_data has been started"}

[melvinkcx]

thank you for the code sample. I can't think of a good general solution so far.
I would imagine a function that wraps asyncio.create_task that would also unset is_req_res_cycle at the same time, but that's not the best solution. 🤔