Help. How to get a Job result async

[superjcvd]

Hello,

First of all let me thank you for thsi awesome library :)
It is really great and much lighter than Celery.

I have a specific use case where I want to schedule Jobs within an external FastAPI.
My worker is running in an external module on docker.
My API is running on another external module on docker.

I have a REST endpoint that permits to create a new Job (Redis backend) and enqueue it.
I want to create another endpoint to asyncly collect my jobs results later.

I am trying to create the job with an uniq uuid, but then within my other API endpoint I dont know how to get the job and then disply the result :)

[superjcvd]

I also understand that I can get the Job result from the results_bucket
But I cant find a function or a clean way to get it from the result_id (other than directly querying redis)

[superjcvd]

I want to achieve exactly what @rajtilak-2020 answered (using AI for sure 🤣 )

But with REPID library :)

[aleksul]

Hello and thanks for your interest. You are correct, in case your job is scheduled in a different context than the context where you are reading the result from, you would have to use a workaround. Here is some kind of MVP example:

import asyncio

from repid import (
    Connection,
    InMemoryBucketBroker,
    InMemoryMessageBroker,
    Job,
    Repid,
    Router,
    Worker,
)


async def main() -> None:
    conn = Repid(
        connection=Connection(
            message_broker=InMemoryMessageBroker(),
            args_bucket_broker=InMemoryBucketBroker(),
            results_bucket_broker=InMemoryBucketBroker(use_result_bucket=True),
        )
    )

    # first FastAPI endpoint, that creates a job
    async with conn.magic():
        j = Job("awesome_job", args={"input_": "Hello!"})
        await j.queue.declare()
        await j.enqueue()

        print(j.result_id)  # you would return this id and expect someone to collect result later

    # this emulates your worker
    router = Router()

    @router.actor
    async def awesome_job(input_: str) -> str:
        return f"I return what I receive: {input_}"

    async with conn.magic():
        myworker = Worker(routers=[router], messages_limit=1)
        await asyncio.wait_for(myworker.run(), timeout=5.0)

    # second FastAPI endpoint, that collects results
    async with conn.magic() as connection:
        result_bucket = await connection.results_bucket_broker.get_bucket(j.result_id)

        if result_bucket is None:
            print("Not ready yet!")
        else:
            print(result_bucket.data)  # this is the return of your actor


if __name__ == "__main__":
    asyncio.run(main())

[superjcvd]

@aleksul Awesome. That's exactly what I was looking for.