Skip to main content

Mokkari

PyPI - Version PyPI - Python Ruff Pre-Commit

A python wrapper for the Metron Comic Book Database API.

Installation

pip install mokkari

Authentication

Authenticate with an API token, which you can generate from your metron.cloud account page:

import mokkari

m = mokkari.api(api_token="your-api-token")

Username/password (Basic Auth) was removed in v5.0. If you're upgrading from 4.x, replace mokkari.api(username, password) with a token as shown above.

Example Usage

import mokkari

# Your own config file to keep your credentials secret
from config import api_token

m = mokkari.api(api_token=api_token)

# Get all Marvel comics for the week of 2021-06-07
this_week = m.issues_list(
    {
        "store_date_range_after": "2021-06-07",
        "store_date_range_before": "2021-06-13",
        "publisher_name": "marvel",
    }
)

# Print the results
for i in this_week:
    print(f"{i.id} {i.issue_name}")

    # Retrieve the detail for an individual issue
    asm_68 = m.issue(31660)

# Print the issue Description
print(asm_68.desc)

Rate Limiting

The API allows at least 20 requests per minute (the server may allow more when load is low), plus a daily limit that starts at 5,000 requests and is raised for OpenCollective donors (up to 25,000/day). Because the daily limit varies per user, mokkari doesn't hardcode it — it reads the X-RateLimit-* headers Metron returns with every response and pre-empts a request once those headers show a window is exhausted, avoiding an HTTP call that would fail anyway. When a rate limit is exceeded, a RateLimitError is raised.

The most recently observed state is available via session.rate_limit_status:

status = m.rate_limit_status
print(f"Sustained remaining: {status.sustained.remaining}/{status.sustained.limit}")

Handling Rate Limits

The RateLimitError includes a retry_after attribute that tells you exactly how many seconds to wait before making another request:

import mokkari
from mokkari.exceptions import RateLimitError
import time

m = mokkari.api(api_token=api_token)

try:
    issue = m.issue(31660)
except RateLimitError as e:
    # Display user-friendly message
    print(f"Rate limited: {e}")

    # Programmatically wait for the exact time needed
    print(f"Waiting {e.retry_after} seconds...")
    time.sleep(e.retry_after)

    # Retry the request
    issue = m.issue(31660)

Thread Safety

A Session can be shared across threads, but the rate-limit check above is advisory rather than a hard gate: it only blocks once the last known response headers show a window is exhausted, and that check isn't synchronized with sending the request. Concurrent threads can therefore each pass the check and send their requests before either response updates rate_limit_status, letting a burst of threads momentarily exceed the per-minute limit (Metron's server-side limit still applies and will reject the excess requests).

If you're calling a shared Session from multiple threads, cap your own concurrency instead of relying on Session to do it for you, e.g. keep a ThreadPoolExecutor at or below the burst limit:

from concurrent.futures import ThreadPoolExecutor

m = mokkari.api(api_token=api_token)

# Keep worker count at or below the burst limit (20/min at minimum) to avoid
# racing past the local rate-limit check.
with ThreadPoolExecutor(max_workers=20) as executor:
    issues = list(executor.map(m.issue, issue_ids))

Pacing (opt-in)

Passing rate_limiter closes the gap described above: instead of racing past an advisory check, every HTTP send is dispatched through the rate limiter first, which can block a caller until capacity actually frees rather than letting it send anyway. mokkari.rate_limit.HeaderPacedRateLimiter is a ready-to-use implementation. It sizes the per-minute window from the X-RateLimit-* headers but paces it from its own monotonic log of send times, so a local clock that has drifted from Metron's doesn't matter, and it spaces sends evenly across the window (at the 20/min floor, one every 3 seconds). If Metron still answers with a 429, it backs every caller off by the Retry-After value the server sent:

from concurrent.futures import ThreadPoolExecutor

import mokkari
from mokkari.rate_limit import HeaderPacedRateLimiter

m = mokkari.api(api_token=api_token, rate_limiter=HeaderPacedRateLimiter())

with ThreadPoolExecutor(max_workers=20) as executor:
    issues = list(executor.map(m.issue, issue_ids))

The limiter only blocks for the per-minute window, whose waits are seconds long. When the daily limit is exhausted it raises RateLimitError instead of blocking for what could be hours, with retry_after set to the time until the daily window resets. That leaves it to your application to tell the user and either wait or quit:

import time

from mokkari.exceptions import RateLimitError
from mokkari.utils import format_time

try:
    issue = m.issue(31660)
except RateLimitError as e:
    if input(f"Daily limit reached. Wait {format_time(e.retry_after)}? (y/n): ") == "y":
        time.sleep(e.retry_after)
        issue = m.issue(31660)

Paginated list calls follow the same rules: if a page is rejected with a 429 they retry it through the limiter, which blocks until it's safe to send, and an exhausted daily limit raises RateLimitError from the list call rather than being waited out.

A rate limiter object is scoped to the Session it's passed to — construct one per Session rather than sharing an instance across sessions using different credentials. Passing your own object works too, as long as it implements the acquire/on_rate_limited/release methods described in mokkari.rate_limit.RateLimiter. Leaving rate_limiter unset (the default) keeps the raise-immediately behavior described above.

Pacing across processes with Redis

HeaderPacedRateLimiter only sees the requests its own process sends, so several workers using one account can overrun the per-minute window together until Metron answers with 429s. mokkari.redis_rate_limit.RedisRateLimiter paces the same way but keeps its state in Redis, so every process and host using the same account shares one per-minute window, one daily estimate and one 429 backoff. Install the redis extra and pass it a client:

pip install mokkari[redis]
import redis

import mokkari
from mokkari.redis_rate_limit import RedisRateLimiter

client = redis.Redis(host="localhost", port=6379, socket_connect_timeout=1, socket_timeout=1)
# Any stable name for your Metron account, such as its username. Don't use the
# token itself: it becomes part of key names anyone with Redis access can read.
m = mokkari.api(
    api_token=api_token,
    rate_limiter=RedisRateLimiter(client, account="your-username"),
)

It raises RateLimitError on an exhausted daily limit just like HeaderPacedRateLimiter. Times come from the Redis server's clock, and every key expires on its own, so a worker that crashes mid-request can't leave the account blocked. If Redis is unreachable, acquire raises the client's connection error and the request isn't sent. Set socket_connect_timeout and socket_timeout on the client as above: without them, a Redis host that stops responding, rather than refusing the connection, makes acquire wait on it indefinitely. The limiter's own waits happen in the client, not in Redis, so a short timeout doesn't cut them off.

Caching

Pass a cache to keep responses locally, so repeated requests don't count against your rate limit:

import mokkari
from mokkari.sqlite_cache import SqliteCache

m = mokkari.api(api_token="your-token", cache=SqliteCache("mokkari_cache.db"))

import mokkari no longer loads mokkari.sqlite_cache in v5.0. If you're upgrading from 4.x and refer to mokkari.sqlite_cache.SqliteCache after only import mokkari, import the module explicitly as shown above.

Connection Reuse

A Session keeps its connections to Metron open between requests, so repeated calls don't pay a new TCP/TLS handshake each time. Cookies are never stored or sent back. To release the connections deterministically, use the session as a context manager or call close():

with mokkari.api(api_token="your-token") as m:
    issue = m.issue(1)

Closing is optional, and a closed session can still be used; it just opens new connections. If the server drops an idle connection, the underlying HTTP library notices and discards it when the next request checks it out, so a request opens a fresh connection instead. The only failure is a race: if the server closes a connection in the instant it is reused, that request raises an ApiError. This is rare, and mokkari doesn't retry it automatically. Don't share a Session across forked processes; create one per process.

Documentation

Read the project documentation

Bugs/Requests

Please use the GitHub issue tracker to submit bugs or request features.

Metadata

Release files for mokkari 5.0.0

For a detailed explanation of source distributions (sdists) and built distributions (wheels), please see the package formats documentation.

Source distribution (sdist)

Source distribution for mokkari 5.0.0
File Size Uploaded Provenance
mokkari-5.0.0.tar.gz 139.7 kB Without provenance Details

Built distribution (wheel)

Table of built distributions (wheels) for mokkari 5.0.0
File Interpreter ABI Platform
mokkari-5.0.0-py3-none-any.whl Python 3 none any Details

Total release size: 234.5 kB

Release files / mokkari-5.0.0.tar.gz

Download URL mokkari-5.0.0.tar.gz
Size 139.7 kB
Tags Source
SHA-256 checksum
How to use checksums
2205bca1799d9877d310e1a3797b5b1ff71b5ebfc8f91da3c9a59f7d3fb3b7e9
BLAKE2b-256 checksum
How to use checksums
8e9c120bb34e03bb12e1609653d99338fea38022a89b5b59eab337736434f0e7
Upload date
Uploaded using Trusted Publishing?
What is trusted publishing?
No
Uploaded via uv/0.13.5 {"installer":{"name":"uv","version":"0.13.5","subcommand":["publish"]},"python":null,"implementation":{"name":null,"version":null},"distro":{"name":"Ubuntu","version":"24.04","id":"noble","libc":null},"system":{"name":null,"release":null},"cpu":null,"openssl_version":null,"setuptools_version":null,"rustc_version":null,"ci":true}

Release files / mokkari-5.0.0-py3-none-any.whl

Download URL mokkari-5.0.0-py3-none-any.whl
Size 94.9 kB
Tags Python 3
SHA-256 checksum
How to use checksums
fe22994dadb5801e3ec619f1805eea56de5bb820858f99a35b3d5ec2d1e4221d
BLAKE2b-256 checksum
How to use checksums
0256c4e8b43e8d2f671934e1df337a5f9e63af1bcecafc57b145bd1495af8551
Upload date
Uploaded using Trusted Publishing?
What is trusted publishing?
No
Uploaded via uv/0.13.5 {"installer":{"name":"uv","version":"0.13.5","subcommand":["publish"]},"python":null,"implementation":{"name":null,"version":null},"distro":{"name":"Ubuntu","version":"24.04","id":"noble","libc":null},"system":{"name":null,"release":null},"cpu":null,"openssl_version":null,"setuptools_version":null,"rustc_version":null,"ci":true}

Security for mokkari 5.0.0

Provenance

Provenance describes where a file came from. On PyPI, provenance is shared via attestations, which provide a verifiable record of the build or publishing details. View details, limitations and caveats.

No provenance found

None of the files in this release were uploaded with provenance metadata.

A note on package security

PyPI is a repository for distributing packages, not a service for auditing them. While third-party security researchers and our staff do their best to detect and remove malicious packages, PyPI cannot guarantee the quality, security or suitability of any given package.

To learn more about protecting yourself from malicious actors, visit our package security documentation.

Release history Release notifications | RSS feed

This release

5.0.0 This release

2 release files No provenance

Anthropic, PBC Visionary sponsor Bloomberg Visionary sponsor Hudson River Trading Visionary sponsor Meta Visionary sponsor NVIDIA Visionary sponsor Microsoft Sustainability sponsor Depot Continuous Integration AWS Cloud computing and Security Sponsor Datadog Monitoring Fastly CDN Google Download Analytics Sentry Error logging StatusPage Status page