utilities.md

Utilities

ClientPool

Distribute requests across multiple accounts with round-robin:

from teletype.utils.pool import ClientPool
from teletype import GetBlogByUri


# From tokens
pool = ClientPool.from_tokens(["token1", "token2", "token3"])

# Or from client instances
pool = ClientPool([client1, client2, client3], max_concurrent=10)

async with pool:
    # Single request (round-robin)
    blog = await pool(GetBlogByUri(uri="techblog"))

    # Batch execution across all clients
    methods = [GetBlogByUri(uri=u) for u in uris]
    results = await pool.gather(methods, max_concurrent=20)

    # Pool health
    stats = await pool.health()
    print(f"{stats.healthy}/{stats.total} clients healthy")

BlogExporter

Export an entire blog to local files (JSON or Markdown):

from teletype import TeletypeClient
from teletype.utils.exporter import BlogExporter, ExportFormat


async with TeletypeClient("token") as client:
    exporter = BlogExporter(
        client,
        format=ExportFormat.MARKDOWN,  # or ExportFormat.JSON
        include_comments=True,
        max_concurrent=5,
    )

    result = await exporter.export("techblog", "./backup")
    # articles_count counts articles actually written — an article that failed to
    # export is logged as a warning and left out of the count.
    print(f"Exported {result.articles_count} articles")
    print(f"Comments: {result.comments_count}")
    print(f"Files: {len(result.files)}")

Output structure:

backup/
├── index.json              # article list with metadata
├── my-first-post.md        # article content
├── another-article.md
└── ...

BulkPublisher

Publish multiple articles with concurrency control and progress tracking:

from teletype import TeletypeClient
from teletype.utils.publisher import BulkPublisher, PublishItem
from teletype.utils.content import DocumentBuilder


async with TeletypeClient("token") as client:
    account = await client.get_account()

    items = [
        PublishItem(
            author_id=account.blog_id,
            creator_id=account.blog_id,
            title=f"Post #{i}",
            text=DocumentBuilder().paragraph(f"Content for post {i}").build(),
            cut_text=f"Content for post {i}",
        )
        for i in range(10)
    ]


    async def on_progress(done: int, total: int) -> None:
        print(f"Published {done}/{total}")


    publisher = BulkPublisher(
        client,
        max_concurrent=3,
        on_progress=on_progress,
    )
    result = await publisher.publish(items)

    print(f"Success: {result.succeeded}, Failed: {result.failed}")
    for err in result.errors:
        print(f"  Failed: {err.item.title} — {err.exception}")

StatsCollector

Aggregate engagement stats across all blog articles:

from teletype import TeletypeClient
from teletype.utils.stats import StatsCollector


async with TeletypeClient("token") as client:
    collector = StatsCollector(client, top_n=5)
    stats = await collector.collect("techblog")

    print(f"Total articles: {stats.total_articles}")
    print(f"Total views: {stats.total_views}")
    print(f"Total comments: {stats.total_comments}")
    print(f"Total reactions: {stats.total_reactions}")

    print("\nTop by views:")
    for s in stats.top_by_views:
        print(f"  {s.views:>6} views  {s.title}")

    print("\nTop by reactions:")
    for s in stats.top_by_reactions:
        print(f"  {s.reactions:>4} reactions  {s.title}")

You can also aggregate from pre-fetched articles:

articles = await client.get_blog_articles("techblog", limit=100)
stats = await collector.collect_from_articles("techblog", articles)

ArticleWatcher

Poll articles for changes and new comments:

from teletype import TeletypeClient
from teletype.utils.watcher import ArticleWatcher


async with TeletypeClient("token") as client:
    watcher = ArticleWatcher(
        client,
        article_ids=[12345, 67890],
        interval=60.0,  # poll every 60 seconds
        watch_comments=True,  # track individual new comments
    )


    @watcher.on_change
    async def handle_change(change):
        print(
            f"Article {change.article_id}: "
            f"{change.field} {change.old_value} → {change.new_value} "
            f"(+{change.delta})"
        )


    @watcher.on_comment
    async def handle_comment(event):
        print(f"New comment on {event.article_id}: {event.comment.text}")


    await watcher.start()

    # ... run your app ...

    await watcher.stop()

The watcher takes an initial snapshot of each article's metrics, then polls periodically. Callbacks fire when views, comments, reactions, or reposts change.

HookRegistry

Register post-parse callbacks per model type:

from teletype import TeletypeClient, Article, Blog


async with TeletypeClient("token") as client:
    @client.on(Article)
    def on_article(article: Article) -> Article:
        print(f"[Article] {article.title} ({article.views} views)")
        return article


    @client.on(Blog)
    def on_blog(blog: Blog) -> Blog:
        print(f"[Blog] {blog.name}")
        return blog


    # Hooks fire automatically
    blog = await client.get_blog_by_uri("techblog")  # triggers on_blog
    article = await client.get_article(12345)  # triggers on_article

Hooks run after response parsing, before the result is returned. They receive the parsed object and must return it ( possibly modified). Multiple hooks per type are supported and execute in registration order.