← cd ..

Building a Telegram Bot with PostgreSQL

Most tutorial bots store nothing — every restart wipes the state. Real bots need to remember users, track orders, store preferences. PostgreSQL is the right tool for this: battle-tested, relational, and fully async with asyncpg.

This guide shows how to wire PostgreSQL into an Aiogram bot cleanly, from connection pooling to migrations.

# Stack

• Python 3.11+
• Aiogram 3.x
• asyncpg (async PostgreSQL driver)
• PostgreSQL 15+

# Setting Up asyncpg Pool

Never create a new connection per request — use a connection pool that's shared across the bot's lifetime.

# db.py
import asyncpg
from typing import Optional

_pool: Optional[asyncpg.Pool] = None

async def create_pool(dsn: str) -> asyncpg.Pool:
    global _pool
    _pool = await asyncpg.create_pool(dsn, min_size=2, max_size=10)
    return _pool

def get_pool() -> asyncpg.Pool:
    if _pool is None:
        raise RuntimeError("Pool not initialized")
    return _pool

async def close_pool():
    if _pool:
        await _pool.close()

# Creating Tables on Startup

Run migrations before the bot starts polling. Keep it simple — one SQL file per migration, applied in order.

-- migrations/001_users.sql
CREATE TABLE IF NOT EXISTS users (
    id         BIGINT PRIMARY KEY,  -- Telegram user ID
    username   TEXT,
    first_name TEXT NOT NULL,
    created_at TIMESTAMPTZ DEFAULT now(),
    settings   JSONB DEFAULT '{}'
);

# main.py
import asyncio
from aiogram import Bot, Dispatcher
from db import create_pool, close_pool, get_pool

async def main():
    pool = await create_pool("postgresql://user:pass@localhost/botdb")

    # run migrations
    async with pool.acquire() as conn:
        with open("migrations/001_users.sql") as f:
            await conn.execute(f.read())

    bot = Bot("TOKEN")
    dp = Dispatcher()
    # ... include routers ...

    try:
        await dp.start_polling(bot)
    finally:
        await close_pool()

asyncio.run(main())

# Saving Users on /start

Insert or update the user record every time they start the bot. ON CONFLICT DO UPDATE handles both new and returning users in one query.

# handlers/start.py
from aiogram import Router
from aiogram.filters import Command
from aiogram.types import Message
from db import get_pool

router = Router()

@router.message(Command("start"))
async def cmd_start(msg: Message):
    pool = get_pool()
    await pool.execute("""
        INSERT INTO users (id, username, first_name)
        VALUES ($1, $2, $3)
        ON CONFLICT (id) DO UPDATE
          SET username   = EXCLUDED.username,
              first_name = EXCLUDED.first_name
    """, msg.from_user.id, msg.from_user.username, msg.from_user.first_name)

    await msg.answer(f"Welcome, {msg.from_user.first_name}!")

# Reading User Data

@router.message(Command("profile"))
async def cmd_profile(msg: Message):
    pool = get_pool()
    row = await pool.fetchrow(
        "SELECT * FROM users WHERE id = $1",
        msg.from_user.id
    )
    if not row:
        await msg.answer("Not registered yet. Send /start.")
        return

    await msg.answer(
        f"ID: {row['id']}\n"
        f"Name: {row['first_name']}\n"
        f"Joined: {row['created_at'].strftime('%Y-%m-%d')}"
    )

# Passing Pool via Middleware

# middleware.py
from typing import Any, Callable, Awaitable
from aiogram import BaseMiddleware
from aiogram.types import TelegramObject
from db import get_pool

class DbMiddleware(BaseMiddleware):
    async def __call__(self, handler: Callable, event: TelegramObject, data: dict) -> Any:
        data["pool"] = get_pool()
        return await handler(event, data)

# In main.py:
dp.message.middleware(DbMiddleware())

# Now handlers receive pool directly:
@router.message(Command("profile"))
async def cmd_profile(msg: Message, pool: asyncpg.Pool):
    row = await pool.fetchrow("SELECT * FROM users WHERE id=$1", msg.from_user.id)
    ...