rail-engine 0.2.0

Python SDK for Railtown AI Rail Engine - Retrieval

Rail Engine Retrieval SDK

Python SDK for retrieving and searching data from Railtown AI Rail Engine - handles embeddings, storage documents, and indexed content.

Overview

The rail-engine package provides a Pythonic interface for retrieving and searching data from Rail Engine. It supports async/await patterns, client-side filtering, explicit storage list paging via StorageListPage, and Pydantic model deserialization.

Installation

uv pip install rail-engine

Quick Start

import asyncio
from railtown.engine import Railengine

async def main():
    # Initialize client (reads from ENGINE_PAT and ENGINE_ID env vars)
    async with Railengine() as client:
        page = await client.list_storage_documents(
            page_number=1,
            page_size=25,
        )
        print(f"total_count={page.total_count} total_pages={page.total_pages}")
        for doc in page.items:
            print(doc)

asyncio.run(main())

Configuration

Environment Variables

• ENGINE_PAT (required) - Personal Access Token
• ENGINE_ID (required) - Engine ID (can also be passed to constructor)
• RAILTOWN_API_URL (optional) - Base API URL (defaults to https://cndr.railtown.ai/api)

Constructor Parameters

• pat (optional) - PAT token (if not provided, reads from ENGINE_PAT env)
• engine_id (optional) - Engine ID (if not provided, reads from ENGINE_ID env, required if not in env)
• api_url (optional) - Base API URL (if not provided, reads from RAILTOWN_API_URL env or defaults to production)
• model (optional) - Pydantic model type for deserializing retrieved data

Features

• Multiple retrieval methods:
  • search_vector_store() - Semantic search in vector stores
  • get_storage_document_by_event_id() - Get document by EventId
  • get_storage_document_by_customer_key() - Same paging as list_storage_documents (page_number, page_size) but scoped to a CustomerKey; returns StorageListPage with items, total_pages, and total_count
  • query_storage_by_jsonpath() - Query using JSONPath
  • list_storage_documents() - Use page_number and page_size to paginate; returns StorageListPage with items, total_pages, and total_count
  • search_index() - Full-text search using Azure Search
  • delete_event() - Delete an event from storage, embeddings, and indexing (raises on API errors)
• Client-side filtering - Filter results using filter_fn parameter
• Storage list pagination - One HTTP GET per page; use page_number / page_size on the request and total_pages / total_count on StorageListPage when looping
• Model deserialization - Optional Pydantic model support with per-call override
• Graceful error handling - Read/search helpers often return None or empty iterables instead of raising
• Async/await support - Built for modern async Python applications

Error Handling

The SDK provides custom exception classes:

• RailtownError - Base exception
• RailtownBadRequestError - 400 Bad Request
• RailtownUnauthorizedError - 401 Unauthorized
• RailtownNotFoundError - 404 Not Found
• RailtownConflictError - 409 Conflict
• RailtownServerError - 5xx Server errors

Read/search helpers return None or empty iterables on many errors (graceful degradation). delete_event() raises Railtown* exceptions on HTTP/API failures.

Requirements

• uv
• Python 3.10+
• httpx >= 0.24.0
• pydantic >= 1.10.0

Related package

For ingesting data into Rail Engine, install and use rail-engine-ingest.

License

MIT