Build High-Throughput Applications with Qdrant’s Async API

Asynchronous programming is being broadly adopted in the Python ecosystem. Tools such as FastAPI have embraced this new paradigm, but it is also becoming a standard for ML models served as SaaS. For example, the Cohere SDK provides an async client next to its synchronous counterpart.

Databases are often launched as separate services and are accessed via a network. All the interactions with them are IO-bound and can be performed asynchronously so as not to waste time actively waiting for a server response. In Python, this is achieved by using async/await syntax. That lets the interpreter switch to another task while waiting for a response from the server.

When to use async API

There is no need to use async API if the application you are writing will never support multiple users at once (e.g it is a script that runs once per day). However, if you are writing a web service that multiple users will use simultaneously, you shouldn’t be blocking the threads of the web server as it limits the number of concurrent requests it can handle. In this case, you should use the async API.

Modern web frameworks like FastAPI and Quart support async API out of the box. Mixing asynchronous code with an existing synchronous codebase might be a challenge. The async/await syntax cannot be used in synchronous functions. On the other hand, calling an IO-bound operation synchronously in async code is considered an antipattern. Therefore, if you build an async web service, exposed through an ASGI server, you should use the async API for all the interactions with Qdrant.

Using Qdrant asynchronously

To follow along, you need a Qdrant instance running at localhost:6333, as set up in the Local Quickstart, and the client library: pip install qdrant-client.

The simplest way to run asynchronous code is to define an async function and pass it to asyncio.run:

from qdrant_client import models

import qdrant_client
import asyncio


async def main():
    client = qdrant_client.AsyncQdrantClient("localhost")

    # Create a collection
    await client.create_collection(
        collection_name="my_collection",
        vectors_config=models.VectorParams(size=4, distance=models.Distance.COSINE),
    )

    # Insert a vector
    await client.upsert(
        collection_name="my_collection",
        points=[
            models.PointStruct(
                id="5c56c793-69f3-4fbf-87e6-c4bf54c28c26",
                payload={
                    "color": "red",
                },
                vector=[0.9, 0.1, 0.1, 0.5],
            ),
        ],
    )

    # Search for nearest neighbors
    response = await client.query_points(
        collection_name="my_collection",
        query=[0.9, 0.1, 0.1, 0.5],
        limit=2,
    )
    print(response.points)

    # Your async code using AsyncQdrantClient might be put here
    # ...

    # Delete the collection so the example can run again
    await client.delete_collection(collection_name="my_collection")


asyncio.run(main())

The script prints the stored point with a score of 1.0, since the query vector is identical to it:

[ScoredPoint(id='5c56c793-69f3-4fbf-87e6-c4bf54c28c26', version=1, score=1.0, payload={'color': 'red'}, vector=None, shard_key=None, order_value=None)]

The AsyncQdrantClient provides the same methods as the synchronous counterpart QdrantClient. If you already have a synchronous codebase, switching to async API is as simple as replacing QdrantClient with AsyncQdrantClient and adding await before each method call.