Checkpoints — TeleGlance

What are checkpoints?

Checkpoints track your position in a message stream so you can resume without reprocessing. They store the oldest and newest message IDs you've seen, along with the channel name to prevent accidental reuse across channels.

JsonCheckpointStore

Built-in file-based checkpoint storage using versioned JSON.

Python

from teleglance import JsonCheckpointStore

store = JsonCheckpointStore("state.json")

# Load checkpoint (returns None if not found)
state = await store.load("stream:telegram")

# Start from checkpoint or beginning
after_id = state.newest_id if state else None

async for message in client.iter_new_messages("telegram", after=after_id):
    # Process message
    print(message.id)

    # Update checkpoint
    if state is None:
        from teleglance import CheckpointState
        state = CheckpointState(channel="telegram", oldest_id=message.id, newest_id=message.id)
    else:
        state = state.record(message)

    # Save periodically
    await store.save("stream:telegram", state)

Resume history iteration

Use checkpoints to resume backward pagination through history.

Python

store = JsonCheckpointStore("history.json")
state = await store.load("history:telegram")

# Resume from where we left off
before_id = state.oldest_id if state else None

async for message in client.iter_messages("telegram", limit=1000, before=before_id):
    process(message)

    # Update checkpoint with oldest seen ID
    if state is None:
        state = CheckpointState(channel="telegram", oldest_id=message.id, newest_id=message.id)
    else:
        state = state.record(message)

    # Save every 50 messages
    if message.id % 50 == 0:
        await store.save("history:telegram", state)

Custom checkpoint backends

Implement the CheckpointStore protocol for databases or other storage:

Python

from teleglance import CheckpointStore, CheckpointState

class RedisCheckpointStore(CheckpointStore):
    def __init__(self, redis_client):
        self.redis = redis_client

    async def load(self, key: str) -> CheckpointState | None:
        data = await self.redis.get(f"checkpoint:{key}")
        if data:
            return CheckpointState.model_validate_json(data)
        return None

    async def save(self, key: str, state: CheckpointState) -> None:
        await self.redis.set(f"checkpoint:{key}", state.model_dump_json())

store = RedisCheckpointStore(redis_client)
state = await store.load("stream:telegram")