groundhog-sdk-python

Python client for the Groundhog M1 API. The client uses a Unix socket or HTTPS.

The SDK connects to an existing Groundhog service. It does not install, start, stop, or manage the Groundhog process.

This source tree builds groundhog-sdk 0.3.0.

Install

python -m pip install groundhog-sdk

Connect through a Unix socket

from groundhog_sdk import Ground, upserted

ground = Ground("unix:/var/run/groundhog.sock")
receipt = ground.send(
    "stripe",
    [upserted("customers", "cus_123", {"id": "cus_123"})],
    batch_id="stripe/customers/page-1",
)

Relative Unix socket paths also work. The default endpoint is unix:data/ground.sock.

Connect through HTTPS

from groundhog_sdk import Ground

ground = Ground(
    "https://groundhog.example.com",
    token="service-token",
)
streams = ground.streams()

The HTTPS transport uses standard certificate and hostname verification. An endpoint can include a port and base path, such as https://groundhog.example.com:8443/service.

Use explicit transport settings

Use TransportConfig to keep the endpoint and timeout in one value.

from groundhog_sdk import Ground, TransportConfig

transport = TransportConfig(
    "https://groundhog.example.com",
    timeout=10,
)
ground = Ground(transport=transport, token="service-token")

Ground also reads GROUND_URL and GROUND_TOKEN. Explicit arguments take precedence over these environment variables.

API operations

Both transports provide the same operations:

• send commits one atomic event batch.
• events reads an authoritative replay page.
• streams reads authoritative stream summaries from the durable log.
• query runs one relation query or a named multi-query request.
• catalog lists published relation metadata.
• catalog_relation reads one relation declaration.
• projection_status reads one projection's publication status.

The event constructors upserted, deleted, and native support both transports. The SDK validates connector-owned fields before it sends a batch.

Groundhog stores and serves the durable event log. Query-enabled deployments also serve Groundhog-owned indexed relations. The Query API uses typed JSON. It does not accept SQL.

Documentation ownership

The docs/sdks/python/ directory owns the published Python SDK documentation and examples. Buildkite copies that path to GroundSystems/groundhog after changes reach main.

Replay events

Use events to read events in authoritative order. Save next_after as the cursor for the next request.

page = ground.events(source="stripe", stream="customers", limit=100)
for event in page.events:
    apply_to_view(event)

next_page = ground.events(
    after=page.next_after,
    source="stripe",
    stream="customers",
    limit=100,
)

Applications can store derived views in their own database or analytical system. Replay provides the durable input for each derived view.

Query a published snapshot

Use query with one relation query object. The SDK adds the version 1 envelope and uses published consistency by default.

response = ground.query(
    {
        "relation": "groundhog.events",
        "select": ["event_id", "source", "stream", "kind"],
        "filter": {"op": "eq", "field": "source", "value": "stripe"},
        "order_by": [{"field": "event_id", "direction": "asc"}],
        "limit": 100,
    }
)

for row in response.results[0].rows:
    print(row)

QueryResponse.snapshot is the immutable query receipt. It contains the exact event frontier, chain head, projection versions, and schema versions used for the response.

For several queries on one snapshot, pass named query objects to query_many:

response = ground.query_many(
    [
        {
            "name": "events",
            "query": {
                "relation": "groundhog.events",
                "select": ["event_id", "kind"],
                "limit": 20,
            },
        }
    ]
)

For keyset pagination, copy the original relation query with the returned cursor. continuation does not change the original request object.

continued = response.results[0].continuation(original_query)
if continued is not None:
    next_response = ground.query(continued)

Use the typed request models when request construction must fail before any network operation:

from groundhog_sdk import QueryRequest, RelationQuery

request = QueryRequest.single(
    RelationQuery.from_dict(
        {
            "relation": "agents.runs_current",
            "select": ["run_id", "status"],
            "limit": 100,
        }
    )
)
response = ground.query(request)

The typed models enforce the closed Query v1 request schema. They reject unknown members, unsupported operators, mixed row and aggregate forms, and invalid cursor use.

Use the Agent Operations saved queries

Version 0.3.0 packages the fixed Agent Operations v1 saved-query contract. Resolve its client-side templates before sending each request:

from groundhog_sdk import load_agent_operations_saved_queries

library = load_agent_operations_saved_queries()
saved = library.get("daily_usage_by_workspace")
requests = saved.render_step(
    "usage",
    parameters={
        "workspace_id": "demo",
        "from_day": "2026-08-01",
        "through_day": "2026-08-13",
    },
)
response = ground.query(requests[0])

Later workflow steps accept prior QueryResponse values by step ID. Result references are typed, and result unions are flattened and deduplicated before serialization.

Read the Catalog

Use catalog to list published relation schemas and metadata. Use catalog_relation for one canonical relation name.

catalog = ground.catalog()
for relation in catalog.relations:
    print(relation.relation, relation.row_count)

events = ground.catalog_relation("groundhog.events")
print(events.relation.fields)

Read projection status

Use projection_status with a canonical projection name. The method maps to GET /v1/projections/{projection}/status.

status = ground.projection_status("agent_operations")
print(status.projection.version)
print(status.snapshot.frontier_event_count)
print(status.freshness.status, status.freshness.lag_events)

The response contains the immutable SnapshotReceipt and the observed ProjectionFreshness. A failed projection includes a non-empty freshness.failure value.

Enumerate streams

Use streams to read the source, stream, event count, and stream frontier. Each page also includes the selected snapshot frontier and a pagination cursor.

page = ground.streams(source="stripe", limit=100)
for stream in page.streams:
    print(
        stream.source,
        stream.stream,
        stream.event_count,
        stream.frontier_event_id,
    )

For another page, pass next_after and snapshot_through_event_id from the first page. Use the same source filter on each request.

if page.next_after is not None:
    next_page = ground.streams(
        source="stripe",
        after=page.next_after,
        through=page.snapshot_through_event_id,
        limit=100,
    )

Handle errors

All SDK exceptions inherit from GroundError. Remote errors expose the HTTP status as status, the machine code as code, and the descriptive text as message. The string form of the exception equals message.

The server can change descriptive text without changing its contract. Use code, not message, for control flow. General API errors retain unknown codes. Query, Catalog, and projection-status responses reject unknown codes.

Query, Catalog, and projection-status HTTP errors raise QueryError. Inspect its stable code attribute for values such as invalid_query, relation_not_found, cursor_expired, query_timeout, and query_unavailable.

The body attribute contains the complete server response. A ValidationError also exposes indexed event errors through errors. Local validation errors and responses from older servers can have code set to None.