cortexdb 0.1.1

Official Python SDK for CortexDB - Multi-modal RAG Platform

CortexDB Python SDK

Official Python client for CortexDB - Multi-modal RAG Platform.

Features

• Async/await support with httpx
• Semantic search with vector embeddings
• File upload with automatic text extraction and vectorization
• Type hints for better IDE support
• Pydantic models for data validation
• Context manager support
• Error handling with custom exceptions

Installation

pip install cortexdb

Or with Poetry:

poetry add cortexdb

Quick Start

import asyncio
from cortexdb import CortexClient, FieldDefinition, FieldType

async def main():
    async with CortexClient("http://localhost:8000") as client:
        # Create a collection
        await client.collections.create(
            name="documents",
            fields=[
                FieldDefinition(name="title", type=FieldType.STRING),
                FieldDefinition(name="content", type=FieldType.TEXT, vectorize=True)
            ]
        )

        # Create a record
        record = await client.records.create(
            collection="documents",
            data={"title": "Hello", "content": "World"}
        )

        # Semantic search
        results = await client.records.query(
            collection="documents",
            query="hello world",
            limit=10
        )

        for result in results:
            print(f"Score: {result.score:.4f} - {result.data['title']}")

asyncio.run(main())

Usage

Initialize Client

from cortexdb import CortexClient

# Local development
client = CortexClient("http://localhost:8000")

# With API key
client = CortexClient("https://api.cortexdb.com", api_key="your-key")

# Custom timeout
client = CortexClient("http://localhost:8000", timeout=60.0)

Collections

from cortexdb import FieldDefinition, FieldType, StoreLocation

# Create collection
schema = await client.collections.create(
    name="articles",
    fields=[
        FieldDefinition(
            name="title",
            type=FieldType.STRING
        ),
        FieldDefinition(
            name="content",
            type=FieldType.TEXT,
            vectorize=True  # Enable semantic search
        ),
        FieldDefinition(
            name="year",
            type=FieldType.INT,
            store_in=[StoreLocation.POSTGRES, StoreLocation.QDRANT_PAYLOAD]
        )
    ]
)

# List collections
collections = await client.collections.list()

# Get collection
schema = await client.collections.get("articles")

# Delete collection
await client.collections.delete("articles")

Records

# Create record
record = await client.records.create(
    collection="articles",
    data={
        "title": "Machine Learning Basics",
        "content": "Introduction to ML concepts...",
        "year": 2024
    }
)

# Get record by ID
record = await client.records.get("articles", record_id="abc-123")

# Update record
updated = await client.records.update(
    collection="articles",
    record_id="abc-123",
    data={"year": 2025}
)

# Delete record
await client.records.delete("articles", record_id="abc-123")

Semantic Search

# Basic search
results = await client.records.query(
    collection="articles",
    query="machine learning fundamentals",
    limit=10
)

# Search with filters
results = await client.records.query(
    collection="articles",
    query="neural networks",
    limit=5,
    filters={
        "year": {"$gte": 2023},  # Year >= 2023
        "category": "AI"          # Exact match
    }
)

# Process results
for result in results:
    print(f"Score: {result.score:.4f}")
    print(f"Title: {result.data['title']}")
    print(f"Year: {result.data['year']}")

File Upload

from pathlib import Path

# Create collection with file field
await client.collections.create(
    name="documents",
    fields=[
        FieldDefinition(name="title", type=FieldType.STRING),
        FieldDefinition(
            name="pdf",
            type=FieldType.FILE,
            vectorize=True  # Extract text and vectorize
        )
    ]
)

# Upload file
record = await client.records.create(
    collection="documents",
    data={"title": "Annual Report"},
    files={"pdf": Path("/path/to/file.pdf")}
)

# Or upload bytes
record = await client.records.create(
    collection="documents",
    data={"title": "Contract"},
    files={"pdf": file_bytes}
)

# Get vectorized chunks
chunks = await client.records.get_vectors("documents", record.id)
for chunk in chunks:
    print(f"Field: {chunk.field}")
    print(f"Chunk {chunk.chunk_index}: {chunk.text[:100]}...")

Filter Operators

# Comparison operators
filters = {
    "year": {"$gte": 2020},    # Greater than or equal
    "score": {"$lte": 100},    # Less than or equal
    "views": {"$gt": 1000},    # Greater than
    "price": {"$lt": 50},      # Less than
}

# Exact match
filters = {
    "category": "technology",
    "published": True
}

# Combine filters
filters = {
    "year": {"$gte": 2020, "$lte": 2024"},
    "category": "AI"
}

Error Handling

from cortexdb import (
    CortexDBError,
    CortexDBNotFoundError,
    CortexDBValidationError,
    CortexDBConnectionError
)

try:
    record = await client.records.get("articles", "invalid-id")
except CortexDBNotFoundError as e:
    print(f"Record not found: {e.message}")
except CortexDBValidationError as e:
    print(f"Validation error: {e.message}")
except CortexDBConnectionError as e:
    print(f"Connection failed: {e.message}")
except CortexDBError as e:
    print(f"General error: {e.message}")

Development

Setup

# Clone repository
git clone https://github.com/yourusername/cortexdb
cd cortexdb/clients/python

# Install with Poetry
poetry install

# Install pre-commit hooks
poetry run pre-commit install

Running Tests

# Run all tests
poetry run pytest

# With coverage
poetry run pytest --cov=cortexdb --cov-report=html

# Run specific test
poetry run pytest tests/test_client.py::test_create_collection

Code Quality

# Format with Black
poetry run black cortexdb tests

# Lint with Ruff
poetry run ruff check cortexdb

# Type check with mypy
poetry run mypy cortexdb

Examples

Check the examples/ directory for more usage examples:

• quickstart.py - Walkthrough of SDK features
• file_upload.py - File upload and vectorization

Run examples:

poetry run python examples/quickstart.py

Requirements

• Python 3.8+
• CortexDB gateway running (local or remote)

License

MIT License - see LICENSE for details.