Metadata-Version: 2.5
Name: ballotdata
Version: 1.0.0
Summary: The official Python client for the BallotData API: elections, contests, endorsers, endorsements and a live stream of endorsement changes for every U.S. ballot.
Project-URL: Homepage, https://ballotdata.net
Project-URL: Documentation, https://github.com/EndorsedByOrg/pylib#readme
Project-URL: API reference, https://api.ballotdata.net/docs
Project-URL: Source, https://github.com/EndorsedByOrg/pylib
Project-URL: Changelog, https://github.com/EndorsedByOrg/pylib/blob/main/CHANGELOG.md
Author-email: BallotData <api@ballotdata.net>
License-Expression: MIT
License-File: LICENSE
Keywords: api,ballot,ballotdata,elections,endorsements,sse,voting
Classifier: Development Status :: 5 - Production/Stable
Classifier: Framework :: AsyncIO
Classifier: Intended Audience :: Developers
Classifier: Operating System :: OS Independent
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3 :: Only
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Typing :: Typed
Requires-Python: >=3.10
Requires-Dist: httpx<1,>=0.25
Description-Content-Type: text/markdown

# ballotdata

The official Python client for the [BallotData API](https://api.ballotdata.net/docs): U.S. ballots
(elections, contests and their choices), endorsers and endorsements, officeholders, and the live
**Endorsement Stream** of changes.

- Covers every endpoint of the API, with type hints for every response and parameter.
- Sync and async (`asyncio`) clients with the same methods.
- One dependency ([httpx](https://www.python-httpx.org/)). Python 3.10 or newer.
- Handles the details: retries with `Retry-After`, timeouts, typed errors, paging, and a stream that
  reconnects and resumes where it stopped.

```sh
pip install ballotdata
```

```python
from ballotdata import BallotData

client = BallotData()  # reads BALLOTDATA_API_KEY

contest = client.contests.get(4410, categories="labor")
for position in contest["endorsements"]:
    print(position["endorser"]["name"], [pick["label"] for pick in position["picks"]])
```

## Contents

- [Getting an API key](#getting-an-api-key)
- [Setting up the client](#setting-up-the-client)
- [What you can ask for](#what-you-can-ask-for)
- [Paging](#paging)
- [Ballots: contests, choices and voting rules](#ballots-contests-choices-and-voting-rules)
- [Endorsers and endorsements](#endorsers-and-endorsements)
- [The Endorsement Stream](#the-endorsement-stream)
- [Polling instead of streaming](#polling-instead-of-streaming)
- [Subscriptions](#subscriptions)
- [Postbacks (signed webhooks)](#postbacks-signed-webhooks)
- [Errors](#errors)
- [Retries and timeouts](#retries-and-timeouts)
- [Rate limits](#rate-limits)
- [Async](#async)
- [Types](#types)
- [More documentation](#more-documentation)

## Getting an API key

API accounts come with a purchase at [ballotdata.net/pricing](https://ballotdata.net/pricing). Once
it goes through, sign in at [api.ballotdata.net](https://api.ballotdata.net/login) and create a key
in the [console](https://api.ballotdata.net/console/keys).

A key covers what your account bought. Asking about another state answers `403 outside_purchase`.
Keys have scopes: `data:read` for the data endpoints, `stream:read` for the stream, polling and
reading subscriptions, and `subscriptions:write` to change subscriptions.

Keep keys on servers. Never put one in a browser page or a mobile app.

## Setting up the client

```python
import os
from ballotdata import BallotData

client = BallotData(
    api_key=os.environ["BALLOTDATA_API_KEY"],  # the default, so you can leave it out
    timeout=30,  # seconds to wait for each response (default 30)
    max_retries=2,  # retries for 429, 503, network errors (default 2)
)
```

| Argument | Default | What it does |
| --- | --- | --- |
| `api_key` | `BALLOTDATA_API_KEY` | Your key, `ebk_…`. |
| `base_url` | `BALLOTDATA_API_URL`, then `https://api.ballotdata.net` | Where requests go. |
| `timeout` | `30` | Seconds to wait for each response. |
| `max_retries` | `2` | Retries after 429, 503, network errors and timeouts, and 500/502/504 on GET, PUT and DELETE. `0` turns retries off. |
| `http_client` | a new `httpx.Client` | Your own `httpx.Client` (proxies, custom TLS, tests). You close it yourself. |
| `default_headers` | `{}` | Extra headers on every request. |
| `on_response` | none | Called with a `ResponseInfo` (`method`, `url`, `status`, `request_id`, `rate_limit`) after every response. |

Every method also takes `timeout`, `max_retries` and `extra_headers` keyword arguments for that one
call. The client holds a connection pool: use it as a context manager (`with BallotData() as
client:`) or call `client.close()` when you're done.

## What you can ask for

| Method | Endpoint | Returns |
| --- | --- | --- |
| `client.states.list()` | `GET /v1/states` | `list[State]` |
| `client.coverage.list(upcoming=)` | `GET /v1/coverage` | `list[Coverage]` |
| `client.elections.list(...)` | `GET /v1/elections` | pages of `ElectionSummary` |
| `client.contests.list(...)` | `GET /v1/contests` | pages of `Contest` |
| `client.contests.get(id, ...)` | `GET /v1/contests/{id}` | `ContestDetail` (with every endorser's position) |
| `client.endorser_categories.list(lang=)` | `GET /v1/endorser-categories` | `list[EndorserCategory]` |
| `client.endorsers.list(...)` | `GET /v1/endorsers` | pages of `Endorser` |
| `client.endorsers.get(slug, lang=)` | `GET /v1/endorsers/{slug}` | `EndorserDetail` |
| `client.endorsements.list(...)` | `GET /v1/endorsements` | pages of `Endorsement` (the basics) |
| `client.endorsements.get(id, lang=)` | `GET /v1/endorsements/{id}` | `EndorsementDetail` (the full record) |
| `client.officeholders.list(...)` | `GET /v1/officeholders` | pages of `Officeholder` |
| `client.officeholders.get(person_id, lang=)` | `GET /v1/officeholders/{id}` | `OfficeholderDetail` |
| `client.stream.connect(...)` | `GET /v1/stream` | a live `EndorsementStream` |
| `client.events.list(...)` | `GET /v1/events` | `EventsPage` |
| `client.events.follow(...)` | `GET /v1/events` (long polling) | changes, forever |
| `client.subscriptions.list()` | `GET /v1/subscriptions` | `list[Subscription]` |
| `client.subscriptions.create(name=, filter=)` | `POST /v1/subscriptions` | `Subscription` |
| `client.subscriptions.get(id)` | `GET /v1/subscriptions/{id}` | `Subscription` |
| `client.subscriptions.update(id, name=, filter=)` | `PUT /v1/subscriptions/{id}` | `Subscription` |
| `client.subscriptions.delete(id)` | `DELETE /v1/subscriptions/{id}` | `None` |
| `verify_postback_signature()`, `construct_postback()` | your endpoint | checks and parses postbacks |

Every argument of every method is listed in the [reference](docs/reference.md). Arguments are
keyword-only and named as in the API, except `from`, a Python keyword, which is `from_`. List
arguments take a list or a comma-separated string: `states=["CA", "NV"]` and `states="CA,NV"` mean
the same. Ballot text comes in Spanish with `lang="es"` where the state publishes it.

Responses are plain dicts with the API's field names, typed with `TypedDict`, so the
[API reference](https://api.ballotdata.net/docs/reference) reads the same as your code.

## Paging

List methods return the first page, a `SyncPage`. Its `data` holds the page's items and
`next_cursor` the cursor for the next page. Iterating the page walks every item on every page,
fetching pages as you go:

```python
# One page
page = client.endorsements.list(states="CA", limit=500)
page.data, page.next_cursor

# Every item, across every page
for endorsement in client.endorsements.list(states="CA", limit=500):
    save(endorsement)

# Every item in a list (pass a maximum for big lists)
everything = client.endorsers.list(categories="labor").to_list(max=1000)

# Page by page
for page in client.contests.list(elections="ca-2026-11-03-general").iter_pages():
    print(len(page.data), page.next_cursor)
```

To page by hand, pass the last page's `next_cursor` as `cursor`, or call `page.next_page()`.

## Ballots: contests, choices and voting rules

A contest's `vote_method` says how it is voted, and its choices are in ballot order:

| `vote_method` | Meaning | Look at |
| --- | --- | --- |
| `plurality` | Vote for one | `votes_allowed` (1) |
| `multi_seat` | Vote for up to N | `votes_allowed` |
| `ranked_choice` | Rank up to N | `ranks_allowed`, and each pick's `rank` |
| `approval` | Vote for as many as you like | |
| `yes_no` | A measure or retention question | each choice's `measure_answer` |

```python
election = client.elections.list(states="CA", types="general", upcoming=True).data[0]

for contest in client.contests.list(elections=election["slug"]):
    if contest["vote_method"] == "ranked_choice":
        rule = f"rank up to {contest['ranks_allowed']}"
    else:
        rule = f"vote for {contest['votes_allowed']}"
    print(f"{contest['title']} ({rule}):", ", ".join(c["label"] for c in contest["choices"]))
```

`client.contests.get(id)` adds every endorser's position. Each position's `picks` hold one pick for
vote-for-one, up to N for vote-for-N, the ranking for ranked choice, or Yes/No for a measure:

```python
contest = client.contests.get(4410, categories=["labor", "editorial"])
for position in contest["endorsements"]:
    marks = [f"{p['rank']}. {p['label']}" if p["rank"] else p["label"] for p in position["picks"]]
    print(f"{position['endorser']['name']}: {'  '.join(marks)}")
```

## Endorsers and endorsements

```python
# The kinds of endorser, for categories filters
categories = client.endorser_categories.list()

# Find endorsers by category, home state (or "national"), election, or name
unions = client.endorsers.list(categories="labor", states="CA", q="council").data

# One endorser and the elections they endorsed in
endorser = client.endorsers.get("sf-labor-council")
```

`client.endorsements.list()` returns one row per endorser and choice with the basics: the endorser,
contest and election (ids, slugs and names), the choice and its rank. It is the way to sync
everything. A ranked-choice endorsement of three candidates is three rows with ranks 1 to 3.
`client.endorsements.get(id)` has the full record: the statement and source, the endorser's details,
the contest with its district, election and voting rules, and the choice with its candidate.

```python
e = client.endorsements.get(90211)
print(e["endorser"]["category"]["name"], e["contest"]["vote_method"], e["statement"], e["source_url"])
```

`group_by_contest()` turns rows back into positions:

```python
from ballotdata import group_by_contest

rows = client.endorsements.list(endorsers="sf-labor-council", elections="ca-2026-11-03-general", limit=500)
for group in group_by_contest(rows):
    print(group["contest"]["title"], [p["label"] for p in group["positions"][0]["picks"]])
```

## The Endorsement Stream

`client.stream.connect()` opens `GET /v1/stream` and yields every endorsement change that matches
your filter (or saved subscription), within seconds of it happening.

```python
with client.stream.connect(states=["CA", "NV"], categories="labor", cursor=load_cursor(), on_cursor=save_cursor) as stream:
    for message in stream:
        if message.type == "ready":
            print("connected")
        elif message.type == "endorsement":
            event = message.event  # event["type"]: endorsement.created | updated | deleted
            print(event["type"], event["endorser"]["name"], event["choice"]["label"], event["contest"]["title"])
```

The stream keeps its place for you:

- Each change moves the **cursor**. Save it in `on_cursor` and pass it back as `cursor` after a
  restart to continue exactly where you stopped. Without a cursor the stream starts at
  `from_="latest"` (or `"earliest"`, or `since=datetime`).
- It **reconnects** with the cursor after the API's hourly `reconnect`, a dropped connection, a
  connection that goes silent for a minute (`idle_timeout`), or a 429/503 (after `Retry-After`).
  Nothing is missed or repeated.
- `checkpoint` events move the cursor past changes your filter skipped. You don't see them; the
  cursor just moves.
- If the cursor is older than the stream keeps, iteration raises `CursorExpiredError`. Catch up with
  `client.endorsements.list()` and start again with `from_="latest"`.
- `stream.close()` ends the loop quietly, even from another thread. Leaving the `with` block closes
  the connection.

`client.stream.endorsements(...)` takes the same arguments and yields only the `EndorsementMessage`s
(each with `.event` and `.cursor`).

The stream needs a key with `stream:read` and a purchase that includes the stream. The
[stream guide](docs/stream.md) covers syncing first, recovering from an expired cursor, running in
a background thread, and serverless functions.

## Polling instead of streaming

Where a connection can't stay open (serverless functions, cron, some proxies), read the same changes
from `GET /v1/events`:

```python
# One page; wait up to 20 s for something new
page = client.events.list(subscription="CA labor", cursor=cursor, wait=20)
# page["data"], page["next_cursor"], page["has_more"]

# Or forever
for message in client.events.follow(states="CA", cursor=load_cursor()):
    handle(message.event)
    save_cursor(message.cursor)
```

## Subscriptions

A subscription is a saved filter. Use it by id or name as `subscription=` on the stream and on events.

```python
sub = client.subscriptions.create(name="CA labor", filter={"categories": ["labor"], "states": ["CA"]})
subs = client.subscriptions.list()
client.subscriptions.update(sub["id"], name="CA labor", filter={"categories": ["labor"], "states": ["CA", "NV"]})
client.subscriptions.delete(sub["id"])

with client.stream.connect(subscription="CA labor") as stream:
    ...
```

A filter's fields combine with AND and the values within a field with OR: `categories`, `endorsers`,
`endorser_states`, `states`, `elections`, `contest_kinds`, `levels`, and `types`
(`created`, `updated`, `deleted`). An account can have up to 50.

## Postbacks (signed webhooks)

With postbacks, BallotData POSTs each batch of changes to your HTTPS endpoint, signed with your
account's signing secret. Check the signature against the **raw** body:

```python
import os
from flask import Flask, request
from ballotdata import PostbackSignatureError, construct_postback

app = Flask(__name__)


@app.post("/ballotdata")
def ballotdata_postback():
    try:
        postback = construct_postback(
            request.get_data(), request.headers.get("BallotData-Signature"), os.environ["BALLOTDATA_SIGNING_SECRET"]
        )
    except PostbackSignatureError:
        return "", 401
    for change in postback["changes"]:
        handle(change)  # change["id"] drops repeats
    return "", 204
```

`verify_postback_signature(raw_body, header, secret)` returns True or False instead of raising.
`sign_postback(raw_body, secret)` makes a header for testing your endpoint. See the
[postbacks guide](docs/postbacks.md) for Django and FastAPI.

## Errors

Every error the API answers with is an `APIError` with `status`, `code` (such as
`outside_purchase`), `message` and `request_id`. Subclasses let you catch by kind:

| Class | Status | When |
| --- | --- | --- |
| `BadRequestError` | 400 | A parameter isn't valid. |
| `AuthenticationError` | 401 | No key, or a bad, revoked or expired one. |
| `PermissionDeniedError` | 403 | Missing scope, IP allowlist, account not active, or data you haven't bought. |
| `NotFoundError` | 404 | No such contest, endorser, officeholder or subscription. |
| `ConflictError` | 409 | A subscription with that name exists. |
| `CursorExpiredError` | 410 | The stream position is too old. |
| `RateLimitError` | 429 | Over a rate limit, or too many open streams. Has `retry_after`. |
| `InternalServerError` | 5xx | A problem at BallotData. |
| `APIConnectionError` | | The API couldn't be reached. |
| `APITimeoutError` | | No answer within `timeout` (a kind of `APIConnectionError`). |

All of them derive from `BallotDataError`.

```python
from ballotdata import APIError, NotFoundError

try:
    contest = client.contests.get(contest_id)
except NotFoundError:
    contest = None
except APIError as err:
    print(err.code, err.message, err.request_id)
    raise
```

Quote the `request_id` when you write to [sales@ballotdata.net](mailto:sales@ballotdata.net). Every
code is listed in [docs/errors.md](docs/errors.md).

## Retries and timeouts

The client retries, up to `max_retries` times (default 2):

- 429 and 503 on any request, after the `Retry-After` the API sends (when it's a minute or less: a
  daily limit isn't worth waiting out).
- Network errors and timeouts, and 500, 502 and 504, on GET, PUT and DELETE. A POST that may have
  reached the API isn't repeated.

```python
# Per call
client.endorsements.list(states="CA", timeout=60, max_retries=5)
```

## Rate limits

Your plan sets requests per minute (and per day on some plans). The latest headers are on
`client.last_response.rate_limit` (`limit`, `remaining`, `reset`, `daily_limit`, `daily_remaining`),
and `on_response` sees every response:

```python
def watch(info):
    if info.rate_limit.remaining is not None and info.rate_limit.remaining < 5:
        print("Nearly out of requests this minute")


client = BallotData(on_response=watch)
```

## Async

`AsyncBallotData` has the same resources and arguments. Methods that answer once are coroutines.
List methods can be awaited for the first page or iterated with `async for` for every item:

```python
import asyncio
from ballotdata import AsyncBallotData


async def main():
    async with AsyncBallotData() as client:
        contest = await client.contests.get(4410)

        page = await client.endorsements.list(states="CA")  # the first page
        async for endorsement in client.endorsements.list(states="CA"):  # every item
            ...

        async with client.stream.connect(subscription="CA labor", cursor=cursor) as stream:
            async for message in stream:
                if message.type == "endorsement":
                    await handle(message.event)


asyncio.run(main())
```

To stop an async stream from another task, `await stream.aclose()` (or cancel the task reading it).

## Types

Every response and parameter has a type in `ballotdata.types`: response objects (`Contest`,
`Endorsement`, `EndorsementEvent`, …), the subscription `Filter`, and enums as `Literal`s
(`VoteMethod`, `Level`, `ErrorCode`, …). The package ships `py.typed`, so mypy, Pyright and editors
such as VS Code check your code and complete field names.

```python
from ballotdata.types import Contest, EndorsementEvent
```

## More documentation

- [Reference](docs/reference.md): every method, argument and type.
- [Stream guide](docs/stream.md): syncing, cursors, reconnecting, threads, serverless.
- [Postbacks guide](docs/postbacks.md): receiving signed changes.
- [Errors](docs/errors.md): every exception and code.
- [API docs](https://api.ballotdata.net/docs) and the [API reference](https://api.ballotdata.net/docs/reference).
- Every class and method has docstrings: try `help(client.stream.connect)`.

## License

MIT. Copyright BallotData LLC.
