Metadata-Version: 2.4
Name: pyteleflow
Version: 0.1.4
Summary: Asynchronous framework for building Telegram bots with linear flows
Keywords: telegram,bot,async,python-telegram-bot,conversation,flow
Author: nebenwelt
License-Expression: MIT
License-File: LICENSE
Classifier: Intended Audience :: Developers
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Programming Language :: Python :: 3 :: Only
Classifier: Framework :: AsyncIO
Classifier: Topic :: Communications :: Chat
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Classifier: Typing :: Typed
Requires-Dist: python-telegram-bot[rate-limiter]>=22.8
Requires-Dist: tenacity>=9.1.4
Requires-Python: >=3.12
Project-URL: Homepage, https://github.com/nebenwelt/pyteleflow
Project-URL: Repository, https://github.com/nebenwelt/pyteleflow
Project-URL: Issues, https://github.com/nebenwelt/pyteleflow/issues
Description-Content-Type: text/markdown

![PyPI Version](https://img.shields.io/pypi/v/pyteleflow?style=plastic&color=blue) ![Static Badge](https://img.shields.io/badge/license-MIT-green?style=plastic) ![Static Badge](https://img.shields.io/badge/python-3.12+-blue?style=plastic&logo=python)

# Introduction

I've been wanting to simplify writing readable code for Telegram bots with deterministic conversation flows, primarily for personal admin panels and low-traffic services.

Most Python frameworks for Telegram bots rely on finite-state machines (FSMs) in one form or another. aiogram relies on a pile of decorators and a rather questionable concept called [Scenes Wizard](https://docs.aiogram.dev/en/latest/dispatcher/finite_state_machine/scene.html). python-telegram-bot provides [ConversationHandler](https://docs.python-telegram-bot.org/en/latest/telegram.ext.conversationhandler.html), which offers a semi-declarative way to define conversation transitions. There's also the well-known [aiogram_dialog](https://github.com/Tishka17/aiogram_dialog), which turns your chat into a form browser. This somewhat defeats the purpose of a chatbot: if you need elaborate web forms, just build a web page.

My approach is different. Instead of defining numerous states in a finite-state machine, scattering them throughout the codebase, and managing transitions that can be difficult to follow, you simply write sequential, procedural code:
```python
async def your_flow(ctx: Context):
  # Get the current chat with the user
  chat = ctx.initial_update.effective_chat

  # Send the user an initial message asking a question
  await chat.send_message("How old are you?")

  # Wait for the user to send a text message in the chat.
  # Other message types are ignored because we specified the TEXT filter.
  msg = await ctx.message(filters.TEXT)
  age = msg.text

  # Ask the user to send a photo
  await chat.send_message("Send me a photo of yourself")

  # Wait until the user sends a photo
  msg = await ctx.message(filters.PHOTO)
  photo = msg.photo
```

And invoke widgets wherever needed:
```python
# ...
result = await widgets.Select(
  ctx=ctx,
  prompt="Which file-sharing service do you use?",
  items=["Skype", "Telegram"],
).show_as_reply(msg)
# ...
```
The result? Less code to write and less code to read - even when the code is generated by an AI agent.
With existing frameworks, even these simple transitions would have to be defined declaratively and implemented using multiple callback handlers


# Dependencies
To interact with the Telegram Bot API, `pyteleflow` uses the popular `python-telegram-bot` library as an API wrapper. You can also take advantage of the extensive collection of built-in filters provided by `python-telegram-bot`.

Why not aiogram? 
`aiogram` relies on heavyweight Pydantic models, and its author doesn't appear to have any plans to move away from them.


# Quickstart
## Installation
Using uv:
```sh
uv add pyteleflow
```

Using pip:
```sh
pip install pyteleflow
```


## Simple Bot
> 💡 The code for this example is in [quickstart.py](./examples/quickstart.py)



1. Import the dependencies and initialize logging:
```python
import asyncio as aio
import os
import logging

from telegram.ext import ExtBot, AIORateLimiter, filters
from pyteleflow import FlowApp, Context, widgets, ConcurrencyStrategy

logging.basicConfig(level=logging.DEBUG)
logging.getLogger("httpcore").setLevel(logging.WARNING)
logging.getLogger("telegram").setLevel(logging.WARNING)
```

2. Create a Flow that collects a simple user profile:
```python
async def quickstart_flow(ctx: Context):
  # Get the current chat with the user
  chat = ctx.initial_update.effective_chat

  # Send the user an initial message asking a question
  await chat.send_message("What’s your name?")

  # Wait for the user to send a text message in the chat.
  # Other message types are ignored because we specified the TEXT filter.
  msg = await ctx.message(filters.TEXT)
  user_name = msg.text

  # Ask the user to send a photo
  await chat.send_message("Send me a photo of yourself")

  # Wait until the user sends a photo
  msg = await ctx.message(filters.PHOTO)
  user_photo = msg.photo[0]

  # Show the user a widget with buttons and wait for their selection
  result = await widgets.Select(
    ctx=ctx,
    prompt="Your profile is complete. Would you like to save it?",
    items=["Save", "View profile"],
    direction="horizontal",
  ).show(chat.id, cancel_button=True)

  if not result.canceled:
    if result.item == "Save":
      alert = await chat.send_message("Profile saved")
      await aio.sleep(2)
      await alert.delete()
    elif result.item == "View profile":
      await chat.send_photo(
        photo=user_photo,
        caption=f"PROFILE\nUser name:  {user_name}\nUser ID:  {msg.from_user.id}",
      )
```

3. Create the application, specifying the Flow to run, and start it:

```python
async def main():
  bot_token = os.environ.get("BOT_TOKEN")

  app = FlowApp(
    ExtBot(bot_token, rate_limiter=AIORateLimiter()),
    entry_coro=quickstart_flow,
    spawn_trigger=filters.Text(["/start"]),
    concurrency_strategy=ConcurrencyStrategy.CHAT
  )

  try:
    await app.run_polling()
  finally:
    await app.shutdown()


if __name__ == "__main__":
  aio.run(main())
```

4. To start a conversation with the bot, send the `/start` command in the bot's chat.


# Overview
## Framework specifics and limitations

__Conversation state does not survive restarts.__

The simplicity of writing and reading Flows comes at a cost. In the current architecture, active Flow instances are stored only in process memory. If the bot restarts, all conversations will be lost, and users will have to start them over. Python cannot serialize and restore the stack of `asyncio` coroutines. As a result, a coroutine cannot resume from its suspension point after the process restarts. There are theoretically a couple of ways to address this problem, but they would make Flow code more complicated.

How critical is this?
There are different opinions on this. For example, web users are accustomed to a browser page's front-end state being reset when the page is refreshed, while the back-end state (user data) is stored persistently and survives restarts. Telegram bots are also the front-end state of your application.

## Widgets
Currently, `pyteleflow.widgets` includes only a few widgets. More may be added in the future, but for now these widgets serve as examples of how you can implement your own.

## Flow cancellation handling
Spawned Flow instances can be cancelled for several reasons:

__Flow inactivity.__ The last activity is considered to be the moment you receive a message through the context, when `ctx.message()` returns a received message that matches the specified filter
```python
msg = await ctx.message(filters.PHOTO)
```
If the period of inactivity exceeds `flow_inactivity_timeout` (set when creating `FlowApp`), the Flow instance will be cancelled. You can change this timeout for an individual Flow during the conversation through the context:
```python
ctx.inactivity_timeout = timedelta(minutes=20)
```

__Flow respawn.__ If a filter is configured in `respawn_trigger` (set when creating `FlowApp`), for example `filters.Text(["/start"])`, the active Flow instance will first be cancelled and then restarted from the beginning.

__Stopping the bot application.__ Always use the `FlowApp.shutdown()` method when stopping the application. This method initiates cancellation of all active Flow instances.


When a coroutine is cancelled, an `asyncio.CancelledError` exception is raised inside it. You can handle it if necessary:
```python
async def your_flow(ctx: Context):
  try:
    # ...
    msg = ctx.message(filter.PHOTO)
    # ...
  except asyncio.CancelledError:
    # Handle Flow cancellation

    raise
```

## Middleware

> 💡 The code for this example is in [middleware.py](./examples/middleware.py)

Middleware allows you to intercept incoming `Update` objects before they reach a Flow. You can use middleware to restrict access, log incoming updates, modify an `Update` before processing, and more.

If a middleware returns `None`, the remaining middleware chain is skipped, and the `Update` is discarded.

```python
from pyteleflow.middleware import Middleware

class Authentication(Middleware):
  def __init__(self, allowed_user_ids: Iterable[int]):
    self._allowed_user_ids: set[int] = set(allowed_user_ids)

  def __call__(self, update: Update | None):
    if update.effective_user is not None and update.effective_user.id in self._allowed_user_ids:
      return update

    return None
```

Middleware is executed in the order in which it is added:

```python
app.add_pre_middlewares(
    [Authentication(allowed_user_ids=(111111, 222222, 3333333))],
)
```

## Background tasks
`pyteleflow` supports running background tasks. See the example in [background_task.py](./examples/background_task.py) 
