Clients¶

Wallhaven¶

class xanax.sources.wallhaven.client.Wallhaven(api_key=None, timeout=30.0, max_retries=0)[source]¶

Bases: object

Synchronous client for the Wallhaven API v1.

Satisfies MediaSource: download and iter_media work identically across all xanax sources.

The API key can be passed directly or read from the WALLHAVEN_API_KEY environment variable. An API key is only required for NSFW content, account settings, and private collections.

Example

client = Wallhaven(api_key=”your-api-key”)

params = SearchParams(query=”anime”, purity=[Purity.SFW]) results = client.search(params)

for wallpaper in results.data:: print(wallpaper.resolution, wallpaper.path)

Parameters:

api_key (str | None) – Wallhaven API key. Falls back to WALLHAVEN_API_KEY env var.
timeout (float) – Request timeout in seconds. Default is 30.
max_retries (int) – Maximum retries on rate limiting (429). Default is 0 (fail-fast). Set to 3 for exponential backoff.

BASE_URL = 'https://wallhaven.cc/api/v1'¶

__init__(api_key=None, timeout=30.0, max_retries=0)[source]¶

property is_authenticated: bool¶: Return True if the client has an API key configured.

wallpaper(wallpaper_id)[source]¶

Get full metadata for a specific wallpaper.

Parameters:: wallpaper_id (str) – The wallpaper ID (e.g., "94x38z").
Return type:: Wallpaper
Returns:: Wallpaper with full details.
Raises:: NotFoundError – If the wallpaper does not exist.

search(params)[source]¶

Search for wallpapers.

Parameters:: params (SearchParams) – SearchParams with search criteria.
Return type:: SearchResult
Returns:: SearchResult with wallpapers and metadata.
Raises:: AuthenticationError – If NSFW is requested without an API key.

tag(tag_id)[source]¶

Get information about a specific tag.

Parameters:: tag_id (int) – The tag ID.
Return type:: Tag
Returns:: Tag.
Raises:: NotFoundError – If the tag does not exist.

settings()[source]¶

Get the authenticated user’s settings. Requires an API key.

Return type:: UserSettings
Returns:: UserSettings.
Raises:: AuthenticationError – If no API key is configured.

collections(username=None)[source]¶

Get collections for the authenticated user or a given username.

Parameters:: username (str | None) – Optional username. If omitted, returns the authenticated user’s own collections (requires an API key).
Return type:: list[Collection]
Returns:: List of Collection.
Raises:: AuthenticationError – If accessing own collections without an API key.

collection(username, collection_id)[source]¶

Get wallpapers in a specific collection.

Parameters:

username (str) – Username who owns the collection.
collection_id (int) – Collection ID.

Return type:

CollectionListing

Returns:

CollectionListing.

Raises:

NotFoundError – If the collection does not exist.

download(wallpaper, path=None)[source]¶

Download the full-resolution image bytes for a wallpaper.

Parameters:

wallpaper (Wallpaper) – The Wallpaper to download.
path (Path | str | None) – Optional path to write the image to disk. Bytes are also returned.

Return type:

bytes

Returns:

Raw image bytes.

iter_pages(params)[source]¶

Iterate over all pages of search results automatically.

Pagination is handled transparently, including carrying forward any seed returned by the API for random-sorted results.

Parameters:: params (SearchParams) – Starting SearchParams.
Yields:: SearchResult per page.

iter_media(params)[source]¶

Iterate over every wallpaper across all pages of search results.

A convenience wrapper around iter_pages() that flattens pages into individual Wallpaper objects.

Parameters:: params (SearchParams) – Starting SearchParams.
Yields:: Wallpaper objects.

close()[source]¶

Close the underlying HTTP client.

Return type:: None

__enter__()[source]¶

Return type:: Wallhaven

__exit__(exc_type, exc_val, exc_tb)[source]¶

Return type:: None

__repr__()[source]¶

Return repr(self).

Return type:: str

class xanax.sources.wallhaven.async_client.AsyncWallhaven(api_key=None, timeout=30.0, max_retries=0)[source]¶

Bases: object

Asynchronous client for the Wallhaven API v1.

Satisfies AsyncMediaSource: download and aiter_media work identically across all xanax async sources.

Example

async with AsyncWallhaven(api_key=”your-api-key”) as client:

results = await client.search(SearchParams(query=”anime”))

async for wallpaper in client.aiter_media(SearchParams(query=”nature”)):: print(wallpaper.path)

Parameters:

api_key (str | None) – Wallhaven API key. Falls back to WALLHAVEN_API_KEY env var.
timeout (float) – Request timeout in seconds. Default is 30.
max_retries (int) – Maximum retries on rate limiting (429). Default is 0 (fail-fast). Set to 3 for exponential backoff.

BASE_URL = 'https://wallhaven.cc/api/v1'¶

__init__(api_key=None, timeout=30.0, max_retries=0)[source]¶

property is_authenticated: bool¶: Return True if the client has an API key configured.

async wallpaper(wallpaper_id)[source]¶

Get full metadata for a specific wallpaper.

Parameters:: wallpaper_id (str) – The wallpaper ID (e.g., "94x38z").
Return type:: Wallpaper
Returns:: Wallpaper with full details.
Raises:: NotFoundError – If the wallpaper does not exist.

async search(params)[source]¶

Search for wallpapers.

Parameters:: params (SearchParams) – SearchParams with search criteria.
Return type:: SearchResult
Returns:: SearchResult.
Raises:: AuthenticationError – If NSFW is requested without an API key.

async tag(tag_id)[source]¶

Get information about a specific tag.

Parameters:: tag_id (int) – The tag ID.
Return type:: Tag
Returns:: Tag.
Raises:: NotFoundError – If the tag does not exist.

async settings()[source]¶

Get the authenticated user’s settings. Requires an API key.

Return type:: UserSettings
Returns:: UserSettings.
Raises:: AuthenticationError – If no API key is configured.

async collections(username=None)[source]¶

Get collections for the authenticated user or a given username.

Parameters:: username (str | None) – Optional username. If omitted, returns the authenticated user’s own collections (requires an API key).
Return type:: list[Collection]
Returns:: List of Collection.
Raises:: AuthenticationError – If accessing own collections without an API key.

async collection(username, collection_id)[source]¶

Get wallpapers in a specific collection.

Parameters:

username (str) – Username who owns the collection.
collection_id (int) – Collection ID.

Return type:

CollectionListing

Returns:

CollectionListing.

Raises:

NotFoundError – If the collection does not exist.

async download(wallpaper, path=None)[source]¶

Download the full-resolution image bytes for a wallpaper.

Parameters:

wallpaper (Wallpaper) – The Wallpaper to download.
path (Path | str | None) – Optional path to write the image to disk. Bytes are also returned.

Return type:

bytes

Returns:

Raw image bytes.

async aiter_pages(params)[source]¶

Async-iterate over all pages of search results.

Parameters:: params (SearchParams) – Starting SearchParams.
Yields:: SearchResult per page.

async aiter_media(params)[source]¶

Async-iterate over every wallpaper across all pages.

A convenience wrapper around aiter_pages() that flattens pages into individual Wallpaper objects.

Parameters:: params (SearchParams) – Starting SearchParams.
Yields:: Wallpaper objects.

async aclose()[source]¶

Close the underlying async HTTP client.

Return type:: None

async __aenter__()[source]¶

Return type:: AsyncWallhaven

async __aexit__(exc_type, exc_val, exc_tb)[source]¶

Return type:: None

__repr__()[source]¶

Return repr(self).

Return type:: str

Unsplash¶

class xanax.sources.unsplash.client.Unsplash(access_key=None, timeout=30.0, max_retries=0)[source]¶

Bases: object

Synchronous client for the Unsplash API.

Authenticates via an access key (Authorization: Client-ID <key>). The key can be passed directly or read from the UNSPLASH_ACCESS_KEY environment variable.

Satisfies MediaSource: download and iter_media work identically to the Wallhaven client, making it straightforward to write source-agnostic code.

Example

unsplash = Unsplash(access_key="your-access-key")

result = unsplash.search(UnsplashSearchParams(query="mountains"))
for photo in result.results:
    print(photo.id, photo.resolution)

photo = unsplash.random()
data = unsplash.download(photo)

Parameters:

access_key (str | None) – Unsplash API access key. Falls back to the UNSPLASH_ACCESS_KEY environment variable.
timeout (float) – Request timeout in seconds. Default is 30.
max_retries (int) – Maximum retries on rate limiting (429). Default is 0 (fail-fast). Set to 3 for exponential backoff.

Raises:

AuthenticationError – If no access key is provided or discoverable.

BASE_URL = 'https://api.unsplash.com'¶

__init__(access_key=None, timeout=30.0, max_retries=0)[source]¶

search(params)[source]¶

Search for photos matching the given parameters.

Parameters:: params (UnsplashSearchParams) – UnsplashSearchParams with query and optional filters.
Return type:: UnsplashSearchResult
Returns:: UnsplashSearchResult with total, total_pages, and results.
Raises:: AuthenticationError – If the access key is invalid.

photo(photo_id)[source]¶

Retrieve a full photo object by ID.

Unlike search results, the returned photo includes exif, location, tags, downloads, and public_domain.

Parameters:: photo_id (str) – Unsplash photo ID (e.g. "Dwu85P9SOIk").
Return type:: UnsplashPhoto
Returns:: Full UnsplashPhoto.
Raises:: NotFoundError – If the photo does not exist.

random(params=None)[source]¶

Retrieve a single random photo.

Without parameters, a completely random photo is returned. Parameters narrow the eligible pool (by collection, topic, user, query, or orientation).

Parameters:: params (UnsplashRandomParams | None) – Optional UnsplashRandomParams to constrain the random selection.
Return type:: UnsplashPhoto
Returns:: A UnsplashPhoto.
Raises:: AuthenticationError – If the access key is invalid.

download(photo, path=None)[source]¶

Download the raw image bytes for a photo.

Unsplash’s API Terms of Service require triggering a tracking request before downloading. This method performs both steps automatically:

GET photo.links.download_location (triggers attribution tracking).
GET the CDN URL returned from step 1 (fetches actual image bytes).

Parameters:

photo (UnsplashPhoto) – The UnsplashPhoto to download.
path (Path | str | None) – Optional file path to save the image. If provided, the bytes are written to this path in addition to being returned.

Return type:

bytes

Returns:

Raw image bytes.

Raises:

httpx.HTTPStatusError – If either request fails.

iter_pages(params)[source]¶

Iterate over all pages of search results automatically.

Each iteration yields a full UnsplashSearchResult page. Pagination is handled transparently.

Parameters:: params (UnsplashSearchParams) – Starting UnsplashSearchParams. The page field is managed automatically.
Yields:: UnsplashSearchResult for each page.

Example

for page in unsplash.iter_pages(UnsplashSearchParams(query=”nature”)):

for photo in page.results:: print(photo.id)

iter_media(params)[source]¶

Iterate over every photo across all pages of search results.

A convenience wrapper around iter_pages() that flattens pages into individual UnsplashPhoto objects.

Parameters:: params (UnsplashSearchParams) – Starting UnsplashSearchParams.
Yields:: UnsplashPhoto objects across all pages.

Example

for photo in unsplash.iter_media(UnsplashSearchParams(query=”forest”)):: data = unsplash.download(photo)

close()[source]¶

Close the underlying HTTP client.

Return type:: None

__enter__()[source]¶

Return type:: Unsplash

__exit__(exc_type, exc_val, exc_tb)[source]¶

Return type:: None

__repr__()[source]¶

Return repr(self).

Return type:: str

class xanax.sources.unsplash.async_client.AsyncUnsplash(access_key=None, timeout=30.0, max_retries=0)[source]¶

Bases: object

Asynchronous client for the Unsplash API.

Drop-in async counterpart to Unsplash. All public methods are coroutines. Use as an async context manager for automatic resource cleanup.

The access key can be passed directly or read from the UNSPLASH_ACCESS_KEY environment variable.

Example

async with AsyncUnsplash(access_key=”your-access-key”) as unsplash:

result = await unsplash.search(UnsplashSearchParams(query=”mountains”))

async for photo in unsplash.aiter_media(UnsplashSearchParams(query=”forest”)):: data = await unsplash.download(photo)

Parameters:

access_key (str | None) – Unsplash API access key. Falls back to the UNSPLASH_ACCESS_KEY environment variable.
timeout (float) – Request timeout in seconds. Default is 30.
max_retries (int) – Maximum retries on rate limiting (429). Default is 0 (fail-fast). Set to 3 for exponential backoff.

Raises:

AuthenticationError – If no access key is provided or discoverable.

BASE_URL = 'https://api.unsplash.com'¶

__init__(access_key=None, timeout=30.0, max_retries=0)[source]¶

async search(params)[source]¶

Search for photos matching the given parameters.

Parameters:: params (UnsplashSearchParams) – UnsplashSearchParams with query and optional filters.
Return type:: UnsplashSearchResult
Returns:: UnsplashSearchResult with total, total_pages, and results.
Raises:: AuthenticationError – If the access key is invalid.

async photo(photo_id)[source]¶

Retrieve a full photo object by ID.

Unlike search results, the returned photo includes exif, location, tags, downloads, and public_domain.

Parameters:: photo_id (str) – Unsplash photo ID (e.g. "Dwu85P9SOIk").
Return type:: UnsplashPhoto
Returns:: Full UnsplashPhoto.
Raises:: NotFoundError – If the photo does not exist.

async random(params=None)[source]¶

Retrieve a single random photo.

Without parameters, a completely random photo is returned. Parameters narrow the eligible pool (by collection, topic, user, query, or orientation).

Parameters:: params (UnsplashRandomParams | None) – Optional UnsplashRandomParams to constrain the random selection.
Return type:: UnsplashPhoto
Returns:: A UnsplashPhoto.
Raises:: AuthenticationError – If the access key is invalid.

async download(photo, path=None)[source]¶

Download the raw image bytes for a photo.

Unsplash’s API Terms of Service require triggering a tracking request before downloading. This method performs both steps automatically:

GET photo.links.download_location (triggers attribution tracking).
GET the CDN URL returned from step 1 (fetches actual image bytes).

Parameters:

photo (UnsplashPhoto) – The UnsplashPhoto to download.
path (Path | str | None) – Optional file path to save the image. If provided, the bytes are written to this path in addition to being returned.

Return type:

bytes

Returns:

Raw image bytes.

Raises:

httpx.HTTPStatusError – If either request fails.

async aiter_pages(params)[source]¶

Async-iterate over all pages of search results automatically.

Each iteration yields a full UnsplashSearchResult page. Pagination is handled transparently.

Parameters:: params (UnsplashSearchParams) – Starting UnsplashSearchParams. The page field is managed automatically.
Yields:: UnsplashSearchResult for each page.

Example

async for page in unsplash.aiter_pages(UnsplashSearchParams(query=”nature”)):

for photo in page.results:: print(photo.id)

async aiter_media(params)[source]¶

Async-iterate over every photo across all pages of search results.

A convenience wrapper around aiter_pages() that flattens pages into individual UnsplashPhoto objects.

Parameters:: params (UnsplashSearchParams) – Starting UnsplashSearchParams.
Yields:: UnsplashPhoto objects across all pages.

Example

async for photo in unsplash.aiter_media(UnsplashSearchParams(query=”forest”)):: data = await unsplash.download(photo)

async aclose()[source]¶

Close the underlying async HTTP client.

Return type:: None

async __aenter__()[source]¶

Return type:: AsyncUnsplash

async __aexit__(exc_type, exc_val, exc_tb)[source]¶

Return type:: None

__repr__()[source]¶

Return repr(self).

Return type:: str

Reddit¶

class xanax.sources.reddit.client.Reddit(client_id=None, client_secret=None, user_agent=None, timeout=30.0, max_retries=0)[source]¶

Bases: object

Synchronous Reddit client.

Fetches media posts from subreddit listings and satisfies the MediaSource protocol.

Authentication uses OAuth2 app-only credentials (client_id + client_secret). No user login is required to read public subreddits. Credentials can be passed explicitly or read from environment variables REDDIT_CLIENT_ID, REDDIT_CLIENT_SECRET, and REDDIT_USER_AGENT.

Reddit requires a descriptive User-Agent on every request. The recommended format is: platform:app_id/version (by u/username)

Example

reddit = Reddit(
    client_id="...",
    client_secret="...",
    user_agent="python:xanax/0.3.0 (by u/yourname)",
)

for post in reddit.iter_media(RedditParams(subreddit="EarthPorn", sort=RedditSort.TOP)):
    reddit.download(post, path=f"{post.id}.jpg")

Parameters:

client_id (str | None) – Reddit app client ID. Falls back to REDDIT_CLIENT_ID.
client_secret (str | None) – Reddit app client secret. Falls back to REDDIT_CLIENT_SECRET.
user_agent (str | None) – Required User-Agent string. Falls back to REDDIT_USER_AGENT.
timeout (float) – Request timeout in seconds. Default is 30.
max_retries (int) – Maximum retries on 429 rate-limit responses. Default is 0 (fail-fast). Set to a positive integer to enable exponential backoff.

Raises:

AuthenticationError – If any of client_id, client_secret, or user_agent cannot be resolved.

BASE_URL = 'https://oauth.reddit.com'¶

__init__(client_id=None, client_secret=None, user_agent=None, timeout=30.0, max_retries=0)[source]¶

listing(params)[source]¶

Fetch one page of posts from a subreddit listing.

Posts are filtered through from_reddit_data(): text posts and unsupported link types are silently excluded.

Parameters:

params (RedditParams) – RedditParams with subreddit, sort, limit, and optional cursor.

Return type:

RedditListing

Returns:

RedditListing containing parsed posts, the after cursor, and the raw dist count.

Raises:

AuthenticationError – If credentials are invalid.
NotFoundError – If the subreddit does not exist.
RateLimitError – If the rate limit is exceeded and retries are exhausted.
APIError – For any other non-success HTTP status.

post(post_id)[source]¶

Fetch a single post by its base-36 ID.

Returns None if the post exists but is not a supported media type (e.g. a text post or external link).

Parameters:

post_id (str) – Base-36 Reddit post ID (e.g. "abc123").

Return type:

RedditPost | None

Returns:

Parsed RedditPost, or None if the post has no supported media.

Raises:

NotFoundError – If the post does not exist.
AuthenticationError – If credentials are invalid.
APIError – For unexpected HTTP errors.

download(post, path=None)[source]¶

Download the raw media bytes for a post.

For IMAGE posts the direct url is fetched. For VIDEO and GIF posts the video_url (the video-only fallback_url from v.redd.it) is fetched instead.

Note

Reddit video does not include audio in the fallback_url stream. Audio is delivered as a separate DASH stream and requires ffmpeg to merge. Only video bytes are returned here.

Parameters:

post (RedditPost) – The RedditPost to download.
path (Path | str | None) – Optional file path to save the bytes. If provided, the bytes are also written to disk.

Return type:

bytes

Returns:

Raw media bytes.

Raises:

ValueError – If the post has no downloadable URL.
httpx.HTTPStatusError – If the download request fails.

iter_pages(params)[source]¶

Iterate through all pages of a subreddit listing using cursor pagination.

Each call to the API fetches the next page using the after cursor from the previous response. Iteration stops when there is no further cursor or when an empty page is returned.

Parameters:: params (RedditParams) – Starting RedditParams. The after field is managed automatically.
Yields:: RedditListing for each page.

Example

for page in reddit.iter_pages(RedditParams(subreddit=”wallpapers”)):

for post in page.posts:: print(post.id)

iter_media(params)[source]¶

Iterate over all media posts, flattening pages and expanding galleries.

Handles all pagination automatically. Gallery posts are expanded into individual RedditPost objects (one per image), each with gallery_index and gallery_id populated.

Posts are filtered according to:

params.media_type: skips posts whose media_type does not match (unless media_type=ANY).
params.include_nsfw: skips NSFW-flagged posts when False (the default).

Parameters:: params (RedditParams) – RedditParams with filter and pagination settings.
Yields:: RedditPost objects.

Example

for post in reddit.iter_media(: RedditParams(subreddit=”EarthPorn”, sort=RedditSort.TOP)
):: reddit.download(post, path=f”{post.id}.jpg”)

close()[source]¶

Close the underlying HTTP client.

Return type:: None

__enter__()[source]¶

Return type:: Reddit

__exit__(exc_type, exc_val, exc_tb)[source]¶

Return type:: None

__repr__()[source]¶

Return repr(self).

Return type:: str

class xanax.sources.reddit.async_client.AsyncReddit(client_id=None, client_secret=None, user_agent=None, timeout=30.0, max_retries=0)[source]¶

Bases: object

Asynchronous Reddit client.

Drop-in async counterpart to Reddit. All public methods are coroutines. Use as an async context manager for automatic resource cleanup.

Authentication uses OAuth2 app-only credentials (client_id + client_secret). No user login is required for public subreddits. Credentials can be passed explicitly or read from environment variables REDDIT_CLIENT_ID, REDDIT_CLIENT_SECRET, and REDDIT_USER_AGENT.

Example

async with AsyncReddit(

client_id=”…”, client_secret=”…”, user_agent=”python:xanax/0.3.0 (by u/yourname)”,

) as reddit:

async for post in reddit.aiter_media(: RedditParams(subreddit=”EarthPorn”, sort=RedditSort.TOP)
):: await reddit.download(post, path=f”{post.id}.jpg”)

Parameters:

client_id (str | None) – Reddit app client ID. Falls back to REDDIT_CLIENT_ID.
client_secret (str | None) – Reddit app client secret. Falls back to REDDIT_CLIENT_SECRET.
user_agent (str | None) – Required User-Agent string. Falls back to REDDIT_USER_AGENT.
timeout (float) – Request timeout in seconds. Default is 30.
max_retries (int) – Maximum retries on 429 rate-limit responses. Default is 0 (fail-fast).

Raises:

AuthenticationError – If any credential cannot be resolved.

BASE_URL = 'https://oauth.reddit.com'¶

__init__(client_id=None, client_secret=None, user_agent=None, timeout=30.0, max_retries=0)[source]¶

async listing(params)[source]¶

Fetch one page of posts from a subreddit listing.

Parameters:

params (RedditParams) – RedditParams with subreddit, sort, limit, and optional cursor.

Return type:

RedditListing

Returns:

RedditListing with parsed posts, pagination cursors, and the raw dist count.

Raises:

AuthenticationError – If credentials are invalid.
NotFoundError – If the subreddit does not exist.
RateLimitError – If the rate limit is exceeded.
APIError – For any other non-success HTTP status.

async post(post_id)[source]¶

Fetch a single post by its base-36 ID.

Returns None if the post exists but has no supported media.

Parameters:

post_id (str) – Base-36 Reddit post ID (e.g. "abc123").

Return type:

RedditPost | None

Returns:

Parsed RedditPost, or None if no media is present.

Raises:

NotFoundError – If the post does not exist.
AuthenticationError – If credentials are invalid.
APIError – For unexpected HTTP errors.

async download(post, path=None)[source]¶

Download the raw media bytes for a post.

For VIDEO and GIF posts the video_url is used (video-only stream, no audio). For IMAGE posts the direct url is fetched.

Note

Reddit video does not include audio in the fallback_url stream. Only video bytes are returned.

Parameters:

post (RedditPost) – The RedditPost to download.
path (Path | str | None) – Optional file path to save the bytes.

Return type:

bytes

Returns:

Raw media bytes.

Raises:

ValueError – If the post has no downloadable URL.
httpx.HTTPStatusError – If the download request fails.

async aiter_pages(params)[source]¶

Async-iterate through all pages of a subreddit listing.

Parameters:: params (RedditParams) – Starting RedditParams. The after cursor is managed automatically.
Yields:: RedditListing for each page.

Example

async for page in reddit.aiter_pages(RedditParams(subreddit=”wallpapers”)):

for post in page.posts:: print(post.id)

async aiter_media(params)[source]¶

Async-iterate over all media posts, flattening pages and expanding galleries.

Applies the same filtering as the sync iter_media():

Skips posts whose media_type does not match params.media_type (unless media_type=ANY).
Skips NSFW posts unless params.include_nsfw=True.
Expands gallery posts into individual RedditPost objects.

Parameters:: params (RedditParams) – RedditParams with filter and pagination settings.
Yields:: RedditPost objects.

Example

async for post in reddit.aiter_media(: RedditParams(subreddit=”EarthPorn”, sort=RedditSort.TOP)
):: await reddit.download(post, path=f”{post.id}.jpg”)

async aclose()[source]¶

Close the underlying async HTTP client.

Return type:: None

async __aenter__()[source]¶

Return type:: AsyncReddit

async __aexit__(exc_type, exc_val, exc_tb)[source]¶

Return type:: None

__repr__()[source]¶

Return repr(self).

Return type:: str