Telegram机器人重复触发/start时出现PostgreSQL语法错误

问题描述 #

我基于Aiogram框架开发了Telegram机器人，采用PostgreSQL数据库搭配Django管理面板。用户首次发送/start命令时，机器人可正常将用户信息存入数据库；但已入库用户再次发送/start命令时，会触发错误：asyncpg.exceptions.PostgresSyntaxError: syntax error at or near "="。

相关代码 #

postgresql.py

from typing import Union

import asyncpg
from asyncpg import Connection
from asyncpg.pool import Pool

from data import config


class Database:
    def __init__(self):
        self.pool: Union[Pool, None] = None

    async def create(self):
        self.pool = await asyncpg.create_pool(
            user=config.DB_USER,
            password=config.DB_PASS,
            host=config.DB_HOST,
            database=config.DB_NAME
        )

    async def execute(self, command, *args,
                      fetch: bool = False,
                      fetchval: bool = False,
                      fetchrow: bool = False,
                      execute: bool = False
                      ):
        async with self.pool.acquire() as connection:
            connection: Connection
            async with connection.transaction():
                if fetch:
                    result = await connection.fetch(command, *args)
                elif fetchval:
                    result = await connection.fetchval(command, *args)
                elif fetchrow:
                    result = await connection.fetchrow(command, *args)
                elif execute:
                    result = await connection.execute(command, *args)
            return result


    @staticmethod
    def format_args(sql, parameters: dict):
        sql += " AND ".join([
            f"{item} = ${num}" for num, item in enumerate(parameters.keys(),
                                                          start=1)
        ])
        return sql, tuple(parameters.values())

    async def add_user(self, full_name, username, telegram_id):
        sql = "INSERT INTO usersinfo_user (full_name, username, telegram_id) VALUES($1, $2, $3) returning *"
        return await self.execute(sql, full_name, username, telegram_id, fetchrow=True)

    async def select_all_users(self):
        sql = "SELECT * FROM usersinfo_user"
        return await self.execute(sql, fetch=True)

    async def select_user(self, **kwargs):
        sql = "SELECT * FROM usersinfo_user WHERE"
        sql, parameters = self.format_args(sql, parameters=kwargs)
        return await self.execute(sql, *parameters, fetchrow=True)

    async def count_users(self):
        sql = "SELECT COUNT(*) FROM usersinfo_user"
        return await self.execute(sql, fetchval=True)

    async def update_user_username(self, username, telegram_id):
        sql = "UPDATE usersinfo_user SET username=$1 WHERE telegram_id=$2"
        return await self.execute(sql, username, telegram_id, execute=True)


    async def delete_users(self):
        await self.execute("DELETE FROM usersinfo_user WHERE TRUE", execute=True)

    async def drop_users(self):
        await self.execute("DROP TABLE usersinfo_user", execute=True)

start.py

import logging

import asyncpg.exceptions
from aiogram import types
from aiogram.dispatcher.filters.builtin import CommandStart

from data.config import ADMINS
from keyboards.inline.languageKeyboard import choose_language
from states.intro import IntroState
from loader import dp, db, bot


@dp.message_handler(CommandStart(), state=None)
async def bot_start(message: types.Message):
    try:
        user = await db.add_user(telegram_id=message.from_user.id,
                                 full_name=message.from_user.full_name,
                                 username=message.from_user.username)
    except asyncpg.exceptions.UniqueViolationError:
        user = await db.select_user(telegram_id=message.from_user.id)

    await message.answer("Choose language:", reply_markup=choose_language)
    await IntroState.language.set()

    logging.info(message)
    # Notify admin about added usr
    count = await db.count_users()
    msg = f"User<b>{user[1]}</b> added.
 All users:<b>{count}</b> ."
    await bot.send_message(chat_id=ADMINS[0], text=msg)

app.py

from aiogram import executor

from loader import dp, db
import middlewares, filters, handlers
from utils.notify_admins import on_startup_notify
from utils.set_bot_commands import set_default_commands


async def on_startup(dispatcher):
    await db.create()

    # Commands
    await set_default_commands(dispatcher)

    # Notify admin
    await on_startup_notify(dispatcher)


if __name__ == '__main__':
    executor.start_polling(dp, on_startup=on_startup)

错误原因 #

错误根源在Database类的SQL拼接逻辑：
当调用select_user(telegram_id=xxx)时，select_user方法的初始SQL为"SELECT * FROM usersinfo_user WHERE"，随后format_args方法直接将条件"telegram_id = $1"拼接到该SQL后，最终生成的无效SQL是：
SELECT * FROM usersinfo_user WHEREtelegram_id = $1
这里WHERE和字段名之间缺少空格，导致PostgreSQL无法解析语法。

修复方案 #

修改format_args静态方法，在拼接条件前添加空格，同时处理空参数的情况（避免后续其他场景出错）：

@staticmethod
def format_args(sql, parameters: dict):
    if not parameters:
        return sql, tuple()
    # 在条件前添加空格，避免与WHERE连写导致语法错误
    sql += " " + " AND ".join([
        f"{item} = ${num}" for num, item in enumerate(parameters.keys(),
                                                      start=1)
    ])
    return sql, tuple(parameters.values())

另外可以顺便修复start.py中消息格式的问题（原代码换行符使用不当，可能导致消息发送失败）：

msg = f"User<b>{user[1]}</b> added.\nAll users:<b>{count}</b>."

内容的提问来源于stack exchange，提问作者John Olso