plexapi-patterns

name	plexapi-patterns
description	Concrete, copyable python-plexapi 4.18.2 patterns for Plex Watchlist Maintainer — reading and writing the watchlist, matching by GUID, the unwatched verdict, except ordering, and retries with backoff. Every claim verified against the installed source, with file:line.
user-invocable	false

python-plexapi 4.18.2 patterns

Every claim here is verified against the installed source and cites file:line. The original project specification contained wrong claims about this API more than once, so verify before asserting — that is what the plexapi-verifier agent is for.

The watchlist is this program's output, not a read-only input. There are no playlist patterns here any more: no addItems, no removeItems, no smart-playlist check, no episode selection. See ADR-010.

Reading the watchlist

items = account.watchlist()

Pass neither libtype nor maxresults.

• watchlist only sends a type parameter when libtype is truthy (myplex.py:961-962), so omitting it returns Movies and Shows (myplex.py:925). Asking for libtype="show" is what made every film on the watchlist invisible to this program — see ADR-013.
• maxresults is the only way to truncate the list by accident (myplex.py → base.py:341-342). Pagination is internal and transparent, 100 items per request, and it continues until the results are exhausted.

Each item carries .type — "movie" or "show" — read straight from the payload (video.py:61).

Writing the watchlist

for item in items:
    try:
        with_retries(lambda i=item: account.addToWatchlist(i), f"adding {item.title}")
    except BadRequest as exc:
        log.debug("%s was already on the watchlist: %s", item.title, exc)

• One item per call. There is no batch form worth using.
• BadRequest on add means it is already there; on remove, that it is already gone. Both mean the desired state already holds, so catch and log at DEBUG rather than treating it as failure.
• Note lambda i=item: — binding the loop variable as a default argument. Without it every retry closure captures the last item.
• Because both operations are idempotent, wrapping them in retries is safe. This is the one place writes may be retried.

Matching a watchlist item to the library

Watchlist items live in Plex's Discover service and carry a global identity (plex://show/..., plex://movie/...) unrelated to the server-local ratingKey. The GUID is the only stable key.

results = with_retries(
    lambda: server.library.search(guid=item.guid, libtype=item.type),
    f"guid lookup {item.guid}",
)
candidates = [r for r in results if getattr(r, "librarySectionID", None) in section_ids[item.type]]

This is the pattern plexapi's own documentation gives for a watchlist item (myplex.py:949).

The section filter is not optional, and it is per type. Library.search walks /library/all, which is global to the server (library.py:162-179), so without it an item that exists in several libraries can come back from one you did not configure. Sharing one set of ids across both types lets a movie be satisfied by a hit in a TV section.

Never match by title. Titles collide, have regional variants, and get reordered.

NotFound from the search means no results, not an error — catch it and treat it as an empty list. A watchlist item that is not on the server is normal and expected.

Duplicates

The same GUID in several configured sections is normal on servers keeping separate HD, 4K and Remux libraries. Pick the copy with the most episodes, and only warn when the picked copy has less progress than another, because only then can the choice change the verdict.

A Movie has neither leafCount nor viewedLeafCount, so read both through helpers that fall back to 1 and to viewCount. Reading candidate.leafCount or 0 directly raises on a movie.

The verdict: is anything unwatched?

if found.type == "movie":
    return not found.isPlayed

leaf = getattr(found, "leafCount", None) or 0
viewed = getattr(found, "viewedLeafCount", None) or 0

if include_specials:
    return viewed < leaf
if viewed >= leaf:
    return False

unwatched = with_retries(lambda: found.unwatched(), f"unwatched episodes of {found.title}")
return any(getattr(ep, "seasonNumber", None) != 0 for ep in unwatched)

• A movie is one thing: isPlayed is viewCount > 0 (mixins/played_unplayed.py:5-7), already on the search result, so it costs no extra request. It also means marking a film unplayed brings it back — a movie is not permanently finished. INCLUDE_SPECIALS never applies to one.
• leafCount / viewedLeafCount are counters the server keeps on the show itself, so this costs no extra request either. Walking the episodes would cost one per show.
• With specials excluded the counters cannot be trusted: they include season 0, so a show whose only unwatched episode is a special looks unfinished. Only then are the episodes read.
• Use ep.seasonNumber, never raw ep.parentIndex — parentIndex is None when seasons are hidden on the show (video.py:1178-1185), and comparing None with int raises TypeError.
• Neither watched() nor unwatched() sorts: both delegate to episodes() over /allLeaves with no sort (video.py:739-741). Sort explicitly if order matters.
• isPlayed is the canonical attribute name. isWatched exists as an alias (mixins/played_unplayed.py:23-26); preferring isPlayed is stylistic, not a compatibility issue.

Connecting

account = with_retries(lambda: MyPlexAccount(token=cfg.token), "authentication")
resource = with_retries(lambda: account.resource(cfg.server_name), f"resolving {cfg.server_name}")

if cfg.baseurl:
    server = PlexServer(cfg.baseurl, resource.accessToken or cfg.token)
else:
    server = resource.connect()

The server's access token is always resolved through account.resource(name), even when a base URL is configured. PLEX_BASEURL overrides the URL used to reach the server, never the identity used to talk to it — which is why the server name stays required. NotFound from resource() is a configuration error, not a transient one.

Except ordering

try:
    ...
except Unauthorized:  # MUST come first
    raise
except NotFound:  # needs its own clause
    ...
except (requests.RequestException, BadRequest) as exc:
    ...

• Unauthorized is a subclass of BadRequest (exceptions.py:26). Put BadRequest first and every authentication error is swallowed, which destroys the whole error policy.
• NotFound does not descend from BadRequest and needs its own clause.
• TwoFactorRequired descends from Unauthorized, which is what you want.

Retries with backoff

RETRY_BACKOFF_SECONDS = (5, 15, 45)

Three attempts on every idempotent read: connecting, the watchlist, the GUID lookup, unwatched(). Unauthorized is re-raised immediately — retrying a bad token only hammers the API. After the last attempt, raise a TransientError carrying a greppable reason slug.

Telling authentication from transient

It is not verified that Discover answers 401 for an invalid token. If it answers 400 or an empty body, a bad token looks transient forever: never notified, never backed off. Two mitigations, both needed:

try:
    items = with_retries(lambda: account.watchlist(), ...)
except TransientError:
    revalidate_account(account)  # account.ping(), raises Unauthorized if the token is dead
    raise

and, at the loop level, three consecutive passes that cannot read the watchlist while the server is reachable are reclassified as suspected authentication.

The watch check between passes

server.history(maxresults=1)

A single request, measured at 76-95 ms, against the 289 requests and ~41 s a full pass costs on a 177-show state. Deliberately not wrapped in retries and never fatal: it is an optimisation, and if it fails the loop falls back to waiting out INTERVAL_MINUTES. See ADR-009.

Nothing survives an iteration

No caching of MyPlexAccount, PlexServer or Show objects between passes. viewCount, viewedLeafCount and isPlayed are captured at fetch time, so a carried-over object makes the script blind to what the user watched since. Only the HTTP session and the auth-notification flag survive.