ballotdata

The official Python client for the BallotData API: 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). 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.

pip install ballotdata

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
• Setting up the client
• What you can ask for
• Paging
• Ballots: contests, choices and voting rules
• Endorsers and endorsements
• The Endorsement Stream
• Polling instead of streaming
• Subscriptions
• Postbacks (signed webhooks)
• Errors
• Retries and timeouts
• Rate limits
• Async
• Types
• More documentation

Getting an API key

API accounts come with a purchase at ballotdata.net/pricing. Once it goes through, sign in at api.ballotdata.net and create a key in the console.

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

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)
)

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

Every argument of every method is listed in the reference. 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 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:

# 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:

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:

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

# 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.

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:

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.

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 EndorsementMessages (each with .event and .cursor).

The stream needs a key with stream:read and a purchase that includes the stream. The stream guide 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:

# 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.

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:

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 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:

All of them derive from BallotDataError.

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. Every code is listed in 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.

# 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:

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:

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 Literals (VoteMethod, Level, ErrorCode, …). The package ships py.typed, so mypy, Pyright and editors such as VS Code check your code and complete field names.

from ballotdata.types import Contest, EndorsementEvent

More documentation

• Reference: every method, argument and type.
• Stream guide: syncing, cursors, reconnecting, threads, serverless.
• Postbacks guide: receiving signed changes.
• Errors: every exception and code.
• API docs and the API reference.
• Every class and method has docstrings: try help(client.stream.connect).

License

MIT. Copyright BallotData LLC.