tessl/pypi-telegraph

Python Telegraph API wrapper for creating and managing Telegraph pages and accounts

Overview

Eval results

Files

Asynchronous API

Name: tessl/pypi-telegraph
Author: tessl

Complete async/await interface for Telegraph operations using httpx for HTTP requests. All synchronous methods have async counterparts with identical signatures and functionality.

Installation

pip install 'telegraph[aio]'

This installs the httpx dependency required for async operations.

Import and Setup

from telegraph.aio import Telegraph
import asyncio

# Create async Telegraph client
telegraph = Telegraph(access_token='your_token')

Capabilities

Account Management (Async)

All account operations with async/await support.

async def create_account(short_name: str, author_name: str = None, author_url: str = None, replace_token: bool = True) -> dict:
    """
    Create a new Telegraph account (async).
    
    Parameters and returns identical to synchronous version.
    """

async def edit_account_info(short_name: str = None, author_name: str = None, author_url: str = None) -> dict:
    """
    Update Telegraph account information (async).
    """

async def get_account_info(fields: list = None) -> dict:
    """
    Get Telegraph account information (async).
    """

async def revoke_access_token() -> dict:
    """
    Revoke current access token and generate new one (async).
    """

Usage example:

import asyncio
from telegraph.aio import Telegraph

async def setup_account():
    telegraph = Telegraph()
    
    # Create account
    response = await telegraph.create_account(
        short_name='async-blog',
        author_name='Async Author'
    )
    
    # Get account info
    info = await telegraph.get_account_info()
    print(f"Account: {info['short_name']}")
    
    return telegraph

# Run async function
telegraph = asyncio.run(setup_account())

Page Operations (Async)

All page management operations with async support.

async def create_page(title: str, content: list = None, html_content: str = None, author_name: str = None, author_url: str = None, return_content: bool = False) -> dict:
    """
    Create a new Telegraph page (async).
    """

async def edit_page(path: str, title: str, content: list = None, html_content: str = None, author_name: str = None, author_url: str = None, return_content: bool = False) -> dict:
    """
    Edit an existing Telegraph page (async).
    """

async def get_page(path: str, return_content: bool = True, return_html: bool = True) -> dict:
    """
    Get a Telegraph page (async).
    """

async def get_page_list(offset: int = 0, limit: int = 50) -> dict:
    """
    Get list of account pages (async).
    """

async def get_views(path: str, year: int = None, month: int = None, day: int = None, hour: int = None) -> dict:
    """
    Get page view statistics (async).
    """

Usage example:

async def manage_pages():
    telegraph = Telegraph(access_token='your_token')
    
    # Create multiple pages concurrently
    tasks = []
    for i in range(3):
        task = telegraph.create_page(
            title=f'Async Page {i+1}',
            html_content=f'<p>This is page {i+1} created asynchronously.</p>'
        )
        tasks.append(task)
    
    # Wait for all pages to be created
    results = await asyncio.gather(*tasks)
    
    # Print URLs
    for result in results:
        print(f"Created: {result['url']}")
    
    # Get page list
    page_list = await telegraph.get_page_list()
    print(f"Total pages: {page_list['total_count']}")

asyncio.run(manage_pages())

File Upload (Async)

Asynchronous file upload functionality.

async def upload_file(f) -> list:
    """
    Upload file to Telegraph servers (async, unofficial API).
    
    Parameters and returns identical to synchronous version.
    """

Usage example:

async def upload_and_create():
    telegraph = Telegraph()
    
    # Upload file
    result = await telegraph.upload_file('image.jpg')
    image_src = result[0]['src']
    
    # Create page with uploaded image
    response = await telegraph.create_page(
        title='Async Upload Example',
        html_content=f'<img src="{image_src}" alt="Async uploaded image">'
    )
    
    print(f"Page with image: {response['url']}")

asyncio.run(upload_and_create())

Async Context Management

For better connection management, consider using the async context manager pattern:

import asyncio
from telegraph.aio import Telegraph

async def main():
    telegraph = Telegraph()
    
    try:
        # Perform operations
        await telegraph.create_account(short_name='context-example')
        
        response = await telegraph.create_page(
            title='Context Example',
            html_content='<p>Using proper async context management.</p>'
        )
        
        print(f"Page created: {response['url']}")
        
    finally:
        # Close the HTTP client session
        await telegraph._telegraph.session.aclose()

asyncio.run(main())

Concurrent Operations

Take advantage of async capabilities for concurrent operations:

async def concurrent_example():
    telegraph = Telegraph(access_token='your_token')
    
    # Perform multiple operations concurrently
    account_task = telegraph.get_account_info()
    pages_task = telegraph.get_page_list(limit=5)
    
    # Wait for both to complete
    account_info, page_list = await asyncio.gather(account_task, pages_task)
    
    print(f"Account: {account_info['short_name']}")
    print(f"Pages: {page_list['total_count']}")
    
    # Process pages concurrently
    view_tasks = []
    for page in page_list['pages']:
        task = telegraph.get_views(page['path'])
        view_tasks.append(task)
    
    view_results = await asyncio.gather(*view_tasks)
    
    for page, views in zip(page_list['pages'], view_results):
        print(f"{page['title']}: {views['views']} views")

asyncio.run(concurrent_example())

Error Handling

Async operations raise the same exceptions as synchronous versions:

from telegraph.aio import Telegraph
from telegraph.exceptions import TelegraphException, RetryAfterError

async def error_handling_example():
    telegraph = Telegraph()
    
    try:
        response = await telegraph.create_account(short_name='test')
    except RetryAfterError as e:
        print(f"Rate limited. Retry after {e.retry_after} seconds")
        await asyncio.sleep(e.retry_after)
        # Retry operation
        response = await telegraph.create_account(short_name='test')
    except TelegraphException as e:
        print(f"API error: {e}")

asyncio.run(error_handling_example())

Migration from Sync

Converting synchronous code to async is straightforward:

# Synchronous version
from telegraph import Telegraph

telegraph = Telegraph()
response = telegraph.create_account(short_name='sync-example')
page = telegraph.create_page(title='Sync Page', html_content='<p>Sync content</p>')

# Asynchronous version
from telegraph.aio import Telegraph
import asyncio

async def async_version():
    telegraph = Telegraph()
    response = await telegraph.create_account(short_name='async-example')
    page = await telegraph.create_page(title='Async Page', html_content='<p>Async content</p>')

asyncio.run(async_version())