A well-designed Python SDK transforms a raw HTTP API into an idiomatic library that feels native to Python developers. Instead of constructing URLs, serializing JSON, and parsing responses manually, users call methods like client.create_collection("embeddings", size=768) and get back typed Pydantic models with autocomplete, validation, and serialization built in. The SDK handles connection pooling, retries, authentication, and error mapping internally, letting users focus on their application logic.
The mydb-py project implements a Python client library for a vector database service. It uses httpx as the HTTP transport (supporting both sync and async usage), Pydantic for request and response modeling, and a connection lifecycle pattern that manages API keys, base URLs, timeouts, and health checks. The SDK supports batch operations for high-throughput vector insertion and a polling pattern that waits for index readiness after mutations.
This document covers the architecture of a production Python SDK: how to structure the client class, model requests and responses with Pydantic, handle connection lifecycle, implement batch operations efficiently, and design a polling mechanism for asynchronous server-side operations.
httpx is a modern Python HTTP client that supports both synchronous and asynchronous usage with the same API. Unlike requests, httpx supports HTTP/2, connection pooling, streaming, and async/await natively. For SDK development, httpx’s AsyncClient maintains a persistent connection pool that reuses TCP connections across requests, reducing latency for the hundreds of small API calls typical in database workloads. The client also supports configurable timeouts, automatic redirect handling, and event hooks for logging and metrics.
Pydantic models define the shape of API requests and responses using Python type annotations. A CreateCollectionRequest model with fields like size: int and distance: Literal["Cosine", "Euclid", "Dot"] validates inputs at construction time, serializes to JSON for the HTTP request, and provides IDE autocomplete. Response models parse the server’s JSON into typed Python objects, catching schema mismatches early rather than surfacing as KeyError deep in application code.
A production SDK manages the full lifecycle of its HTTP connection: initialization (configuring base URL, API key, timeouts, retry policy), health checking (verifying the server is reachable before sending workload), usage (sending requests through the connection pool), and cleanup (closing the connection pool and releasing resources). Python’s context manager protocol (async with) is the natural fit, ensuring connections are cleaned up even if an exception occurs.
Inserting vectors one at a time over HTTP is inefficient due to per-request overhead (TCP round trip, JSON encoding, server-side transaction). Batch operations group multiple vectors into a single request, amortizing the overhead. The SDK accepts a list of vectors and chunks them into batches of configurable size (e.g., 100 vectors per request), sending each batch as a separate HTTP request but parallelizing them with asyncio.gather for maximum throughput.
After inserting or updating vectors, the database needs time to build or update its index. The SDK provides a wait_for_ready method that polls the collection status endpoint at increasing intervals until the index reports a “green” status. This pattern abstracts away the asynchronous nature of server-side indexing, giving users a simple synchronous-feeling API: insert vectors, wait for ready, then query.
The SDK’s main entry point is a client class that wraps an httpx client and provides typed methods for each API endpoint. The class supports both sync and async usage patterns:
from __future__ import annotations
import httpx
from pydantic import BaseModel, Field
from typing import Literal
class VectorParams(BaseModel):
size: int
distance: Literal["Cosine", "Euclid", "Dot"]
class CollectionInfo(BaseModel):
name: str
vectors_count: int
status: Literal["green", "yellow", "red"]
class PointStruct(BaseModel):
id: int | str
vector: list[float]
payload: dict | None = None
class DatabaseClient:
def __init__(
self,
url: str = "http://localhost:6333",
api_key: str | None = None,
timeout: float = 30.0,
):
self._base_url = url.rstrip("/")
headers = {}
if api_key:
headers["Authorization"] = f"Bearer {api_key}"
self._client = httpx.Client(
base_url=self._base_url,
headers=headers,
timeout=httpx.Timeout(timeout),
)
def close(self):
self._client.close()
def __enter__(self):
return self
def __exit__(self, *args):
self.close()
def _url(self, path: str) -> str:
return path
def _raise_for_status(self, response: httpx.Response) -> None:
if response.status_code >= 400:
detail = response.json().get("message", response.text)
raise DatabaseAPIError(response.status_code, detail)Each API endpoint maps to a method that accepts Pydantic models (or their constituent fields) and returns Pydantic models. The method handles serialization, HTTP transport, and deserialization:
class DatabaseClient:
# ... (continued from above)
def get_collection(self, name: str) -> CollectionInfo:
response = self._client.get(f"/collections/{name}")
self._raise_for_status(response)
return CollectionInfo.model_validate(response.json()["result"])
def create_collection(
self, name: str, vectors: VectorParams
) -> CollectionInfo:
response = self._client.put(
f"/collections/{name}",
json={"vectors": vectors.model_dump()},
)
self._raise_for_status(response)
return CollectionInfo.model_validate(response.json()["result"])
def delete_collection(self, name: str) -> bool:
response = self._client.delete(f"/collections/{name}")
self._raise_for_status(response)
return response.json().get("result", False)
def upsert_points(
self, collection: str, points: list[PointStruct]
) -> dict:
response = self._client.put(
f"/collections/{collection}/points",
json={
"points": [p.model_dump() for p in points],
},
)
self._raise_for_status(response)
return response.json()["result"]The async variant mirrors the sync API but uses httpx.AsyncClient and async/await. A common pattern is to define both sync and async clients that share the same Pydantic models but use different HTTP transports:
class AsyncDatabaseClient:
def __init__(
self,
url: str = "http://localhost:6333",
api_key: str | None = None,
timeout: float = 30.0,
):
self._base_url = url.rstrip("/")
headers = {}
if api_key:
headers["Authorization"] = f"Bearer {api_key}"
self._client = httpx.AsyncClient(
base_url=self._base_url,
headers=headers,
timeout=httpx.Timeout(timeout),
)
async def close(self):
await self._client.aclose()
async def __aenter__(self):
return self
async def __aexit__(self, *args):
await self.close()
async def get_collection(self, name: str) -> CollectionInfo:
response = await self._client.get(f"/collections/{name}")
self._raise_for_status(response)
return CollectionInfo.model_validate(response.json()["result"])
async def upsert_batch(
self,
collection: str,
points: list[PointStruct],
batch_size: int = 100,
parallel: int = 4,
) -> list[dict]:
import asyncio
batches = [
points[i : i + batch_size]
for i in range(0, len(points), batch_size)
]
semaphore = asyncio.Semaphore(parallel)
async def _send_batch(batch: list[PointStruct]) -> dict:
async with semaphore:
response = await self._client.put(
f"/collections/{collection}/points",
json={"points": [p.model_dump() for p in batch]},
)
self._raise_for_status(response)
return response.json()["result"]
return await asyncio.gather(*[_send_batch(b) for b in batches])After bulk insertion, the server’s index may be in a “yellow” state while it rebuilds. The SDK provides a polling helper that waits until the collection status is “green”:
import asyncio
import time
class AsyncDatabaseClient:
# ... (continued)
async def wait_for_ready(
self,
collection: str,
timeout: float = 120.0,
poll_interval: float = 1.0,
max_interval: float = 10.0,
) -> CollectionInfo:
deadline = time.monotonic() + timeout
interval = poll_interval
while time.monotonic() < deadline:
info = await self.get_collection(collection)
if info.status == "green":
return info
await asyncio.sleep(interval)
interval = min(interval * 1.5, max_interval)
raise TimeoutError(
f"Collection '{collection}' not ready after {timeout}s"
)
# Usage pattern
async def ingest_and_query():
async with AsyncDatabaseClient(api_key="secret") as client:
await client.create_collection(
"docs", VectorParams(size=768, distance="Cosine")
)
await client.upsert_batch("docs", vectors, batch_size=200)
await client.wait_for_ready("docs")
results = await client.search("docs", query_vector)The mydb-py SDK is used across several scenarios in a typical vector database ecosystem:
A well-designed SDK defines a custom exception hierarchy that maps HTTP status codes to semantic errors. Users can catch broad exceptions (DatabaseError) or specific ones (CollectionNotFoundError, RateLimitError):
class DatabaseError(Exception):
pass
class DatabaseAPIError(DatabaseError):
def __init__(self, status_code: int, message: str):
self.status_code = status_code
self.message = message
super().__init__(f"HTTP {status_code}: {message}")
class CollectionNotFoundError(DatabaseAPIError):
pass
class RateLimitError(DatabaseAPIError):
pass
def _raise_for_status(response: httpx.Response) -> None:
if response.status_code == 404:
detail = response.json().get("message", "Not found")
raise CollectionNotFoundError(404, detail)
elif response.status_code == 429:
detail = response.json().get("message", "Rate limited")
raise RateLimitError(429, detail)
elif response.status_code >= 400:
detail = response.json().get("message", response.text)
raise DatabaseAPIError(response.status_code, detail)The SDK can optionally wrap requests with retry logic for transient errors (5xx, network timeouts). Using httpx’s transport hooks or a custom transport keeps retry logic separate from business logic.
To avoid duplicating every method for sync and async, some SDKs generate the sync version from the async version by stripping await keywords, or use a shared internal layer that accepts either transport. This approach keeps them separate for clarity but shares all Pydantic models and utility functions.
client.close() (or use a context manager) leaks TCP connections. In long-running processes, this leads to connection pool exhaustion and ConnectionError. Always document and encourage the context manager pattern.DatabaseClient inside an async event loop blocks the entire loop. Always use AsyncDatabaseClient in async code, or run sync calls in a thread pool with asyncio.to_thread.asyncio.gather can overwhelm the server or hit rate limits. Use a semaphore to cap concurrency (e.g., 4-8 parallel requests)..dict() → .model_dump(), .parse_obj() → .model_validate()). SDKs should pin a Pydantic version range and use the correct API.