Add Async Client to be able to make asynchronous requests

[StefanBogdan]

No description provided.

[reppertj]

We'll need to do this very soon for our use case and are happy to contribute something back.

A lot of projects have found maintaining both sync and async versions of clients or libraries in python have been tricky (Elasticsearch, httpx, Mongo/Motor, to name a few) -- any thoughts on the preferred implementation?

From our point of view, the easiest thing to do is "fake" async by just running requests in a separate threadpool and awaiting them from the event loop, which is what Motor/Mongo does. This results in minimal code duplication and maintenance overhead with a small performance penalty. The major downside from our point of view is that without better support for variadic generics it's impossible to infer types for MyPy users via existing types in the sync version.

Elasticsearch-py uses code gen unasync but that can add a lot of flakiness/complexity and would also require a fairly large refactor.

The other solution is to accept some level of duplication while trying to minimize the number of places in the code & API where async vs sync actually makes a difference. This would require a larger refactor and possibly breaking API changes, but it is, for example, more or less the solution httpx ended up with. This works better when anticipating a major release or designing something from scratch.

Just thought I would get your thoughts on any preferences here before starting something!

[bobvanluijt]

cc: @michaverhagen and @StefanBogdan

[StefanBogdan]

Hi @reppertj , we are going to use the aiohttp library for the AsyncClient because it has very similar requests methods. The way I want it to be structured is to have an Abstract Class for all the classes that make HTTP requests to Weaviate, then implement a Sync and Async versions where all the pre-process is done either in the Abstract class or by some function, that is used by both sync/async Classes. the Sync/Async classes will have only to handle the errors.

The current status of the version 4.0.0 is 80% done, but not tested.

[creatorrr]

@StefanBogdan @bobvanluijt any news/plans for this?

[bobvanluijt]

Cc @dirkkul

[dirkkul]

No concrete plans, sorry. This is something I'd like to explore at some point, but we need to see when we'll find the time

[creatorrr]

Gotcha. It can be a big blocker for a lot of people (including us) because the code currently blocks during the request and when the cluster is struggling with too many requests it could lock up the handler processes. Please see if you can prioritize this at some point. :)

[kubre]

Is there no other way around this?

[netapy]

Any news on this ? My service is slowing down lots of requests because of this :/

[kubre]

@netapy Nope, I decided simply write this part using aiohttp and aiographql clients. It's not as clean looking as it might have been with lib itself but wrapping all the connection and querying part inside the a single class would help you alot.

[netapy]

@netapy Nope, I decided simply write this part using aiohttp and aiographql clients. It's not as clean looking as it might have been with lib itself but wrapping all the connection and querying part inside the a single class would help you alot.

@kubre I'm sorry could you show me some code ?
I'm just trying out a basic example but it doesn't seem to treat the queries concurrently.

@app.get("/search/{query_string}")
async def read_search(query_string: str):

    url = 'http://api.example.tech:8080/v1/graphql'
    headers = {
        'Content-Type': 'application/json',
        'Authorization': f'Bearer MYTOKEN
    }

    query = '''
        query {
            Get {
                Article(
                    hybrid: {
                        query: "xxxxxx"
                        alpha: 0.7
                    }) {
                    titre
                    pathTitle
                    texte
                    code
                }
            }
        }
    '''.replace("xxxxxx", query_string)

    await asyncio.sleep(3)

    async with aiohttp.ClientSession() as session:
        async with session.post(url, headers=headers, data=json.dumps({'query': query})) as response:
            return await response.json()

[kubre]

@netapy Do you mean your entire thread is blocked until it finishes this request? I can't see any issues with the above code. Just calling API with aiohttp should stop it from blocking the main thread. Even when the below code is executing any other request coming into the server is processed without any issues even DB queries like reading objects while some other are being inserted

  async with self.session.post(
      f"{self.url}/v1/batch/objects?consistency_level=ALL",
      json={"objects": batch},
  ) as response:
      await response.json()

[barbu110]

Hi people, is there any update on this?

[netapy]

@barbu110 Hi people, is there any update on this?

I don't think they integrated aysnc into the client yet – however I managed to make it work using a classi HTTP call with aiohttp.

async def search(query_string: str):   

    query = '''
        {
            Get{
                Article(
                    limit: 20
                    hybrid: {
                        query: "xxxxxx"
                        alpha: 0.75
                        fusionType: relativeScoreFusion
                    }){
                        texte
                        title
                    }
                    _additional {
                        score
                    }
                }
            }
        }
    '''.replace("xxxxxx", query_string)

    async with aiohttp.ClientSession() as session:
        async with session.post(url, headers=headers, data=json.dumps({'query': query})) as response:
            return await response.json()

This doesn't block the main thread and works flawlessly.
I don't use the client in production anymore, I find it useful only for data batch upload.

I hope this helps.

[andersfylling]

I'm successfully using aiohttp as well, and we saw a change from around 40rps to 130rps improvement. Ofcourse our service does more than just query weaviate, but it really helps out.

Some details to note:

• create aiohttp.ClientSession only once for all future weaviate queries
• you can re-use the weaviate python client as a query builder

# example
query = weaviate_client.query.get(class_name=class_name, properties=properties).with_limit(1).build()
async with aiohttp_client_session.post(url, headers=headers, data=json.dumps({"query": request})) as response:
    result = await response.json()

[andersfylling]

Any reason you don't commit to a pure async implementation, and then have a lightweight sync wrapper around it?

[plv]

@dirkkul Would you accept design proposals/PRs? We have a patched private fork of the Weaviate Python client with async capability. This is quite dangerous and hacky so I would rather just work on upstreaming it :)

Also, for what it's worth I think we can start with just an async client for reads. This is the use case that most people seem to want when they talk about wanting async support

[kranthi419]

@andersfylling what is the url value here.

example

query = weaviate_client.query.get(class_name=class_name, properties=properties).with_limit(1).build()
async with aiohttp_client_session.post(url, headers=headers, data=json.dumps({"query": request})) as response:
result = await response.json()

[mohit-sarvam]

@kranthi419 the url looks like 'http://api.example.tech:8080/v1/graphql'.

[TweedBeetle]

Would love an update on this!

[flavour-of-qualia]

@andersfylling so how do use use the query from = weaviate_client.query? Later on you use json.dumps({"query": request})) , not the query variable

[dirkkul]

we are currently working on bringing v4 out of beta as fast as possible - async is the next item right afterwards

[TweedBeetle]

@dirkkul Loving v4!
Any news on async functionality? :D 👉👈

[dirkkul]

It is on the roadmap for this quarter

[netapy]

Any news on this ? :)

[dirkkul]

There is a draft PR but it is not ready yet - we have some other important things to work on, but it is on its way

[tsmith023]

Hi all, we have released v4.7.0b0 here as the first beta release of the new async implementation!

We intend on conducting this community beta testing for a few weeks to get feedback and uncover any potentially missed bugs before we fully release it in v4.7.0. If you're happy to get involved then we'd love to hear your feedback! Cheers 😁

[dirkkul]

@TweedBeetle @netapy

It took longer than we thought, but we would be very happy to get your feedback and if everything is working for you, thx! :)

[tsmith023]

Closed by #1007