Patreon Archiver

Python versions PyPI - Version GitHub tag (with filter) License GitHub commits since latest release (by SemVer including pre-releases) CodeQL QA Tests Coverage Status Dependabot Documentation Status mypy uv pytest Ruff Downloads Stargazers pre-commit Prettier Follow @Tatsh Mastodon Follow

Save Patreon content you have access to.

Commands

patreon-archiver

Archive Patreon data you have access to.

Usage

patreon-archiver [OPTIONS] CAMPAIGN_ID

Options

-o, --output-dir <output_dir>

Output directory.

-b, --browser <browser>

Browser to read cookies from.

-p, --profile <profile>

Browser profile.

-c, --cookies-json <cookies_json>

Path to JSON file containing cookies (overrides –browser/–profile).

-x, --fail

Do not continue processing after a failed yt-dlp command.

-P, --use-yt-dlp-for-podcasts

Use yt-dlp to download podcasts.

-S, --sleep-time <sleep_time>

Number of seconds to wait between requests.

-d, --debug

Enable debug output.

-q, --quiet

Disable progress spinner updates.

Arguments

CAMPAIGN_ID

Required argument

Constants

Constants.

patreon_archiver.constants.FIELDS

Field parameters for Patreon API requests.

patreon_archiver.constants.MEDIA_POST_TYPES

Post types treated as media posts.

patreon_archiver.constants.MEDIA_URI

URI for the Patreon media endpoint.

patreon_archiver.constants.PATREON_API_URI

Base URI for the Patreon API.

patreon_archiver.constants.POSTS_URI

URI for the Patreon posts endpoint.

patreon_archiver.constants.SHARED_HEADERS

Headers sent with every Patreon API request.

patreon_archiver.constants.SHARED_PARAMS

Shared query parameters for Patreon API requests.

patreon_archiver.constants.USER_AGENT

User agent string used for Patreon API requests.

Typing

Typing helpers.

class patreon_archiver.typing.AudioEmbedAttributes

Attributes for audio embed posts.

post_type : Literal['audio_embed']

Post type.

class patreon_archiver.typing.AudioFileAttributes

Attributes for audio file posts.

post_type : Literal['audio_file']

Post type.

class patreon_archiver.typing.CommonAttributes

Attributes shared by all post types.

url : str

URL of the post.

patreon_archiver.typing.IMAGES_PROCESSED

Counter key for image posts that have been saved successfully.

class patreon_archiver.typing.ImageFileAttributes

Attributes for image file posts.

post_file : PostFile

File.

post_metadata : ImageFileAttributesPostMetadata | None

Metadata.

post_type : Literal['image_file']

Post type.

class patreon_archiver.typing.ImageFileAttributesPostMetadata

Metadata for image file posts.

image_order : Sequence[str]

Order of images in the post.

Links for pagination.

next : str | None

Next page URI.

class patreon_archiver.typing.LivestreamCrowdcastAttributes

Attributes for ‘livestream crowdcast’ posts.

post_type : Literal['livestream_crowdcast']

Post type.

class patreon_archiver.typing.Media

Container for media data.

data : MediaData

Media data.

class patreon_archiver.typing.MediaData

Data for a media item.

attributes : MediaDataAttributes

Media data attributes.

id : str

ID.

class patreon_archiver.typing.MediaDataAttributes

Attributes for media data.

download_url : str

URL to download the media file.

file_name : str

Original file name of the media.

image_urls : MediaDataAttributesImageURLs

URLs for images.

mimetype : str

MIME type of the media.

class patreon_archiver.typing.MediaDataAttributesImageURLs

URLs for media data attributes.

original : str

URL for the original image.

patreon_archiver.typing.OTHERS_PROCESSED

Counter key for non-media posts that have been saved successfully.

patreon_archiver.typing.OnMessage

Callback used to report human-readable progress updates.

alias of Callable[[str], None]

patreon_archiver.typing.PODCASTS_PROCESSED

Counter key for podcast posts that have been saved successfully.

patreon_archiver.typing.POSTS_HANDLED

Counter key for posts routed by the producer.

class patreon_archiver.typing.PodcastAttributes

Attributes for podcast posts.

post_type : Literal['podcast']

Post type.

patreon_archiver.typing.PostAttributes : TypeAlias = patreon_archiver.typing.AudioEmbedAttributes | patreon_archiver.typing.AudioFileAttributes | patreon_archiver.typing.ImageFileAttributes | patreon_archiver.typing.VideoEmbedAttributes | patreon_archiver.typing.LivestreamCrowdcastAttributes | patreon_archiver.typing.PodcastAttributes

Union of all supported post attribute payloads.

class patreon_archiver.typing.PostFile

File information for posts.

name : str

Name of the file.

url : str

URL of the file.

class patreon_archiver.typing.Posts

Container for Patreon posts from API.

data : Sequence[PostsData]

List of post data.

Links for pagination.

class patreon_archiver.typing.PostsData

Data for a Patreon post.

attributes : AudioEmbedAttributes | AudioFileAttributes | ImageFileAttributes | VideoEmbedAttributes | LivestreamCrowdcastAttributes | PodcastAttributes

Attributes of the post.

id : str

ID of the post.

relationships : PostsDataRelationships

Relationships for the post.

class patreon_archiver.typing.PostsDataRelationships

Relationships for post data.

media : NotRequired[PostsDataRelationshipsMedia]

Media data.

class patreon_archiver.typing.PostsDataRelationshipsMedia

Media data of a post.

data : NotRequired[list[PostsDataRelationshipsMediaData]]

Media data.

class patreon_archiver.typing.PostsDataRelationshipsMediaData

Media data of a post.

id : str

ID of the media.

type : Literal['media']

Type of the media.

class patreon_archiver.typing.SaveInfo

Information about a saved post.

post_data_dict : PostsData

Data dictionary for the post.

target_dir : Path

Directory where the post is saved.

class patreon_archiver.typing.Stats

Live pipeline statistics shown in the progress spinner.

class patreon_archiver.typing.VideoEmbedAttributes

Attributes for video embed posts.

post_type : Literal['video_embed']

Post type.

class patreon_archiver.typing.YTDLPState(current_index: int = 0, current_uri: str | None = None, total_uris: int = 0)

Mutable yt-dlp progress state shared between the producer and the yt-dlp worker.

current_index : int = 0

1-based index of the URI currently being processed.

current_uri : str | None = None

URI yt-dlp is currently downloading, or None when idle.

render() str | None

Build the YT_DLP_STATUS value from the current state.

Returns:

Rendered status string, or None when no URI is active.

Return type:

str | None

total_uris : int = 0

Running total of URIs enqueued for the yt-dlp worker.

patreon_archiver.typing.YT_DLP_STATUS

Status-line key for the current yt-dlp URI.

Utilities

Utilities.

exception patreon_archiver.utils.UnknownMimetypeError

Exception raised when an unknown MIME type is encountered.

async patreon_archiver.utils.get_all_posts(campaign_id: str, session: AsyncSession, *, on_message: OnMessage | None = None) AsyncIterator[PostsData]

Yield all posts for a campaign.

Parameters:
campaign_id : str

The campaign ID to fetch posts for.

session : AsyncSession

The niquests async session to use for API calls.

on_message : OnMessage | None

Optional callback that receives progress text updates.

Yields:

PostsData – A single post payload from the Patreon API.

async patreon_archiver.utils.save_images(session: AsyncSession, pdd: PostsData, *, on_message: OnMessage | None = None) SaveInfo

Save images.

Parameters:
session : AsyncSession

The niquests async session to use for downloads.

pdd : PostsData

The post data dictionary.

on_message : OnMessage | None

Optional callback that receives progress text updates.

Returns:

Information about the saved images.

Return type:

SaveInfo

patreon_archiver.utils.save_other(pdd: PostsData, *, on_message: OnMessage | None = None) SaveInfo

Save other post types.

Parameters:
pdd : PostsData

The post data dictionary.

on_message : OnMessage | None

Optional callback that receives progress text updates.

Returns:

Information about the saved post.

Return type:

SaveInfo

async patreon_archiver.utils.save_podcast(session: AsyncSession, pdd: PostsData, *, on_message: OnMessage | None = None) SaveInfo

Save podcast posts.

Parameters:
session : AsyncSession

The niquests async session to use for downloads.

pdd : PostsData

The post data dictionary.

on_message : OnMessage | None

Optional callback that receives progress text updates.

Returns:

Information about the saved podcast.

Return type:

SaveInfo

Workers

Worker orchestration for asynchronous post processing.

exception patreon_archiver.workers.WorkerAbort

Worker-level abort signal for graceful CLI handling.

async patreon_archiver.workers.image_worker(image_queue: asyncio.Queue[PostsData | None], first_exception: list[BaseException], session: AsyncSession, stop_event: asyncio.Event, *, on_cleanup: OnMessage | None = None, on_message: OnMessage | None = None, stats: Stats | None = None) None

Save image posts sequentially.

Parameters:
image_queue : asyncio.Queue[PostsData | None]

Queue containing image post payloads.

first_exception : list[BaseException]

Mutable container for the first observed fatal exception.

session : AsyncSession

Session used for image downloads.

stop_event : asyncio.Event

Event indicating that workers should stop.

on_cleanup : OnMessage | None

Optional callback that receives cleanup status updates.

on_message : OnMessage | None

Optional callback that receives progress text updates.

stats : Stats | None

Optional live statistics object updated after each saved image post.

async patreon_archiver.workers.other_worker(first_exception: list[BaseException], other_queue: asyncio.Queue[PostsData | None], stop_event: asyncio.Event, *, on_cleanup: OnMessage | None = None, on_message: OnMessage | None = None, stats: Stats | None = None) None

Save other post types sequentially.

Parameters:
first_exception : list[BaseException]

Mutable container for the first observed fatal exception.

other_queue : asyncio.Queue[PostsData | None]

Queue containing unsupported post payloads.

stop_event : asyncio.Event

Event indicating that workers should stop.

on_cleanup : OnMessage | None

Optional callback that receives cleanup status updates.

on_message : OnMessage | None

Optional callback that receives progress text updates.

stats : Stats | None

Optional live statistics object updated after each saved non-media post.

async patreon_archiver.workers.podcast_worker(first_exception: list[BaseException], podcast_queue: asyncio.Queue[PostsData | None], session: AsyncSession, stop_event: asyncio.Event, *, on_cleanup: OnMessage | None = None, on_message: OnMessage | None = None, stats: Stats | None = None) None

Save podcast posts sequentially.

Parameters:
first_exception : list[BaseException]

Mutable container for the first observed fatal exception.

podcast_queue : asyncio.Queue[PostsData | None]

Queue containing podcast post payloads.

session : AsyncSession

Session used for podcast downloads.

stop_event : asyncio.Event

Event indicating that workers should stop.

on_cleanup : OnMessage | None

Optional callback that receives cleanup status updates.

on_message : OnMessage | None

Optional callback that receives progress text updates.

stats : Stats | None

Optional live statistics object updated after each saved podcast post.

async patreon_archiver.workers.producer(campaign_id: str, image_queue: asyncio.Queue[PostsData | None], other_queue: asyncio.Queue[PostsData | None], podcast_queue: asyncio.Queue[PostsData | None], session: AsyncSession, stop_event: asyncio.Event, *, on_cleanup: OnMessage | None = None, on_message: OnMessage | None = None, stats: Stats | None = None, use_yt_dlp_for_podcasts: bool, yt_dlp_queue: asyncio.Queue[str | None], yt_dlp_state: YTDLPState | None = None) None

Produce classified work items from Patreon posts.

Parameters:
campaign_id : str

Campaign ID to fetch posts for.

image_queue : asyncio.Queue[PostsData | None]

Queue receiving image posts.

other_queue : asyncio.Queue[PostsData | None]

Queue receiving unsupported post types.

podcast_queue : asyncio.Queue[PostsData | None]

Queue receiving podcast posts for native processing.

session : AsyncSession

Session used to fetch Patreon posts.

stop_event : asyncio.Event

Event indicating that workers should stop.

on_cleanup : OnMessage | None

Optional callback that receives cleanup status updates.

on_message : OnMessage | None

Optional callback that receives progress text updates.

stats : Stats | None

Optional live statistics object updated as posts are routed.

use_yt_dlp_for_podcasts : bool

If True, route podcast URLs to yt-dlp.

yt_dlp_queue : asyncio.Queue[str | None]

Queue receiving media URLs for yt-dlp.

yt_dlp_state : YTDLPState | None

Optional yt-dlp progress state whose total_uris counter the producer increments each time a URI is enqueued for yt-dlp.

async patreon_archiver.workers.run_workers(campaign_id: str, first_exception: list[BaseException], session: AsyncSession, stop_event: asyncio.Event, *, fail: bool, on_cleanup: OnMessage | None = None, on_message: OnMessage | None = None, stats: Stats | None = None, use_yt_dlp_for_podcasts: bool, ydl: AsyncYoutubeDL, yt_dlp_idle_event: asyncio.Event | None = None, yt_dlp_state: YTDLPState | None = None) None

Run producer and workers for all post types.

Parameters:
campaign_id : str

Campaign ID to fetch posts for.

first_exception : list[BaseException]

Mutable container for the first observed fatal exception.

session : AsyncSession

Session used by producer and download workers.

stop_event : asyncio.Event

Event indicating that workers should stop.

fail : bool

Whether yt-dlp failures should abort processing.

on_cleanup : OnMessage | None

Optional callback that receives cleanup status updates.

on_message : OnMessage | None

Optional callback that receives progress text updates.

stats : Stats | None

Optional live statistics object. When provided, the producer updates posts_handled and each worker updates its processed counter. The yt_dlp_status line is updated when yt_dlp_state is also provided.

use_yt_dlp_for_podcasts : bool

If True, route podcast URLs to yt-dlp instead of the podcast worker.

ydl : AsyncYoutubeDL

Configured yt-dlp wrapper instance.

yt_dlp_idle_event : asyncio.Event | None

Optional event forwarded to yt_dlp_worker() to observe when the yt-dlp worker is idle.

yt_dlp_state : YTDLPState | None

Optional yt-dlp progress state shared between the producer and the yt-dlp worker for composing the rendered status line.

async patreon_archiver.workers.yt_dlp_worker(*, fail: bool, first_exception: list[BaseException], idle_event: asyncio.Event | None = None, on_cleanup: OnMessage | None = None, on_message: OnMessage | None = None, stats: Stats | None = None, stop_event: asyncio.Event, ydl: AsyncYoutubeDL, yt_dlp_queue: asyncio.Queue[str | None], yt_dlp_state: YTDLPState | None = None) None

Process yt-dlp URIs one download at a time.

Parameters:
fail : bool

Whether to stop on the first yt-dlp failure.

first_exception : list[BaseException]

Mutable container for the first observed fatal exception.

idle_event : asyncio.Event | None

Optional event that is set whenever the worker is not executing a ydl.download call and cleared while one is in progress. Starts set.

on_cleanup : OnMessage | None

Optional callback that receives cleanup status updates.

on_message : OnMessage | None

Optional callback that receives progress text updates.

stats : Stats | None

Optional live statistics object updated with the current yt-dlp URI.

stop_event : asyncio.Event

Event indicating that workers should stop.

ydl : AsyncYoutubeDL

Configured yt-dlp wrapper returned by get_configured_yt_dlp().

yt_dlp_queue : asyncio.Queue[str | None]

Queue containing URIs for yt-dlp to handle.

yt_dlp_state : YTDLPState | None

Optional yt-dlp progress state updated with the current URI and index.

Indices and tables