themeparks

A typed, modern Python SDK for the ThemeParks.wiki API. Built on httpx and pydantic v2, with first-class sync and async clients, default-on caching, and ergonomic helpers for the common workflows (list destinations, walk a park's children, fetch live wait times, pull a date-ranged schedule).

📚 Full documentation, API reference, and cookbook

Install

pip install themeparks

Python 3.9+ is required.

Print live wait times (sync)

from themeparks import ThemeParks, current_wait_time

MAGIC_KINGDOM = "75ea578a-adc8-4116-a54d-dccb60765ef9"

with ThemeParks() as tp:
    live = tp.entity(MAGIC_KINGDOM).live()
    for entry in sorted(live.liveData or [], key=lambda e: e.name):
        wait = current_wait_time(entry)
        if wait is None:
            print(f"{entry.name:50s}  --")
        else:
            print(f"{entry.name:50s}  {wait:>3d} min")

Sample output:

Astro Orbiter                                       15 min
Big Thunder Mountain Railroad                       45 min
Buzz Lightyear's Space Ranger Spin                  20 min
...

Print live wait times (async)

import asyncio
from themeparks import AsyncThemeParks, current_wait_time

MAGIC_KINGDOM = "75ea578a-adc8-4116-a54d-dccb60765ef9"

async def main() -> None:
    async with AsyncThemeParks() as tp:
        live = await tp.entity(MAGIC_KINGDOM).live()
        for entry in sorted(live.liveData or [], key=lambda e: e.name):
            wait = current_wait_time(entry)
            if wait is None:
                print(f"{entry.name:50s}  --")
            else:
                print(f"{entry.name:50s}  {wait:>3d} min")

asyncio.run(main())

The sync and async clients mirror each other method-for-method. Only the call sites need await and you use async with instead of with.

Just the open rides, sorted longest-wait first

from themeparks import ThemeParks, current_wait_time

with ThemeParks() as tp:
    live = tp.entity("75ea578a-adc8-4116-a54d-dccb60765ef9").live()
    waits = [
        (entry.name, current_wait_time(entry))
        for entry in live.liveData or []
    ]
    waits = [(name, w) for name, w in waits if w is not None]
    waits.sort(key=lambda pair: pair[1], reverse=True)
    for name, wait in waits:
        print(f"{wait:>3d} min  {name}")

Client options

Both ThemeParks and AsyncThemeParks take the same keyword-only options:

Option	Type	Default	Purpose
`base_url`	`str`	`https://api.themeparks.wiki/v1`	API base URL (point at a mock / staging if you need to).
`api_key`	`str \| None`	`None`	Sent as the `x-api-key` header. Needed for anything beyond the free tier: deeper history, higher rate limits.
`user_agent`	`str \| None`	`themeparks-sdk-py/<version>`	Sent as the `User-Agent` header. Set this to identify your app.
`timeout`	`float` (seconds)	`10.0`	Per-request timeout.
`retry`	`RetryConfig \| None`	`RetryConfig(max_retries=3, respect_429=True, max_retry_after=120.0, respect_remaining=True)`	Retry/backoff behavior. `max_retries` is N retries beyond the first attempt (so N+1 total calls). `max_retry_after` is the TOTAL the client will block for within one call, across both the shared 429 gate and any spent-window wait. Past a single `Retry-After` that long you get `RateLimitError` instead of a silent wait.
`cache`	`Cache \| CacheConfig \| bool \| None`	`True` (in-memory LRU)	See Caching below. `False` disables caching entirely.

Example:

import os

from themeparks import ThemeParks, RetryConfig

tp = ThemeParks(
    api_key=os.environ["THEMEPARKS_API_KEY"],
    user_agent="my-app/1.2.3 (+https://example.com)",
    timeout=15.0,
    retry=RetryConfig(max_retries=5, respect_429=True),
)

Without a key you get the anonymous tier: the most recent seven days of history and the lowest rate limit. Keys are issued from your account at api.themeparks.wiki.

Ergonomic helpers

from datetime import date
from themeparks import ThemeParks

with ThemeParks() as tp:
    # Directory lookup
    wdw = tp.destinations.find("waltdisneyworldresort")
    print(wdw.id, wdw.name)

    # Walk a destination and yield every descendant (parks, lands, attractions, ...)
    for child in tp.entity(wdw.id).walk():
        print(child.entityType, child.name)

    # Schedule across a date range (stitches monthly responses and filters)
    mk = "75ea578a-adc8-4116-a54d-dccb60765ef9"
    entries = tp.entity(mk).schedule.range(date(2026, 5, 1), date(2026, 5, 31))
    print(f"{len(entries)} schedule entries")

Reading every queue type

current_wait_time covers the standby-queue case. There are six queue variants in total, and an attraction may have more than one populated at once (e.g. STANDBY + SINGLE_RIDER + PAID_STANDBY for a Lightning Lane ride).

Each variant is exposed as an attribute on entry.queue. All are Optional — None if that queue type isn't offered for the attraction:

Attribute	Type	Fields
`queue.STANDBY`	`StandbyQueue`	`waitTime: float \| None`
`queue.SINGLE_RIDER`	`SingleRiderQueue`	`waitTime: float \| None`
`queue.PAID_STANDBY`	`PaidStandbyQueue`	`waitTime: float \| None`
`queue.RETURN_TIME`	`ReturnTimeQueue`	`state`, `returnStart`, `returnEnd`
`queue.PAID_RETURN_TIME`	`PaidReturnTimeQueue`	`state`, `returnStart`, `returnEnd`, `price`
`queue.BOARDING_GROUP`	`BoardingGroupQueue`	`allocationStatus`, `currentGroupStart`, `currentGroupEnd`, `nextAllocationTime`, `estimatedWait`

waitTime is a float because the API's schema declares it a JSON number, not an integer, so a 45-minute wait arrives as 45.0 and a raw row dumped to JSON says 45.0. Use current_wait_time(entry) or int(...) when you want an int, and format with {wait:.0f}, not {wait:d}, if you print the field directly.

Direct access

from themeparks import ThemeParks

with ThemeParks() as tp:
    live = tp.entity("75ea578a-adc8-4116-a54d-dccb60765ef9").live()
    for entry in live.liveData or []:
        if entry.queue is None:
            continue

        # Standby
        if entry.queue.STANDBY and entry.queue.STANDBY.waitTime is not None:
            print(f"{entry.name}: standby {entry.queue.STANDBY.waitTime} min")

        # Lightning Lane / paid line
        if entry.queue.PAID_RETURN_TIME:
            prt = entry.queue.PAID_RETURN_TIME
            price = prt.price.formatted if prt.price else "?"
            print(f"{entry.name}: Lightning Lane {price}, return {prt.returnStart} → {prt.returnEnd}")

        # Boarding group
        if entry.queue.BOARDING_GROUP:
            bg = entry.queue.BOARDING_GROUP
            print(
                f"{entry.name}: boarding group {bg.currentGroupStart}–{bg.currentGroupEnd}, "
                f"~{bg.estimatedWait} min wait, status {bg.allocationStatus}"
            )

        # Return-time only (no paid component)
        if entry.queue.RETURN_TIME:
            rt = entry.queue.RETURN_TIME
            print(f"{entry.name}: virtual queue {rt.returnStart} → {rt.returnEnd} ({rt.state})")

Generic iteration

If you'd rather not branch on every variant, iter_queues(entry) flattens all populated queue types into one sequence of dicts keyed by type:

from themeparks import ThemeParks, iter_queues

with ThemeParks() as tp:
    live = tp.entity("75ea578a-adc8-4116-a54d-dccb60765ef9").live()
    for entry in live.liveData or []:
        for q in iter_queues(entry):
            # q is a dict, e.g. {"type": "STANDBY", "waitTime": 35.0}
            #                or {"type": "PAID_RETURN_TIME", "state": "AVAILABLE", ...}
            print(entry.name, q)

The type key matches the API's variant name (STANDBY, SINGLE_RIDER, RETURN_TIME, PAID_RETURN_TIME, BOARDING_GROUP, PAID_STANDBY). The remaining keys are whatever fields that variant carries.

Other helpers

parse_api_datetime(value, timezone) parses any API date/time string into a timezone-aware datetime, honoring the entity's IANA timezone for naive inputs.

Rate limits

client.rate_limit is a read-only property, not a constructor option.

The API meters requests per minute, and history requests again per hour. Both are advertised on every response that can carry them, and the client reads them:

with ThemeParks(api_key=KEY) as tp:
    tp.entity(park_id).live()

    print(tp.rate_limit.rest.remaining)          # 299
    print(tp.rate_limit.rest.seconds_until_reset())
    print(tp.rate_limit.history.remaining)       # on a history call

None means the server did not say, never "nothing left". Use .exhausted, which is true only when the server actually said zero.

Which figures you get depends on the response:

The per-minute figures ride most responses, anonymous ones included.
The hourly history figures are withheld from anything a shared cache may store, because they are per-caller and a cache would hand one caller's budget to another. In practice you get them on calls made with a key.
An unmetered plan advertises nothing at all.

A response served from a cache is ignored entirely. Its figures belong to whoever populated the entry and its countdown is already wrong: a cached remaining: 0 would otherwise make the client sleep out someone else's window.

The client also acts on what it reads. When a response says the window is spent, the next request waits for the advertised reset rather than sending a request that is certain to be refused, and to cost a unit of budget being refused. Turn that off with RetryConfig(respect_remaining=False).

A 429 is held once for the whole client. The wait belongs to the caller, not to whichever request happened to meet it, so it goes on a shared gate with a little jitter. Without that, ten concurrent requests each sleep their own copy of Retry-After and then all retry at the same instant, re-tripping the limit together.

History

tp.entity(id).history reads the archive. Both methods page for you and yield rows as they arrive, so a resort's five years never has to fit in memory.

from themeparks import ThemeParks

DISNEYLAND = "7340550b-c14d-4def-80bb-acdb51d49a66"

with ThemeParks(api_key=KEY) as tp:
    history = tp.entity(DISNEYLAND).history

    # What exists, and what your key may read. Same three fields whether the
    # id is a park or a single ride. `final_through` is the newest day whose
    # the archive has recorded: store through that, ask for the rest later.
    span = history.span()
    print(span.archive_from, span.recorded_to, span.retrievable_through)
    print(span.final_through)

    # One summary row per park-local day, as (entity id, row).
    for entity_id, row in history.days(span.archive_from, span.retrievable_through):
        print(row.date, entity_id, row.operatingMinutes, row.standby.p50 if row.standby else None)

    # Every recorded change on one day, and the state before the first of them.
    changes = history.changes("2026-09-20")
    for entity_id, row in changes:
        print(row.time, entity_id, row.status)
    for entity_id, opening in changes.opening.items():
        print(entity_id, "at the start of the day:", opening.status)

A day rebuilds from opening plus the rows. Each row is the entity's complete state from its time until the next row. changes.opening is the state in force before the first row, keyed by entity id, so the minutes between midnight and a ride's first change have a status too: a ride still running from the night before, say. It covers every entity in the response, including one that did not change all day. Iterating changes() yields exactly what it always did, and reading opening costs no extra request.

Ask the park, not the rides. Both history endpoints answer every entity in a park in one request. Pulling the same data ride by ride is around a hundred times more calls for a large resort, against the same budget. Pass a park id and you are on the cheap path without having to know the expensive one exists.

History has its own hourly budget, separate from the per-minute rate limit. A large backfill will hit it, and the wait can be most of an hour because that is when the window rolls. The client will not sleep through that: past retry.max_retry_after (120s) it stops retrying, and the history layer turns the result into BudgetExhaustedError (a RateLimitError) carrying retry_after, so you can checkpoint and come back:

from themeparks import BudgetExhaustedError

try:
    for entity_id, row in history.days(start, end):
        write(entity_id, row)
        last_day = row.date
except BudgetExhaustedError as exc:
    checkpoint(last_day)
    print(f"resume in {exc.retry_after:.0f}s")

Or skip the code: there is a command

Installing the library installs themeparks-backfill, which does all of the above and stops before the walls:

# How far back it reaches is your plan, so set the key first: without one you get
# the 7 days anonymous access allows, and the run still succeeds, quietly.
export THEMEPARKS_API_KEY=tpw_your_key

themeparks-backfill "Disneyland Park"           # a park, by name or id
themeparks-backfill "Walt Disney World Resort"  # a destination: every park in it
themeparks-backfill --list disney               # find an id. This part needs no key.

It reads how far back your own key may ask and starts there, writes NDJSON or --format csv, names every row with the park and the entity, records what it has done so re-running never duplicates a file, and exits 75 when the hourly history budget runs out so a scheduler retries rather than alerts.

themeparks-backfill "Epcot" --since 2025-01-01                     # not the whole archive
themeparks-backfill "Epcot" --since 2025-01-01 --until 2025-12-31  # one year, both days inclusive

Run it again to bring a file up to date. A finished park is carried forward from the day after its last one, so the same command in a nightly cron appends the new days and nothing else. Only final days are written: today's row is the day so far, and the archive records days 2 to 3 behind live data, so the newest days can still change. A run stops at the newest final day (span().final_through) and says so, and the next run adds the rest. Each day is fetched once, as the archive recorded it.

Fetching a range again. The archive can occasionally re-record past days, for example when a park's feed is repaired. A file never rewrites rows it already holds, so to pick up a correction, download the affected days into a separate directory and replace those (entityId, date) rows where you load the data, or start the file again:

themeparks-backfill "Epcot" --since 2026-06-01 --until 2026-06-30 --out ./refetch
themeparks-backfill "Epcot" --overwrite              # or: the whole file again

Stopping a run at any point is safe. The state file is written after every page with the size of the file at that moment, and the next run first cuts off anything written after it, so no day is ever appended twice. Ctrl-C and SIGTERM exit 130 and 143; a killed run needs nothing either. Two runs on the same park and --out at once are refused.

--since applies when a file is started. Later runs continue that file and accept the same --since, or a later one, such as a cron line computing "30 days ago". One earlier than the file's first day, or one that would leave a gap, is refused rather than ignored: pass --overwrite, or a different --out. A fixed --since older than your key's window starts the file at the first day your key can read, and the same command line keeps working every night.

A file is never continued past a gap. If the day it would continue from is older than your key can read, because a cron missed more days than your window or a plan lapsed, the run is refused with exit 1 and the file is left alone.

Files written by 4.0.x ended on today, so their newest rows can be partial. The first run of this version removes the rows from the last week of such a file and fetches those days again, final this time. Everything else in the file is left exactly as it was. A 4.0 file that lies wholly inside that week, as every anonymous 7-day file does, is simply downloaded again.

python -m themeparks.backfill is the same thing, which is the one to use if pip install --user put the script somewhere off your PATH. themeparks-backfill --help has the rest.

Low-level escape hatch

Every ergonomic helper is built on top of tp.raw, which is a thin, typed 1:1 wrapper over the OpenAPI operations. Use it directly when you want the raw response shape:

with ThemeParks() as tp:
    live = tp.raw.get_entity_live("75ea578a-adc8-4116-a54d-dccb60765ef9")
    dests = tp.raw.get_destinations()
    children = tp.raw.get_entity_children(wdw.id)

The raw methods return pydantic models, so you still get full type checking and attribute access.

Error handling

All SDK errors inherit from ThemeParksError. The ones you will want to catch in application code:

from themeparks import ThemeParks, APIError, RateLimitError, NetworkError, TimeoutError

with ThemeParks() as tp:
    try:
        live = tp.entity("75ea578a-adc8-4116-a54d-dccb60765ef9").live()
    except RateLimitError as exc:
        # 429; exc.retry_after is seconds if the server told us
        print(f"rate limited, retry after {exc.retry_after}s")
    except APIError as exc:
        # any non-2xx status
        print(f"api error {exc.status} at {exc.url}: {exc.body}")
    except (NetworkError, TimeoutError) as exc:
        # transport failure or slow server
        print(f"transport: {exc!r}")

RateLimitError is a subclass of APIError, so the order of the except blocks matters if you want to handle 429 specially.

Debugging — see every HTTP request

The SDK is built on httpx, which has a built-in logger. Turn it on to see every outbound request and response status:

import logging
logging.basicConfig(level=logging.INFO)
logging.getLogger("httpx").setLevel(logging.DEBUG)

from themeparks import ThemeParks
with ThemeParks() as tp:
    tp.entity("75ea578a-adc8-4116-a54d-dccb60765ef9").live()

Output:

INFO httpx HTTP Request: GET https://api.themeparks.wiki/v1/entity/75ea578a-adc8-4116-a54d-dccb60765ef9/live "HTTP/1.1 200 OK"

For raw byte-level traces (TLS handshake, header bytes, etc.), also enable the httpcore logger:

logging.getLogger("httpcore").setLevel(logging.DEBUG)

Note: requests served from the in-memory cache do not appear in httpx logs — they're returned before the transport is touched. To see every call as a network round-trip while debugging, pass cache=False.

Caching

The default client caches GET responses in-memory with sensible per-endpoint TTLs:

Endpoint	TTL	Rationale
`GET /destinations`	1 hour	Directory rarely changes.
`GET /entity/{id}`	1 hour	Entity metadata is static.
`GET /entity/{id}/children`	1 hour	Park topology is stable.
`GET /entity/{id}/schedule[/yyyy/mm]`	5 minutes	Schedules update but not rapidly.
`GET /entity/{id}/live`	0 (bypass)	Live data is always fetched.

Disable caching

tp = ThemeParks(cache=False)

Plug in your own adapter

Cache is a Protocol; any object implementing get, set, and delete works. Here is a minimal dict-backed example (for real-world use you would want TTL enforcement and bounded size):

from typing import Any
from themeparks import ThemeParks, Cache

class DictCache:
    def __init__(self) -> None:
        self._data: dict[str, Any] = {}

    def get(self, key: str) -> Any | None:
        return self._data.get(key)

    def set(self, key: str, value: Any, ttl_seconds: float) -> None:
        self._data[key] = value

    def delete(self, key: str) -> None:
        self._data.pop(key, None)

tp = ThemeParks(cache=DictCache())

The per-endpoint TTL table is applied by the transport layer, so your adapter receives the correct ttl_seconds for each call and can honor it however it likes (Redis EXPIRE, filesystem mtime, etc.).

What's new in v2

v2 is a full rewrite on httpx + pydantic v2. It replaces the generated openapi_client surface with a hand-crafted client, fixes the nullable queue-field crash from issues #1 and #2, and adds native async support.

See MIGRATION.md for a side-by-side v1 to v2 guide.

Supported Python versions

3.9, 3.10, 3.11, 3.12, 3.13.

Links

SDK documentation: https://themeparks.github.io/ThemeParks_Python/
API reference: https://themeparks.github.io/ThemeParks_Python/api/client/
Cookbook: https://themeparks.github.io/ThemeParks_Python/cookbook/
Underlying API: https://api.themeparks.wiki
Issues: https://github.com/ThemeParks/ThemeParks_Python/issues
Changelog: CHANGELOG.md

License

MIT.

Name		Name	Last commit message	Last commit date
.github		.github
docs		docs
examples		examples
scripts		scripts
tests		tests
themeparks		themeparks
.gitignore		.gitignore
CHANGELOG.md		CHANGELOG.md
LICENSE		LICENSE
MIGRATION.md		MIGRATION.md
README.md		README.md
mkdocs.yml		mkdocs.yml
pyproject.toml		pyproject.toml
uv.lock		uv.lock

Navigation Menu

Repository files navigation

themeparks

Install

Print live wait times (sync)

Print live wait times (async)

Just the open rides, sorted longest-wait first

Client options

Ergonomic helpers

Reading every queue type

Direct access

Generic iteration

Other helpers

Rate limits

History

Or skip the code: there is a command

Low-level escape hatch

Error handling

Debugging — see every HTTP request

Caching

Disable caching

Plug in your own adapter

What's new in v2

Supported Python versions

Links

License

About

Releases

Packages

Used by

Contributors

Languages

Latest commit

History

Folders and files

Repository files navigation

themeparks

Install

Print live wait times (sync)

Print live wait times (async)

Just the open rides, sorted longest-wait first

Client options

Ergonomic helpers

Reading every queue type

Direct access

Generic iteration

Other helpers

Rate limits

History

Or skip the code: there is a command

Low-level escape hatch

Error handling

Debugging — see every HTTP request

Caching

Disable caching

Plug in your own adapter

What's new in v2

Supported Python versions

Links

License

About

Resources

Contributing

Stars

Watchers

Forks

Releases

Packages

Used by

Contributors

Languages