tessl/pypi-qdrant-client

Client library for the Qdrant vector search engine

—

Pending

Quality

Pending

Does it follow best practices?

Impact

Pending

No eval scenarios have been run

Securityby

Pending

The risk profile of this skill

Overview

Eval results

Files

Qdrant Client

Name: tessl/pypi-qdrant-client
Author: tessl

A comprehensive Python client library for the Qdrant vector search engine, enabling developers to interact with Qdrant databases for vector similarity search, neural matching, and AI-powered search applications. The library offers both synchronous and asynchronous API methods with full type hints, supports both REST and gRPC protocols for optimal performance, includes a local mode for development without running a separate server, and provides helper methods for common operations like collection management and bulk data uploading.

Package Information

Package Name: qdrant-client
Language: Python
Installation: pip install qdrant-client
Optional Features: pip install qdrant-client[fastembed] for embedding generation

Core Imports

from qdrant_client import QdrantClient

For async operations:

from qdrant_client import AsyncQdrantClient

Import models and types:

from qdrant_client import models
from qdrant_client.models import Distance, VectorParams, PointStruct, Filter

Basic Usage

from qdrant_client import QdrantClient, models
import numpy as np

# Initialize client (local mode for development)
client = QdrantClient(":memory:")  # or QdrantClient(path="./qdrant_storage")

# For remote server
# client = QdrantClient(host="localhost", port=6333)

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

# Insert vectors
vectors = np.random.rand(10, 100)
client.upsert(
    collection_name="my_collection",
    points=[
        models.PointStruct(
            id=idx,
            vector=vector.tolist(),
            payload={"category": "example", "number": idx}
        )
        for idx, vector in enumerate(vectors)
    ]
)

# Search for similar vectors
query_vector = np.random.rand(100)
hits = client.query_points(
    collection_name="my_collection",
    query=query_vector,
    limit=5
)

# Search with filtering
filtered_hits = client.query_points(
    collection_name="my_collection",
    query=query_vector,
    query_filter=models.Filter(
        must=[
            models.FieldCondition(
                key="number",
                range=models.Range(gte=3)
            )
        ]
    ),
    limit=5
)

Architecture

The Qdrant client library is built around several key components:

Client Classes: QdrantClient and AsyncQdrantClient provide the main interface
Transport Layer: Automatic REST/gRPC protocol handling with conversion between formats
Local Mode: QdrantLocal enables development without running a separate server
FastEmbed Integration: Optional automatic embedding generation for text and images
Model System: Comprehensive type definitions ensuring compatibility between REST and gRPC APIs
Authentication: Bearer token support for secure connections

Capabilities

Client Initialization & Connection

Core client setup, connection management, and configuration options supporting local mode, remote servers, and cloud instances with authentication.

class QdrantClient:
    def __init__(
        self,
        location: Optional[str] = None,
        url: Optional[str] = None,
        port: Optional[int] = 6333,
        grpc_port: Optional[int] = 6334,
        prefer_grpc: bool = False,
        https: Optional[bool] = None,
        api_key: Optional[str] = None,
        prefix: Optional[str] = None,
        timeout: Optional[float] = None,
        host: Optional[str] = None,
        path: Optional[str] = None,
        **kwargs
    ): ...

class AsyncQdrantClient:
    def __init__(self, **kwargs): ...

Client Setup

Collection Management

Collection creation, deletion, configuration, and metadata management including vector space configuration, optimization settings, and aliasing.

def create_collection(
    self,
    collection_name: str,
    vectors_config: Union[VectorParams, VectorsConfig],
    shard_number: Optional[int] = None,
    replication_factor: Optional[int] = None,
    write_consistency_factor: Optional[int] = None,
    on_disk_payload: Optional[bool] = None,
    hnsw_config: Optional[HnswConfig] = None,
    optimizers_config: Optional[OptimizersConfig] = None,
    wal_config: Optional[WalConfig] = None,
    quantization_config: Optional[QuantizationConfig] = None,
    init_from: Optional[InitFrom] = None,
    timeout: Optional[int] = None,
) -> bool: ...

def delete_collection(self, collection_name: str, timeout: Optional[int] = None) -> bool: ...
def get_collection(self, collection_name: str) -> CollectionInfo: ...
def get_collections(self) -> CollectionsResponse: ...
def collection_exists(self, collection_name: str) -> bool: ...

Collection Management

Vector Operations

Point insertion, updating, deletion, and retrieval operations including batch processing and bulk upload utilities.

def upsert(
    self,
    collection_name: str,
    points: Iterable[PointStruct],
    wait: bool = True,
    ordering: Optional[WriteOrdering] = None,
    shard_key_selector: Optional[ShardKeySelector] = None,
    **kwargs
) -> UpdateResult: ...

def upload_points(
    self,
    collection_name: str,
    points: Iterable[PointStruct],
    batch_size: int = 100,
    parallel: int = 1,
    max_retries: int = 3,
    wait: bool = True,
    shard_key_selector: Optional[ShardKeySelector] = None,
) -> None: ...

def delete(
    self,
    collection_name: str,
    points_selector: Union[PointIdsList, FilterSelector],
    wait: bool = True,
    ordering: Optional[WriteOrdering] = None,
    shard_key_selector: Optional[ShardKeySelector] = None,
    **kwargs
) -> UpdateResult: ...

Vector Operations

Search & Query

Vector similarity search, recommendations, discovery, and hybrid search capabilities with filtering and result ranking.

def query_points(
    self,
    collection_name: str,
    query: Union[QueryRequest, NumpyArray, QueryResponse, Document, str, List[float]],
    query_filter: Optional[Filter] = None,
    search_params: Optional[SearchParams] = None,
    limit: int = 10,
    offset: Optional[int] = None,
    with_payload: Union[bool, List[str], PayloadSelector] = True,
    with_vectors: Union[bool, List[str]] = False,
    score_threshold: Optional[float] = None,
    using: Optional[str] = None,
    timeout: Optional[int] = None,
    shard_key_selector: Optional[ShardKeySelector] = None,
    **kwargs
) -> QueryResponse: ...

def recommend(
    self,
    collection_name: str,
    positive: Optional[List[PointId]] = None,
    negative: Optional[List[PointId]] = None,
    query_filter: Optional[Filter] = None,
    search_params: Optional[SearchParams] = None,
    limit: int = 10,
    offset: int = 0,
    with_payload: Union[bool, List[str], PayloadSelector] = True,
    with_vectors: Union[bool, List[str]] = False,
    score_threshold: Optional[float] = None,
    using: Optional[str] = None,
    lookup_from: Optional[LookupLocation] = None,
    timeout: Optional[int] = None,
    shard_key_selector: Optional[ShardKeySelector] = None,
    **kwargs
) -> List[ScoredPoint]: ...

def discover(
    self,
    collection_name: str,
    target: Optional[PointId] = None,
    context: Optional[List[ContextExamplePair]] = None,
    query_filter: Optional[Filter] = None,
    search_params: Optional[SearchParams] = None,
    limit: int = 10,
    offset: int = 0,
    with_payload: Union[bool, List[str], PayloadSelector] = True,
    with_vectors: Union[bool, List[str]] = False,
    using: Optional[str] = None,
    lookup_from: Optional[LookupLocation] = None,
    timeout: Optional[int] = None,
    shard_key_selector: Optional[ShardKeySelector] = None,
    **kwargs
) -> List[ScoredPoint]: ...

Search & Query

FastEmbed Integration

Automatic embedding generation for text and images using the FastEmbed library, enabling semantic search without manual vector creation.

def get_embedding_size(self, model_name: str) -> int: ...

def upload_collection(
    self,
    collection_name: str,
    vectors: Union[
        Iterable[VectorStruct],
        Iterable[PointStruct], 
        Iterable[Record],
        Iterable[Document],
        Iterable[ImageDocument]
    ],
    ids: Optional[Iterable[PointId]] = None,
    batch_size: int = 100,
    parallel: int = 1,
    max_retries: int = 3,
    payload: Optional[Iterable[Payload]] = None,
    wait: bool = True,
    shard_key_selector: Optional[ShardKeySelector] = None,
) -> None: ...

def query_qdrant(
    self,
    collection_name: str,
    query: Union[str, Document, ImageDocument],
    query_filter: Optional[Filter] = None,
    limit: int = 10,
    search_params: Optional[SearchParams] = None,
    **kwargs
) -> List[QueryResponse]: ...

FastEmbed Integration

Indexing & Optimization

Payload field indexing, collection optimization, and performance tuning capabilities.

def create_payload_index(
    self,
    collection_name: str,
    field_name: str,
    field_schema: Optional[PayloadFieldSchema] = None,
    wait: bool = True,
    ordering: Optional[WriteOrdering] = None,
    **kwargs
) -> UpdateResult: ...

def delete_payload_index(
    self,
    collection_name: str,
    field_name: str,
    wait: bool = True,
    ordering: Optional[WriteOrdering] = None,
    **kwargs
) -> UpdateResult: ...

Indexing & Optimization

Clustering & Sharding

Distributed operation support including shard key management, cluster operations, and multi-tenant configurations.

def create_shard_key(
    self,
    collection_name: str,
    shard_key: ShardKey,
    shards_number: Optional[int] = None,
    replication_factor: Optional[int] = None,
    placement: Optional[List[int]] = None,
    timeout: Optional[int] = None,
    **kwargs
) -> bool: ...

def delete_shard_key(
    self,
    collection_name: str,
    shard_key: ShardKey,
    timeout: Optional[int] = None,
    **kwargs
) -> bool: ...

Clustering & Sharding

Snapshots & Backup

Collection and full database snapshot creation, management, and restoration capabilities for backup and disaster recovery.

def create_snapshot(
    self,
    collection_name: str,
    wait: bool = True,
    **kwargs
) -> SnapshotDescription: ...

def create_full_snapshot(self, wait: bool = True, **kwargs) -> SnapshotDescription: ...
def list_snapshots(self, collection_name: str, **kwargs) -> List[SnapshotDescription]: ...
def delete_snapshot(
    self,
    collection_name: str,
    snapshot_name: str,
    wait: bool = True,
    **kwargs
) -> bool: ...

Snapshots & Backup

Core Types

class Distance(str, Enum):
    COSINE = "Cosine"
    EUCLID = "Euclid" 
    DOT = "Dot"
    MANHATTAN = "Manhattan"

class VectorParams(BaseModel):
    size: int
    distance: Distance
    hnsw_config: Optional[HnswConfig] = None
    quantization_config: Optional[QuantizationConfig] = None
    on_disk: Optional[bool] = None

class PointStruct(BaseModel):
    id: PointId
    vector: Union[VectorStruct, Dict[str, VectorStruct]]
    payload: Optional[Payload] = None

class Filter(BaseModel):
    must: Optional[List[Condition]] = None
    must_not: Optional[List[Condition]] = None
    should: Optional[List[Condition]] = None
    min_should: Optional[MinShould] = None

class ScoredPoint(BaseModel):
    id: PointId
    version: int
    score: float
    payload: Optional[Payload] = None
    vector: Optional[Union[List[float], Dict[str, List[float]]]] = None
    shard_key: Optional[ShardKey] = None
    order_value: Optional[OrderValue] = None

# Type aliases
PointId = Union[str, int]
Payload = Dict[str, Any]
VectorStruct = List[float]

Workspace: tessl
Visibility: Public
Created: 7 months ago
Last updated: 2 months ago
Describes: pkg:pypi/qdrant-client@1.15.x
Publish Source: CLI
Badge