基于aiogram的Telegram机器人：PostgreSQL图片存显实现问询

场景说明 #

正在用aiogram开发Telegram机器人，采用PostgreSQL存储产品ID、图片、名称、价格及描述信息。现有两个核心需求：

1. 已将图片URL存入数据库，需要在机器人消息中直接显示图片而非纯文本URL；
2. 希望实现直接将图片二进制数据存入数据库（必须存储在库内），并在消息中正常显示。

以下是当前使用的数据库操作代码与回调处理代码：

原数据库操作代码（BD1类） #

import psycopg2
import asyncpg
import asyncio
import aiogram
from aiogram import types


class BD1:
    def ss():
        try:
            conn = psycopg2.connect(host="localhost",
                                    port=5432,
                                    database="postgres",
                                    user="postgres",
                                    password="135")
            cur = conn.cursor()
            print("Database opened successfully")

            cur.execute("""SELECT * FROM perfumes""")
            query_results = cur.fetchall()
            text = '\n\n'.join(['\n\n'.join(map(str, x)) for x in query_results])
            return (str(text))

        except psycopg2.Error as e:
            print("Error connecting to database:", e)
        finally:
            cur.close()
            conn.close()

原回调处理代码（handlers.py） #

@router.callback_query()
async def callback_keyboard(callback: CallbackQuery):
    if callback.data == 'teas':
        await callback.answer()
        await callback.message.answer(BD1.ss())

解决方案一：存储图片URL并直接显示图片 #

如果已将图片URL存在数据库photo字段中，无需将所有结果拼接成纯文本发送，可遍历查询结果，逐个调用send_photo接口发送图片，同时附带产品信息作为图片说明。

修改后的数据库操作代码（适配异步环境） #

import asyncpg
from aiogram import types

class BD1:
    @staticmethod
    async def get_perfumes():
        conn = None
        try:
            # 使用asyncpg异步连接PostgreSQL，适配aiogram的异步运行环境
            conn = await asyncpg.connect(
                host="localhost",
                port=5432,
                database="postgres",
                user="postgres",
                password="135"
            )
            # 查询产品核心字段：ID、图片URL、名称、价格、描述
            records = await conn.fetch("SELECT id, photo, name, price, description FROM perfumes")
            return records
        except asyncpg.Error as e:
            print(f"数据库连接错误: {e}")
            return []
        finally:
            if conn:
                await conn.close()

修改后的回调处理代码 #

@router.callback_query()
async def callback_keyboard(callback: CallbackQuery):
    if callback.data == 'teas':
        await callback.answer()
        # 获取产品列表
        perfumes = await BD1.get_perfumes()
        for perfume in perfumes:
            # 提取字段信息
            perfume_id = perfume['id']
            photo_url = perfume['photo']
            name = perfume['name']
            price = perfume['price']
            description = perfume['description']
            # 构造图片说明文本
            caption = f"ID: {perfume_id}\n名称: {name}\n价格: {price}\n描述: {description}"
            # 直接传入URL发送图片，Telegram会自动解析并显示图片
            await callback.message.photo(photo=photo_url, caption=caption)

解决方案二：直接将图片二进制数据存入数据库 #

若必须将图片存储在数据库内，可利用PostgreSQL的bytea类型存储图片二进制数据，读取时将二进制数据转为InputFile对象发送。

1. 数据库表结构调整 #

确保perfumes表的photo字段类型为bytea：

ALTER TABLE perfumes ALTER COLUMN photo TYPE bytea;

2. 图片存入数据库的示例代码（上传场景） #

async def save_photo_to_db(perfume_id, photo_path):
    conn = await asyncpg.connect(
        host="localhost",
        port=5432,
        database="postgres",
        user="postgres",
        password="135"
    )
    try:
        # 读取本地图片的二进制数据
        with open(photo_path, "rb") as f:
            photo_data = f.read()
        # 将二进制数据存入数据库
        await conn.execute(
            "UPDATE perfumes SET photo = $1 WHERE id = $2",
            photo_data, perfume_id
        )
    except asyncpg.Error as e:
        print(f"保存图片错误: {e}")
    finally:
        await conn.close()

3. 修改后的查询与显示代码 #

import asyncpg
from aiogram import types
from io import BytesIO

class BD1:
    @staticmethod
    async def get_perfumes():
        conn = None
        try:
            conn = await asyncpg.connect(
                host="localhost",
                port=5432,
                database="postgres",
                user="postgres",
                password="135"
            )
            records = await conn.fetch("SELECT id, photo, name, price, description FROM perfumes")
            return records
        except asyncpg.Error as e:
            print(f"数据库连接错误: {e}")
            return []
        finally:
            if conn:
                await conn.close()

修改后的回调处理代码 #

@router.callback_query()
async def callback_keyboard(callback: CallbackQuery):
    if callback.data == 'teas':
        await callback.answer()
        perfumes = await BD1.get_perfumes()
        for perfume in perfumes:
            perfume_id = perfume['id']
            # 取出数据库中的二进制图片数据
            photo_data = perfume['photo']
            name = perfume['name']
            price = perfume['price']
            description = perfume['description']
            caption = f"ID: {perfume_id}\n名称: {name}\n价格: {price}\n描述: {description}"
            
            # 将二进制数据转为字节流，再用InputFile包装
            photo_stream = BytesIO(photo_data)
            photo = types.InputFile(photo_stream, filename=f"{name}.jpg")
            
            # 发送图片
            await callback.message.photo(photo=photo, caption=caption)

关键注意事项 #

• 优先使用asyncpg替代psycopg2，因为aiogram是异步框架，同步数据库操作会阻塞事件循环；
• 直接存储二进制图片会显著增大数据库体积，需根据实际业务权衡存储成本；
• 发送图片时，InputFile可直接接受BytesIO字节流，无需将图片保存到本地文件。

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