milvusql

A PEP 249 DBAPI (sync + async) for Milvus — parses/generates MilvusQL via sqlglot-milvus and executes the resulting AST against pymilvus.

📚 Documentation · PyPI · sqlglot-milvus

Why a DBAPI, not a client wrapper?

Writing MilvusQL instead of chaining pymilvus calls means the same SELECT ... ORDER BY embedding <=> :q LIMIT n string works whether it's typed by hand, generated by an ORM, or built by an LLM tool call — and it works the same way from cursor.execute() or await acursor.execute(), off one shared parser and dispatch table (translate.ast_to_pymilvus).

Installation

pip install milvusql

Quick start

import milvusql

conn = milvusql.connect(uri="./items.db")  # Milvus Lite, or a real server's URI
cur = conn.cursor()

cur.execute(
    """
    CREATE TABLE items (
        id BIGINT PRIMARY KEY AUTO_INCREMENT,
        embedding VECTOR(8),
        category VARCHAR(64)
    ) WITH (shards=1, consistency_level='Strong')
    """
)
cur.execute(
    "CREATE INDEX idx_embedding ON items (embedding) USING HNSW WITH (metric_type='COSINE')"
)
cur.executemany(
    "INSERT INTO items (embedding, category) VALUES (:embedding, :category)",
    [{"embedding": [0.1] * 8, "category": "book"}],
)

cur.execute("SELECT id FROM items WHERE category = :cat LIMIT 10", {"cat": "book"})
print(cur.fetchall())

cur.execute(
    "SELECT id FROM items ORDER BY embedding <=> :q LIMIT 5",
    {"q": [0.1] * 8},
)
print(cur.fetchall())

The same program, asyncio-native, over milvusql.aio (built on pymilvus.AsyncMilvusClient):

from milvusql import aio

conn = aio.connect(uri="./items.db")
cur = conn.cursor()

await cur.execute("SELECT id FROM items WHERE category = :cat", {"cat": "book"})
async for row in cur:
    print(row)

await conn.close()

API

milvusql.connect()

milvusql.connect(
    uri="http://localhost:19530",  # or a Milvus Lite file path
    token="",                      # "user:password", or a full token string
    db_name="",
    consistency_level=None,        # per-connection default; a query's own CONSISTENCY LEVEL wins
    **kwargs,                      # passed straight through to pymilvus.MilvusClient
) -> Connection

milvusql.aio.connect()/AsyncConnection/AsyncCursor mirror the same shape, async/await throughout — deliberately not PEP 249 itself (execute() as a coroutine can't be), but built on the same parser, dispatch table, and error hierarchy as the sync path.

Errors

Standard PEP 249 hierarchy, importable from milvusql:

Warning
Error
├── InterfaceError
└── DatabaseError
    ├── DataError
    ├── OperationalError
    ├── IntegrityError
    ├── InternalError
    ├── ProgrammingError
    └── NotSupportedError

Every pymilvus exception and gRPC error raised while executing a statement is translated into one of these before it reaches your code.

Packages

This is the core of a uv workspace. Two packages build on milvusql's DBAPI:

Each is installed and versioned separately; both depend on this package as their DBAPI layer.

Examples

See also milvusql-sqlalchemy's own examples (a FastAPI image-search service, a pydantic-ai agent).

Development

Requires Python 3.12+, uv, task.

task install           # uv sync --all-groups --all-packages
task lint              # ruff + ty + bandit for core + all packages
task tests             # all tests (core + sqlalchemy + django) -- integration tests need Docker (testcontainers)

Individual package tasks:

task core:lint         task core:test
task sqlalchemy:lint   task sqlalchemy:test
task django:lint       task django:test

License

MIT