Python Client

Python client for ORMDB with SQLAlchemy and Django support.

Installation

pip install ormdb

# With SQLAlchemy support
pip install ormdb[sqlalchemy]

# With Django support
pip install ormdb[django]

Direct Client

Basic Usage

from ormdb import OrmdbClient

client = OrmdbClient("http://localhost:8080")

# Query
result = client.query(
    "User",
    fields=["id", "name", "email"],
    filter={"field": "status", "op": "eq", "value": "active"},
    order_by=[{"field": "name", "direction": "asc"}],
    limit=10,
)

for user in result.entities:
    print(f"{user['name']} <{user['email']}>")

# Insert
result = client.insert("User", {
    "name": "Alice",
    "email": "alice@example.com",
})
user_id = result.inserted_ids[0]

# Update
client.update("User", user_id, {"name": "Alice Smith"})

# Delete
client.delete("User", user_id)

Async Client

import asyncio
from ormdb import AsyncOrmdbClient

async def main():
    async with AsyncOrmdbClient("http://localhost:8080") as client:
        result = await client.query("User", limit=10)
        for user in result.entities:
            print(user["name"])

asyncio.run(main())

Configuration

client = OrmdbClient(
    base_url="http://localhost:8080",
    timeout=30.0,
    headers={"Authorization": "Bearer token"},
)

Query Features

Filtering

# Simple filter
result = client.query("User",
    filter={"field": "status", "op": "eq", "value": "active"})

# AND filter
result = client.query("User",
    filter={
        "and": [
            {"field": "status", "op": "eq", "value": "active"},
            {"field": "age", "op": "ge", "value": 18},
        ]
    })

# OR filter
result = client.query("User",
    filter={
        "or": [
            {"field": "role", "op": "eq", "value": "admin"},
            {"field": "role", "op": "eq", "value": "moderator"},
        ]
    })

# Pattern matching
result = client.query("User",
    filter={"field": "email", "op": "like", "value": "%@example.com"})

# IN operator
result = client.query("User",
    filter={"field": "status", "op": "in", "value": ["active", "pending"]})

Including Relations

# Single include
result = client.query("User",
    includes=[{"relation": "posts"}])

for user in result.entities:
    print(f"{user['name']} has {len(user.get('posts', []))} posts")

# Nested includes
result = client.query("User",
    includes=[{
        "relation": "posts",
        "fields": ["id", "title"],
        "filter": {"field": "published", "op": "eq", "value": True},
        "includes": [{"relation": "comments"}],
    }])

Pagination

# First page
result = client.query("User", limit=10, offset=0)

# Iterate through pages
offset = 0
while True:
    result = client.query("User", limit=100, offset=offset)
    for user in result.entities:
        process(user)

    if not result.has_more:
        break
    offset += 100

Search Methods

The client provides specialized methods for vector, geographic, and full-text search.

Vector Search

Find similar items using HNSW k-nearest neighbor search:

# Find 10 most similar products
similar = client.vector_search(
    "Product",
    "embedding",
    query_vector=[0.1, 0.2, 0.3, ...],
    k=10,
    max_distance=0.5,  # optional threshold
)

for product in similar.entities:
    print(f"Similar: {product['name']}")

Geographic Search

Find entities by location:

# Radius search
nearby = client.geo_search(
    "Restaurant",
    "location",
    center_lat=37.7749,
    center_lon=-122.4194,
    radius_km=5.0,
)

# Bounding box search
in_box = client.geo_box_search(
    "Restaurant",
    "location",
    min_lat=37.7, min_lon=-122.5,
    max_lat=37.85, max_lon=-122.35,
)

# Polygon search
in_area = client.geo_polygon_search(
    "Restaurant",
    "location",
    vertices=[
        (37.7, -122.5),
        (37.8, -122.5),
        (37.85, -122.4),
    ],
)

# k-nearest
closest = client.geo_nearest(
    "Restaurant",
    "location",
    center_lat=37.7749,
    center_lon=-122.4194,
    k=10,
)

Full-Text Search

Search text content with BM25 ranking:

# Basic text search
articles = client.text_search(
    "Article",
    "content",
    "rust programming",
    min_score=0.5,
)

# Phrase search
exact = client.text_phrase_search(
    "Article",
    "content",
    "quick brown fox",
)

# Boolean search
advanced = client.text_boolean_search(
    "Article",
    "content",
    must=["rust"],
    should=["performance", "safety"],
    must_not=["deprecated"],
)

Using SearchFilter Types

from ormdb.types import VectorSearchFilter, GeoRadiusFilter, TextMatchFilter

# Create typed filters
vector_filter = VectorSearchFilter(
    field="embedding",
    query_vector=[0.1, 0.2, 0.3],
    k=10,
    max_distance=0.5,
)

geo_filter = GeoRadiusFilter(
    field="location",
    center_lat=37.7749,
    center_lon=-122.4194,
    radius_km=5.0,
)

text_filter = TextMatchFilter(
    field="content",
    query="rust programming",
    min_score=0.5,
)

# Use with search method
results = client.search("Product", vector_filter)

For more details, see the Search Guide.

SQLAlchemy Integration

from sqlalchemy import create_engine, text, MetaData
from sqlalchemy.orm import sessionmaker

# Create engine
engine = create_engine("ormdb://localhost:8080")

# Raw SQL-like queries
with engine.connect() as conn:
    result = conn.execute(text("SELECT * FROM User LIMIT 10"))
    for row in result:
        print(row)

# Reflect tables
metadata = MetaData()
metadata.reflect(bind=engine)

# Use ORM
Session = sessionmaker(bind=engine)
session = Session()

# Query using reflected tables
User = metadata.tables["User"]
users = session.query(User).filter(User.c.status == "active").all()

Django Integration

Settings

# settings.py
DATABASES = {
    'default': {
        'ENGINE': 'ormdb.django',
        'HOST': 'localhost',
        'PORT': 8080,
    }
}

Models

# models.py
from django.db import models

class User(models.Model):
    name = models.CharField(max_length=255)
    email = models.EmailField()
    status = models.CharField(max_length=50)

    class Meta:
        db_table = 'User'

class Post(models.Model):
    title = models.CharField(max_length=255)
    content = models.TextField()
    author = models.ForeignKey(User, on_delete=models.CASCADE)

    class Meta:
        db_table = 'Post'

Queries

# views.py
from .models import User, Post

# Query
active_users = User.objects.filter(status="active").order_by("name")[:10]

# Create
user = User.objects.create(name="Alice", email="alice@example.com")

# Update
User.objects.filter(id=user.id).update(status="inactive")

# Delete
User.objects.filter(id=user.id).delete()

# Relations
user = User.objects.prefetch_related("post_set").get(id=user_id)
for post in user.post_set.all():
    print(post.title)

Error Handling

from ormdb import OrmdbClient, ConnectionError, QueryError, MutationError

client = OrmdbClient()

try:
    result = client.insert("User", {"email": "existing@example.com"})
except MutationError as e:
    if e.code == "UNIQUE_VIOLATION":
        print("Email already exists")
    elif e.code == "FOREIGN_KEY_VIOLATION":
        print("Invalid reference")
    else:
        print(f"Mutation failed: {e.message}")
except QueryError as e:
    print(f"Query failed: {e.message}")
except ConnectionError as e:
    print(f"Connection failed: {e}")

Change Data Capture

# Stream changes
for change in client.stream_changes(from_lsn=0, entities=["User", "Post"]):
    if change["type"] == "insert":
        print(f"New {change['entity']}: {change['id']}")
    elif change["type"] == "update":
        print(f"Updated {change['entity']}: {change['id']}")
    elif change["type"] == "delete":
        print(f"Deleted {change['entity']}: {change['id']}")

Best Practices

1. Use context managers for async client
2. Handle pagination for large datasets
3. Use appropriate ORM integration for existing projects
4. Set timeouts for production use

Next Steps

• Query API Reference
• Error Reference