Async Client

AsyncWallhaven is a complete async counterpart to Wallhaven. Every method is async def, it uses httpx.AsyncClient internally, and it supports async context managers and async generators.

Basic usage

import asyncio
from xanax import AsyncWallhaven, SearchParams

async def main():
    client = AsyncWallhaven(api_key="your-api-key")

    results = await client.search(SearchParams(query="anime"))
    for wallpaper in results.data:
        print(wallpaper.resolution, wallpaper.path)

    await client.aclose()

asyncio.run(main())

Auto-pagination

async def main():
    async with AsyncWallhaven(api_key="your-api-key") as client:
        # Flat iteration over all wallpapers
        async for wallpaper in client.aiter_media(SearchParams(query="space")):
            print(wallpaper.id, wallpaper.path)

        # Page-by-page
        async for page in client.aiter_pages(SearchParams(query="forest")):
            print(f"Page {page.meta.current_page}: {len(page.data)} results")

Available methods

All sync Wallhaven methods have async equivalents:

Sync

Async

wallpaper(id)

await client.wallpaper(id)

search(params)

await client.search(params)

tag(id)

await client.tag(id)

settings()

await client.settings()

collections(username?)

await client.collections(username?)

collection(username, id)

await client.collection(username, id)

download(wallpaper, path?)

await client.download(wallpaper, path?)

iter_pages(params)

client.aiter_pages(params) (async generator)

iter_media(params)

client.aiter_media(params) (async generator)

Rate limit retry

Retry with exponential backoff uses asyncio.sleep(), so it never blocks the event loop:

client = AsyncWallhaven(api_key="your-api-key", max_retries=3)

Authentication and environment variables

AsyncWallhaven reads WALLHAVEN_API_KEY from the environment the same way Wallhaven does:

import os
os.environ["WALLHAVEN_API_KEY"] = "your-api-key"

async with AsyncWallhaven() as client:  # key picked up automatically
    ...

Concurrent requests

Because AsyncWallhaven is fully async, you can use asyncio.gather() or asyncio.TaskGroup to run requests concurrently:

import asyncio
from xanax import AsyncWallhaven

async def main():
    async with AsyncWallhaven(api_key="your-api-key") as client:
        wp1, wp2, wp3 = await asyncio.gather(
            client.wallpaper("94x38z"),
            client.wallpaper("3l3mpl"),
            client.wallpaper("kw7k78"),
        )
        print(wp1.resolution, wp2.resolution, wp3.resolution)

asyncio.run(main())

Integration with web frameworks

AsyncWallhaven works naturally with async web frameworks like FastAPI:

from contextlib import asynccontextmanager
from fastapi import FastAPI
from xanax import AsyncWallhaven, SearchParams

client: AsyncWallhaven

@asynccontextmanager
async def lifespan(app: FastAPI):
    global client
    client = AsyncWallhaven(api_key="your-api-key")
    yield
    await client.aclose()

app = FastAPI(lifespan=lifespan)

@app.get("/search")
async def search(q: str):
    results = await client.search(SearchParams(query=q))
    return {"total": results.meta.total, "wallpapers": [w.id for w in results.data]}