audora-provider-base (0.1.1)
Installation
pip install --index-url About this package
Async abstract base class and shared machinery for online publishing platform API clients
audora-provider-base
An async abstract base class and the shared machinery for building Python API clients against online publishing platforms.
These platforms all expose "stories with chapters, written by users, read and commented on
by other users" — but they disagree about almost everything else: how many hosts they use,
how they paginate, whether they have search at all, and what a Story object even contains
from one endpoint to the next. audora-provider-base absorbs that variance so each platform client
is a thin mapping layer rather than a from-scratch HTTP client.
This package contains no platform-specific code. It ships the contract; the clients ship separately.
from audora_provider_base import Capability, Detail
async with SomeProvider(cache=RedisBackend(cache_name="sp", expire_after=43200)) as p:
story = await p.get_story("12345", detail=Detail.FULL) # exactly 1 request
# Chapters arrive as stubs. Fill them in with one bulk call where the
# platform supports it; otherwise this refuses rather than firing N requests.
await p.load_chapters(story, with_text=True)
print(story.title, len(story.chapters), p.transport.requests_issued)
if Capability.SEARCH_STORIES in p.capabilities:
async for hit in p.search_stories("space opera"):
...
What it gives you
| Transport | aiohttp + injectable aiohttp-client-cache CacheBackend, retries with backoff, per-provider static headers, ETag/304 handling, request counters |
| Models | Pydantic v2 canonical entities with an UNSET sentinel that distinguishes "not fetched" from "no value", and a .raw escape hatch for everything else |
| Pagination | Five strategies (next-URL, offset, page-number, cursor, after-token) behind one Paginator, with explicit termination |
| Capabilities | Feature flags so a platform without search says so, instead of returning nothing |
| Cost control | Every method documents its request count; data-dependent fan-out must be opted into |
| Bulk & collections | Library, archive, and reading lists as one Collection type, with BulkResult for partial failure |
| Contract suite | A reusable pytest suite your subclass must pass, so the ABC is a real contract |
Design notes
UNSET is not None. Platforms return different field sets from different endpoints for
the same entity. story.cover is UNSET means you fetched a projection that omits covers;
story.cover is None means the platform says this story has no cover. Accessing UNSET
never triggers a request.
Request cost is part of the method signature. Nothing makes a data-dependent number of calls behind your back:
await p.get_story(sid) # exactly 1
await p.load_chapters(story) # 1 where a bulk endpoint exists
await p.get_stories(ids) # 1 if the platform batches...
await p.get_stories(ids, allow_fanout=True) # ...else N, and you had to say so
Bulk writes report partial failure instead of raising, because platforms genuinely return per-item outcomes:
result = await p.add_stories(collection, story_ids)
if not result:
for failure in result.failed:
print(failure.item, failure.reason) # e.g. "inArchive"
Caching policy is yours. The CacheBackend you pass is handed straight to
CachedSession. The base does not decide what is safe to cache — notably, it will not stop
you sharing one cache across two authenticated accounts. See transport.py.
Implementing a provider
Subclass BaseProvider, declare slug and capabilities, implement the abstract methods,
and run the contract suite against it:
import pytest_asyncio
from audora_provider_base.testing import ProviderContractTests
class TestMyProvider(ProviderContractTests):
story_ref = "12345"
missing_story_ref = "0"
user_ref = "some-author"
@pytest_asyncio.fixture
async def provider(self):
async with MyProvider(base_url=fixture_server_url) as p:
yield p
Your package needs asyncio_mode = "auto" under [tool.pytest.ini_options] — every
test in the suite is an async def.
audora_provider_base.testing.FakeProvider is a complete reference implementation over a local
fixture server — read it as the worked example, alongside
docs/guides/implementing-a-provider.md.
Development
This project formats and lints with ruff, and nothing else. Run it after every
change, across src/, tests/, and examples/.
uv pip install -e ".[dev]"
ruff format . && ruff check --fix . && mypy src/ tests/ examples/ && pytest
See docs/contributing.md for the full gate, including the documentation drift tests.
Set AUDORA_PROVIDER_BASE_STRICT=1 to make unknown upstream fields a hard validation error. Do this
in CI so API drift fails a test; never do it in production, where degrading is better than
crashing.
It only bites on payloads passed through Model.model_validate(). A provider that maps
keys by hand never looks at the unrecognised ones, so nothing raises — validate through the
model if you want strict mode to guard your mapping.
Releases
GitHub Actions builds and validates the wheel and source distribution on every push. A
version tag publishes them to the aaron Forgejo PyPI registry. Create a Forgejo access
token with write:package permission, save it as the GitHub repository Actions secret
FORGEJO_PACKAGE_TOKEN, then release the version declared in pyproject.toml:
git tag v0.1.0
git push origin v0.1.0
The tag must exactly match the package version with a leading v; the workflow rejects a
mismatch rather than publishing a mislabeled build.
Packages derived from this one ship the same way. A provider package copies
.github/workflows/package.yaml verbatim — it names nothing specific to this repo, and
reads the version from pyproject.toml — sets its own FORGEJO_PACKAGE_TOKEN secret, and
releases by pushing a matching v tag to the same registry. See
implementing a provider.