haystack-py 0.1.7

Async Project Haystack client and ontology library for Python

haystack-py

Asynchronous Project Haystack client and server library for Python 3.13+. HTTP and WebSocket transports, four wire formats, SCRAM-SHA-256 and mTLS authentication, role-based access control (Admin/Operator/Viewer), user management with CRUD API, pluggable storage backends (Redis, TimescaleDB), and full ontology support. Built on native asyncio.

Documentation | Getting Started | API Reference | Changelog

from hs_py import Client

async with Client("http://server/api", "admin", "secret") as c:
    about = await c.about()
    points = await c.read("point and temp and sensor")

Table of Contents

• Features
• Installation
• Quick Start
• Authentication, Users & Permissions
• Storage Backends
• Architecture
• Configuration
• Testing
• Requirements
• License

Features

Category	Highlights
Transports	HTTP with aiohttp, WebSocket with websockets sans-I/O, persistent connections, per-message deflate compression
Client	Client (HTTP) and WebSocketClient with all 13 standard ops, batch requests, watch subscriptions, auto-auth
Server	FastAPI application factory, SCRAM-SHA-256 middleware, content negotiation (JSON/Zinc/Trio/CSV), standalone WebSocketServer
Wire Formats	JSON v3/v4 (orjson), Zinc (text), Trio (tagged records), CSV (export-only)
Authentication	SCRAM-SHA-256 over HTTP and WebSocket, PLAINTEXT fallback, token-based WebSocket auth, mTLS with CertAuthenticator, StorageAuthenticator for DB-backed credentials
TLS	TLS 1.3 enforced, mutual authentication, test certificate generation (EC P-256), TLSConfig dataclass
Users & Permissions	User model with SCRAM credentials, Role enum (Admin/Operator/Viewer), CRUD REST API, UserStore protocol, admin bootstrap from env vars
Storage	Pluggable StorageAdapter + UserStore protocols with Redis (RediSearch + RedisTimeSeries), TimescaleDB (asyncpg + JSONB), and in-memory backends
Data Model	All Haystack value types as frozen dataclasses, Grid / GridBuilder as universal message format
Filters	Recursive descent parser, AST representation, evaluation against dicts and grids, SQL pushdown for JSONB/RediSearch
Ontology	Def/Lib/Namespace model, taxonomy queries, tag normalization, dict-to-def reflection
WebSocket Extras ⚠️	ReconnectingWebSocketClient with backoff, WebSocketPool / ChannelClient multiplexing, binary frame codec (experimental — API subject to change)
Observability	MetricsHooks for connection, message, request, and error callbacks
Watch	Server-side WatchState delta encoding, client-side WatchAccumulator delta merging
Quality	1,200+ tests, 69 end-to-end integration tests, 122 TimescaleDB tests, mypy strict, ruff linting, frozen dataclasses throughout

Installation

pip install haystack-py

Optional extras:

pip install haystack-py[server]       # FastAPI + Redis backend (server-side)
pip install haystack-py[timescale]    # TimescaleDB/PostgreSQL backend
pip install haystack-py[rdf]          # RDF ontology import (rdflib)
pip install haystack-py[all]          # All optional dependencies

Development

git clone https://github.com/jscott3201/hs-py.git
cd hs-py
uv sync --group dev

Quick Start

HTTP Client

import asyncio
from hs_py import Client, Ref


async def main():
    async with Client("http://server/api", "admin", "secret") as c:
        # Server info
        about = await c.about()
        print(about[0]["serverName"])

        # Filter-based read
        sites = await c.read("site")
        for row in sites:
            print(row.get("dis"), row.get("id"))

        # ID-based read
        entities = await c.read_by_ids([Ref("site-1"), Ref("equip-2")])

        # Navigation
        nav = await c.nav()  # root sites
        children = await c.nav(Ref("site-1"))  # site's equips

        # History
        history = await c.his_read(Ref("point-1"), "yesterday")
        for row in history:
            print(row["ts"], row["val"])

        # Write a point value
        await c.point_write(Ref("point-1"), level=8, val=72.5)


asyncio.run(main())

WebSocket Client

⚠️ Experimental: The WebSocket transport API is experimental and subject to breaking changes in future releases.

import asyncio
from hs_py import WebSocketClient, Grid, Ref


async def main():
    async with WebSocketClient("ws://server/api/ws", auth_token="token") as ws:
        about = await ws.about()

        # Batch: multiple ops in one round-trip
        results = await ws.batch(
            ("about", Grid.make_empty()),
            ("read", Grid.make_rows([{"filter": "site"}])),
            ("read", Grid.make_rows([{"filter": "point and temp"}])),
        )

        # Watch: subscribe to entity changes
        watch = await ws.watch_sub([Ref("p-1"), Ref("p-2")], "my-watch")
        watch_id = watch.meta["watchId"]
        poll = await ws.watch_poll(watch_id)
        await ws.watch_close(watch_id)


asyncio.run(main())

Server

import asyncio
from hs_py import MARKER, Ref
from hs_py.storage.memory import InMemoryAdapter
from hs_py.auth_types import StorageAuthenticator
from hs_py.fastapi_server import create_fastapi_app
from hs_py.user import Role, create_user
import uvicorn


async def main():
    storage = InMemoryAdapter()
    await storage.start()
    await storage.load_entities([
        {"id": Ref("s1"), "site": MARKER, "dis": "My Building"},
    ])

    # Create an admin user and wire storage-backed auth
    admin = create_user("admin", "secret", role=Role.ADMIN)
    await storage.create_user(admin)
    auth = StorageAuthenticator(storage)

    app = create_fastapi_app(
        storage=storage,
        authenticator=auth,
        user_store=storage,
    )
    config = uvicorn.Config(app, host="0.0.0.0", port=8080)
    server = uvicorn.Server(config)
    await server.serve()


asyncio.run(main())

Wire Formats

from hs_py.encoding import json, zinc, trio, csv
from hs_py.encoding.json import JsonVersion

# JSON v4
grid = json.decode_grid(data, version=JsonVersion.V4)
json_bytes = json.encode_grid(grid, version=JsonVersion.V4)

# Zinc
zinc_text = zinc.encode_grid(grid)
grid = zinc.decode_grid(zinc_text)

# Trio records
records = trio.parse_trio(trio_text)
trio_text = trio.encode_trio(records)

# CSV (encode-only)
csv_text = csv.encode_grid(grid)

Filters

from hs_py import MARKER, parse, evaluate, evaluate_grid

# Parse filter to AST
ast = parse("point and temp and sensor")

# Evaluate against a dict
entity = {"point": MARKER, "temp": MARKER, "sensor": MARKER}
assert evaluate(ast, entity) is True

# Filter a grid
matching = evaluate_grid(ast, grid)

Ontology

from hs_py.ontology.namespace import Namespace, load_lib_from_trio
from hs_py.ontology.reflect import reflect

lib = load_lib_from_trio(trio_text)
ns = Namespace([lib])

# Taxonomy queries
assert ns.is_subtype("ahu", "equip")
subtypes = ns.subtypes("equip")  # [ahu, vav, ...]

# Reflect entities against definitions
defs = reflect(ns, entity_dict)

Authentication, Users & Permissions

User Model

Users are managed via frozen User dataclasses with SCRAM-SHA-256 credentials (passwords are never stored). Each user has a Role that controls access:

Role	Level	Capabilities
Admin	Full	User management, all read/write Haystack ops
Operator	Read + Write	hisWrite, pointWrite, invokeAction, watches, plus all read ops
Viewer	Read-only	read, nav, hisRead, defs, libs, filetypes

from hs_py.user import Role, create_user

# Create users with specific roles
admin = create_user("admin", "secret", role=Role.ADMIN)
operator = create_user("operator", "pass", role=Role.OPERATOR)
viewer = create_user("viewer", "pass", role=Role.VIEWER, email="[e-mail]")

Storage-Backed Authentication

StorageAuthenticator reads SCRAM credentials from any UserStore backend (InMemory, Redis, or TimescaleDB). Disabled users are automatically denied authentication.

from hs_py.auth_types import StorageAuthenticator
from hs_py.storage.memory import InMemoryAdapter

storage = InMemoryAdapter()
auth = StorageAuthenticator(storage)
app = create_fastapi_app(storage=storage, authenticator=auth, user_store=storage)

Admin Bootstrap

On startup, the server verifies at least one enabled Admin user exists. If none is found, it seeds one from environment variables:

Variable	Description
HS_SUPERUSER_USERNAME	Username for the seeded admin account
HS_SUPERUSER_PASSWORD	Password for the seeded admin account

If neither an admin user nor environment variables are provided, the server exits with an error.

User Management API

Admin users can manage users via REST endpoints:

Method	Endpoint	Description
POST	/api/users/	Create a user (with role, username, password)
GET	/api/users/	List all users
GET	/api/users/{username}	Get a single user
PUT	/api/users/{username}	Update user fields (password, role, enabled, etc.)
DELETE	/api/users/{username}	Delete a user (cannot self-delete)

Permission Enforcement

Roles are enforced on all Haystack operations:

• Write ops (hisWrite, pointWrite, invokeAction, watchSub/Unsub/Poll) require Operator or Admin role.
• Read ops (read, nav, hisRead, defs, libs, filetypes) and GET ops (about, ops, formats) are available to all authenticated users.
• User management endpoints require Admin role.
• Enforcement applies to both HTTP and WebSocket transports.

Storage Backends

haystack-py defines StorageAdapter and UserStore protocols that decouple server operations from data storage. Three implementations are provided, each supporting both entity storage and user management:

Backend	Module	Best For
Memory	storage.memory	Testing, prototyping, small datasets
Redis	storage.redis	Production with RediSearch full-text and RedisTimeSeries
TimescaleDB	storage.timescale	Production with PostgreSQL JSONB and time-series hypertables

Redis

from hs_py.storage.redis import RedisAdapter, create_redis_client

r = await create_redis_client("redis://localhost:6379")
adapter = RedisAdapter(r)
await adapter.start()

Features: RediSearch indexes for filter queries, RedisTimeSeries for history, pub/sub for watch notifications.

TimescaleDB

from hs_py.storage.timescale import TimescaleAdapter, create_timescale_pool

pool = await create_timescale_pool("postgresql://localhost/haystack")
adapter = TimescaleAdapter(pool)
await adapter.start()  # Creates schema + hypertable

Features: JSONB entity storage with GIN indexes, filter AST → SQL pushdown, hypertable time-series, COPY-based bulk loading.

Architecture

src/hs_py/
  kinds.py            Haystack value types (Marker, Number, Ref, Coord, etc.)
  grid.py             Grid, Col, GridBuilder -- universal message format
  errors.py           Exception hierarchy (HaystackError, CallError, AuthError)
  auth.py             SCRAM-SHA-256 / PLAINTEXT client auth handshake
  auth_types.py       Authenticator protocol, SimpleAuthenticator, CertAuthenticator, StorageAuthenticator
  user.py             User model, Role enum, create_user(), SCRAM credential derivation
  bootstrap.py        Admin user bootstrap from env vars on startup
  client.py           Async HTTP client with all standard ops
  ops.py              HaystackOps base class with storage-backed op dispatch
  fastapi_server.py   FastAPI application factory, SCRAM middleware, user CRUD API, role enforcement
  metrics.py          MetricsHooks for transport-level observability
  tls.py              TLSConfig, SSL context builders, certificate generation
  security.py         Security hardening utilities
  watch.py            WatchState (server delta), WatchAccumulator (client merge)
  ws.py               Sans-I/O WebSocket wrapper (websockets library)
  ws_client.py        WebSocketClient, ReconnectingWebSocketClient, WebSocketPool
  ws_server.py        Standalone WebSocket server with SCRAM auth and batch dispatch
  ws_codec.py         Binary frame codec (4-byte header + JSON payload)
  encoding/
    json.py           JSON v3/v4 encode/decode via orjson
    zinc.py           Zinc text format encode/decode
    trio.py           Trio tagged record format
    csv.py            CSV export (encode-only, lossy)
    scanner.py        Shared Zinc value scanning
  filter/
    ast.py            Filter AST nodes (Has, Missing, Cmp, And, Or, Path)
    lexer.py          Filter expression tokenizer
    parser.py         Recursive descent parser
    eval.py           Filter evaluation against dicts/grids
  storage/
    protocol.py       StorageAdapter + UserStore protocols
    memory.py         In-memory adapter for testing
    redis.py          Redis + RediSearch + RedisTimeSeries adapter
    timescale.py      PostgreSQL/TimescaleDB adapter via asyncpg
  ontology/
    defs.py           Def and Lib frozen dataclasses
    namespace.py      Namespace container, symbol resolution
    taxonomy.py       Subtype tree, tag inheritance
    normalize.py      Normalization pipeline
    reflect.py        Dict-to-def reflection engine

Key Classes

Class	Module	Purpose
Client	client	Async HTTP client with SCRAM auth and all Haystack ops
WebSocketClient	ws_client	Persistent WebSocket client with batch and watch support
ReconnectingWebSocketClient	ws_client	Auto-reconnecting WebSocket client with exponential backoff
WebSocketPool	ws_client	Multiplexed channels over a single WebSocket connection
HaystackOps	ops	Storage-backed server operation handler for all 13 ops
WebSocketServer	ws_server	Standalone WebSocket server with SCRAM auth and push
Grid	grid	Universal Haystack message format (immutable)
GridBuilder	grid	Fluent builder for constructing grids
StorageAdapter	storage.protocol	Protocol for pluggable storage backends
UserStore	storage.protocol	Protocol for user management backends
User	user	Frozen user record with SCRAM credentials and role
Role	user	Permission enum: ADMIN, OPERATOR, VIEWER
StorageAuthenticator	auth_types	SCRAM authenticator backed by a UserStore
TLSConfig	tls	TLS certificate configuration
MetricsHooks	metrics	Optional observability callbacks
WatchState	watch	Server-side watch delta computation
WatchAccumulator	watch	Client-side watch delta merging
Namespace	ontology.namespace	Resolved ontology with taxonomy queries

Error Handling

All client methods raise from a common exception hierarchy:

from hs_py import HaystackError, CallError, AuthError, NetworkError

# HaystackError         Base for all haystack-py errors
#   CallError            Server returned an error grid
#   AuthError            Authentication failure
#   NetworkError         Transport-level failure

Configuration

Docker Compose

The included docker/docker-compose.yml provides a complete development stack:

docker compose -f docker/docker-compose.yml up -d

Services:

Service	Port	Description
server	8080	FastAPI Haystack server with SCRAM auth
redis	6379	Redis with RediSearch and RedisTimeSeries
timescaledb	5432	TimescaleDB (PostgreSQL 16)
redis-tls	6380	Redis with mTLS for TLS integration tests

Environment variables for the server:

Variable	Default	Description
REDIS_URL	redis://redis:6379	Redis connection URL
HAYSTACK_USER	admin	SCRAM username (legacy SimpleAuthenticator)
HAYSTACK_PASS	secret	SCRAM password (legacy SimpleAuthenticator)
HS_SUPERUSER_USERNAME	—	Admin username for bootstrap seeding
HS_SUPERUSER_PASSWORD	—	Admin password for bootstrap seeding

Seed Data

The _data/ directory contains Project Haystack example building datasets in JSON v4 format:

• Alpha — 2,032 entities (1 site, 184 equips, 1,846 points)
• Bravo — 1,077 entities (1 site, 149 equips, 918 points)

These are loaded automatically by the Docker server on startup.

Testing

make test          # 1,200+ unit tests
make lint          # ruff check + format verification
make typecheck     # mypy strict
make check         # lint + typecheck + test (all of the above)
make coverage      # tests with coverage report
make fix           # auto-fix lint/format issues
make docs          # sphinx-build

Docker Integration Tests

End-to-end testing against real services with full SCRAM authentication:

make docker-server         # Start Redis + FastAPI server stack
make docker-test-e2e       # 69 end-to-end tests (HTTP, WebSocket, auth, history, watch)
make docker-server-clean   # Tear down server stack

Storage Backend Tests

make docker-test              # Redis adapter integration tests
make docker-test-tls          # Redis mTLS integration tests
make docker-test-timescale    # 122 TimescaleDB integration tests

Cleanup

make docker-clean             # Remove Redis containers
make docker-clean-timescale   # Remove TimescaleDB containers

Requirements

• Python >= 3.13
• aiohttp >= 3.10
• orjson >= 3.10
• cryptography >= 42.0
• websockets >= 14.0
• Optional: FastAPI + Redis for server mode
• Optional: asyncpg for TimescaleDB backend
• Docker and Docker Compose for integration tests

License

MIT