Building a Telegram Bot for Order and Inventory Management

This architecture uses Python, SQLite, and the Telegram Bot API to implement real-time inventory tracking and order handling for a small retail setup.

1. Database Schema Design

The state must be stored relationally. Atomic SQL transactions ensure inventory levels remain accurate during simultaneous purchasing attempts.

Create a database initialization script (setup.py):

import sqlite3

def init_db():
    conn = sqlite3.connect('shop.db')
    cursor = conn.cursor()

    cursor.execute('''
        CREATE TABLE IF NOT EXISTS inventory (
            product_id INTEGER PRIMARY KEY AUTOINCREMENT,
            name TEXT UNIQUE NOT NULL,
            stock INTEGER NOT NULL CHECK(stock >= 0),
            price REAL NOT NULL
        )
    ''')

    cursor.execute('''
        CREATE TABLE IF NOT EXISTS orders (
            order_id INTEGER PRIMARY KEY AUTOINCREMENT,
            user_id INTEGER NOT NULL,
            product_id INTEGER NOT NULL,
            quantity INTEGER NOT NULL,
            status TEXT DEFAULT 'PENDING',
            FOREIGN KEY (product_id) REFERENCES inventory(product_id)
        )
    ''')
    conn.commit()
    conn.close()

if __name__ == '__main__':
    init_db()

2. Telegram Bot Configuration

- Open Telegram and start a chat with **@BotFather**.
- Send `/newbot`, follow the prompts, and record your HTTP API token.
- Install the Python Telegram framework: `pip install pyTelegramBotAPI`.

3. Core Bot Implementation

This script processes catalog requests and handles purchasing logic within database transactions to eliminate race conditions.

import sqlite3
import telebot

TOKEN = 'YOUR_BOT_TOKEN_HERE'
ADMIN_IDS = [123456789]  # Replace with actual Telegram User IDs

bot = telebot.TeleBot(TOKEN)

def get_db():
    return sqlite3.connect('shop.db')

@bot.message_handler(commands=['inventory'])
def list_inventory(message):
    conn = get_db()
    cursor = conn.cursor()
    cursor.execute("SELECT product_id, name, stock, price FROM inventory WHERE stock > 0")
    items = cursor.fetchall()
    conn.close()

    if not items:
        bot.reply_to(message, "Out of stock.")
        return

    text = "Available Products:\n"
    for item in items:
        text += f"ID: {item[0]} | {item[1]} - {item[2]} in stock | ${item[3]:.2f}\n"
    bot.reply_to(message, text)

@bot.message_handler(commands=['order'])
def place_order(message):
    # Syntax: /order  
    args = message.text.split()
    if len(args) != 3:
        bot.reply_to(message, "Usage: /order <product_id> <quantity>")
        return

    product_id, qty = int(args[1]), int(args[2])
    conn = get_db()
    cursor = conn.cursor()

    try:
        cursor.execute("BEGIN TRANSACTION")
        cursor.execute("SELECT stock FROM inventory WHERE product_id = ?", (product_id,))
        row = cursor.fetchone()

        if not row or row[0]   
    try:
        _, name, qty, price = message.text.split()
        conn = get_db()
        cursor = conn.cursor()
        cursor.execute('''
            INSERT INTO inventory (name, stock, price) VALUES (?, ?, ?)
            ON CONFLICT(name) DO UPDATE SET stock = stock + ?
        ''', (name, int(qty), float(price), int(qty)))
        conn.commit()
        conn.close()
        bot.reply_to(message, f"Updated stock for '{name}'.")
    except Exception:
        bot.reply_to(message, "Usage: /addstock <name> <quantity> <price>")

bot.infinity_polling()

4. Operational Deployment Requirements

- **Process Management:** Use `systemd` or Docker to run the script as a daemon, ensuring automatic restart on crashes.
- **Backups:** Schedule periodic backups of `shop.db` using a cron job.
- **Scaling Limit:** SQLite works for low to moderate traffic. Switch to PostgreSQL if handling simultaneous, high-frequency writes.

Need this done fast? order a Telegram bot on FreelanceHunt (https://freelancehunt.com/freelancer/sspoisk.html).