使用 Python 和 Fauna 构建 Telegram 机器人 什么是无服务器数据库？Fauna 简介 Fauna 入门 Fauna 与 Python 集成 使用 Python 构建 Telegram 机器人 向 Fauna 进行查询 结论

2025-06-09

使用 Python 和 Fauna 构建 Telegram 机器人

什么是无服务器数据库？

动物简介

开始与动物相处

将 Fauna 与 Python 集成

使用 Python 构建 Telegram 机器人

向 Fauna 进行查询

结论

不久前，我偶然发现了"Serverless Databases"“Fauna”这个词，并了解了它。出于好奇，我决定尝试一下，它的简洁性和无服务器数据库的整体概念给我留下了深刻的印象。我很喜欢Fauna自带的库，使其与Python编程语言兼容，并且还支持其他几种语言。

本文的目的是向读者介绍 Fauna 的无服务器数据库世界，并通过使用 Python 构建 Telegram Bot 来探索其 CRUD 功能，以帮助跟踪我们的日常任务（或多或少是一个待办事项列表机器人）。

因此，我们将在 Fauna 中创建机器人用户，添加任务、检索任务、更新任务和删除任务。

什么是无服务器数据库？

无服务器数据库是指所有维护和运维责任均由开发人员或应用程序自行承担，所有流量/CRUD 都可通过按需扩展进行处理。这减少了管理数据库资源的负载和工作量，尤其是在应用程序开始扩展以支持大量用户时。

除了减少管理数据库资源的工作量之外，为应用程序选择无服务器数据库还有很多优势。它们包括：

  • 降低数据库运营成本
  • 实时数据库访问
  • 提高生产力（因为消除了数据库管理的负担）
  • 无限可扩展性
  • 提高数据库安全性

动物简介

当然，我们现在知道 Fauna 是一个无服务器数据库，但它是如何存储和处理数据的呢？Fauna 的核心是一个文档数据库，提供两种接口：GraphQL 和 Fauna 查询语言 (FQL)。数据库可以存储集合、索引，甚至其他数据库（多租户！）。集合内包含文档，默认情况下，文档没有严格的模式要求。Fauna 能够处理各种数据类型（例如时间数据），但尤其令人印象深刻的是它对关系数据提供了一流的支持。我们稍后会一起探讨其中的大部分内容，但如果您想了解更多信息，请查看Fauna 文档！

开始与动物相处

使用 Fauna 的第一步是在官方网站上创建一个账户。您可以使用您的电子邮件地址、GitHub 或 Netlify 账户进行创建：https ://dashboard.fauna.com/accounts/register

创建动物数据库

我们首先要在 Fauna 上创建一个实际的数据库，应用程序将与该数据库交互，并将数据存储在其中。为此，请单击按钮“NEW DATABASE”，您将看到类似下面的屏幕。

创建数据库映像

在此屏幕上，系统会询问您数据库名称，是否要预填充数据库，如果您要从其他数据库迁移到 Fauna，则会提供提示。现在，我们只需输入数据库名称并按下“SAVE”按钮即可。然后，您将看到如下所示的屏幕。

数据库概览图

创建动物收藏

Fauna 将其数据组织成集合，并使用索引来浏览数据。集合类似于包含具有类似特征的数据的 SQL 表，例如，users 集合包含数据库中用户的信息。在开始将实际数据存储到 Fauna 数据库之前，我们需要先创建一个集合。创建过程通过点击“NEW COLLECTION”按钮完成。

创建收藏图像

然后，您将看到一个表单，用于填写该集合的详细信息。系统会要求您输入集合名称（我们将“users”在本例中使用），然后会要求您输入History Days和TTL。“历史天数”用于定义 Fauna 应保留该特定集合中任何数据的历史记录的天数，而 TTL 则表示该集合中数据的到期日期。例如，如果 TTL 设置为 7，则该集合中存储的任何数据都将在其上次修改日期后 7 天自动删除。

用户收藏图片

输入集合名称并点击“保存”按钮后，您将看到类似上述屏幕。在 Fauna 中，集合中的数据称为文档。文档类似于基于 SQL 的数据库中的表行。

创建动物群索引

我们还需要创建一个索引，以便我们浏览数据库中存储的数据。要创建索引，请导航到 Fauna 侧边栏（屏幕左侧）上的“数据库概览”选项卡，然后单击“NEW INDEX”按钮。

创建索引图像

您将看到一个表单，其中会要求您选择要链接到索引的集合并提供名称。此外，请指定Terms索引的 ，这是索引允许浏览的特定数据。在本例中，我们将使用“id”。填写完成后，点击“SAVE”按钮。

将 Fauna 与 Python 集成

创建 Fauna 数据库 API 密钥

在开始构建使用 Fauna 的 Python 应用之前，我们需要创建一个 API 密钥，以便我们的应用能够轻松地与数据库通信。要创建 API 密钥，我们需要导航到 Fauna 侧边栏（位于屏幕左侧）的“安全”选项卡。

创建关键图像

您将看到一个类似于上面的屏幕，单击“NEW KEY”按钮继续生成密钥。

分配键图像

然后，系统会要求您提供密钥关联的数据库、密钥的角色以及 API 密钥的可选名称。填写完这些信息后，点击“SAVE”按钮。

密钥创建的图像

完成此操作后，您将获得 API 密钥（出于隐私原因，此处隐藏）。密钥生成后应立即复制，并存储在易于检索的位置。

使用 Python 与动物群沟通

现在，我们要安装 Fauna 的 Python 驱动程序。幸运的是，它可以通过 Python 包管理器 (pip) 获得，这为我们省去了通过 Github 仓库构建的麻烦。

$ pip install faunadb
Enter fullscreen mode Exit fullscreen mode

安装完成后，我们将运行 Fauna Python 驱动程序文档https://docs.fauna.com/fauna/current/drivers/python.html中提供的示例代码

from faunadb import query as q
from faunadb.objects import Ref
from faunadb.client import FaunaClient

client = FaunaClient(secret="your-secret-here")

indexes = client.query(q.paginate(q.indexes()))

print(indexes)
Enter fullscreen mode Exit fullscreen mode

使用 Python 构建 Telegram 机器人

什么是 Telegram？

Telegram 是顶级的社交媒体消息平台之一。其核心功能包括向其他 Telegram 用户发送消息、创建群聊、呼叫联系人以及发送文件和贴纸。您可以访问https://telegram.org了解更多关于 Telegram 的信息。

什么是 Telegram 机器人？

Telegram 机器人是由软件而非真人操作的账户。它们可以做很多事情，比如教学、玩游戏、充当搜索引擎、广播消息、提醒用户、连接和集成其他服务，甚至向物联网 (IOT) 设备发送命令。@GroupButler_bot是一款非常流行的 Telegram 机器人，用于管理 Telegram 群组。

群组管家机器人

入门

步骤 1：与 BotFather 对话

登录你的 Telegram 账户，搜索@botfather并开始与该账户对话。PS：BotFather 也是一个 Telegram 机器人

机器人父亲搜索
机器人父亲开始

步骤2：使用BotFather创建Telegram Bot界面

现在我们将使用该/newbot命令创建一个新的 Telegram 机器人。使用 BotFather 创建 Telegram 机器人意味着您需要为其命名并分配用户名。

创建新机器人
提交机器人信息

在为您的 Telegram Bot 提供名称和用户名后，BotFather 将为您提供 API 令牌，该令牌将用于通过 Telegram API 与 Bot 帐户进行交互。

获取机器人令牌

步骤3：用Python驱动我们的机器人

我们要做的第一件事是安装构建 Telegram 机器人所需的 Python 库。当然，我们可以直接按照官方文档使用 Telegram API ，但 Python 已经提供了一些易于集成的库，简化了这一过程。

$ pip install telegram
$ pip install python_telegram_bot
Enter fullscreen mode Exit fullscreen mode

我们接下来要做的是编写一个脚本来接收发送给我们机器人的任何消息。

import telegram
from telegram.ext import Updater
from telegram.ext import CommandHandler

telegram_bot_token = "your-telegram-token"

updater = Updater(token=telegram_bot_token, use_context=True)
dispatcher = updater.dispatcher


def start(update, context):
    print("Hello World!")


dispatcher.add_handler(CommandHandler("start", start))
updater.start_polling()
Enter fullscreen mode Exit fullscreen mode

另外，我们需要使用 telegram python 模块为我们的 Telegram 机器人创建一个更新器和调度器。更新器类跟踪并监控发送给机器人的消息，并将其传递给调度器，而调度器则负责处理接收到的消息。

接下来我们需要为调度程序设置一个命令处理程序。命令处理程序是一个函数，当机器人使用某个命令时会调用它。Telegram Bot 命令可以是任何以 开头的消息，“/”例如：/start /help /info。还有其他类型的处理程序可以添加到调度程序中，例如图像处理程序、文本处理程序、正则表达式处理程序等等。我们从命令处理程序开始，因为这是每个用户进入 Telegram Bot 的第一件事。

当命令处理程序被触发时，更新程序会传递两个参数 (update, context)，其中包含有关消息请求的所有信息。我们希望启动命令的处理程序“Hello World!”在每次触发时都能在控制台中打印文本。

设置处理程序后，我们将启动更新程序，开始监控来自 Telegram 的消息，并使用该start_polling()方法触发调度程序。让我们将文件另存为，bot.py然后开始监控来自 Telegram 的消息。

$ python bot.py
Enter fullscreen mode Exit fullscreen mode

运行脚本后，让我们向机器人发送一条消息，看看“Hello World!”控制台中是否打印了 。
如果您还没有这样做，我们需要使用之前提供给 BotFather 的用户名来搜索我们的 Telegram 机器人。

寻找我们的机器人
启动我们的机器人
收到问候消息

我们的命令处理程序已成功触发，并且我们收到了问候消息，让我们稍微调整一下，向用户发送文本“Hello World”而不是在我们的控制台上打印。

def start(update, context):
   chat_id = update.effective_chat.id
   context.bot.send_message(chat_id=chat_id, text="Hello World")
Enter fullscreen mode Exit fullscreen mode

要将消息从我们的机器人发送回用户，我们需要使用类send_message的方法，context.bot该方法接受一个名为的参数chat_id。Telegram 为每个可以通过名称访问的用户分配一个唯一标识符，chat_id机器人可以使用它来发送消息或进行引用。

当我们再次运行脚本并进行更新并再次发送开始消息时，我们应该得到类似于下图的响应。

发送机器人消息

向 Fauna 进行查询

使用 Fauna 保存数据

现在让我们继续将我们的 Telegram Bot 与 Fauna 集成，我们要尝试的第一件事是保存启动我们的 Telegram Bot 的用户的详细信息。

接下来我们要做的是导入 Fauna 运行所需的必要库并与我们现有的代码合并。

import telegram
from telegram.ext import Updater
from telegram.ext import CommandHandler
from faunadb import query as q
from faunadb.objects import Ref
from faunadb.client import FaunaClient

telegram_bot_token = "your-telegram-token"
fauna_secret = "your-secret-here"

updater = Updater(token=telegram_bot_token, use_context=True)
dispatcher = updater.dispatcher
client = FaunaClient(secret=fauna_secret)


def start(update, context):
    chat_id = update.effective_chat.id
    context.bot.send_message(chat_id=chat_id, text="Hello World")


dispatcher.add_handler(CommandHandler("start", start))
updater.start_polling()
Enter fullscreen mode Exit fullscreen mode

接下来我们要做的是存储与 Telegram Bot 开始对话的用户的所有必要信息。

import pytz
from datetime import datetime
Enter fullscreen mode Exit fullscreen mode

我们将导入datetime和pytz库，以便我们可以将时间戳保存到我们之前创建的数据库中，因为 Fauna 只接受偏移感知日期时间。

def start(update, context):
    chat_id = update.effective_chat.id
    first_name = update["message"]["chat"]["first_name"]
    username = update["message"]["chat"]["username"]

    user = client.query(q.create(q.collection("users"), {
        "data": {
            "id": chat_id,
            "first_name": first_name,
            "username": username,
            "last_command": "",
            "date": datetime.now(pytz.UTC)
        }
    }))
    context.bot.send_message(chat_id=chat_id, text="Welcome to Fauna TO-DO, your details have been saved 😊")
Enter fullscreen mode Exit fullscreen mode

接下来，我们对启动处理程序进行一些更改。一旦触发，我们将首先提取用户的名字和 Telegram 用户名。然后，我们查询之前创建的用户集合并保存用户的详细信息。最后，向用户发送一条欢迎消息。

我们还应该在用户注册中添加一些检查，这样就不会重复注册同一个用户。在保存用户详细信息之前，我们应该检查他们是否已经在数据库中注册过，只有未注册过才保存。

def start(update, context):
    chat_id = update.effective_chat.id
    first_name = update["message"]["chat"]["first_name"]
    username = update["message"]["chat"]["username"]

    try:
        client.query(q.get(q.match(q.index("users"), chat_id)))
    except:
        user = client.query(q.create(q.collection("users"), {
            "data": {
                "id": chat_id,
                "first_name": first_name,
                "username": username,
                "last_command": "",
                "date": datetime.now(pytz.UTC)
            }
        }))
    context.bot.send_message(chat_id=chat_id, text="Welcome to Fauna TO-DO, your details have been saved 😊")
Enter fullscreen mode Exit fullscreen mode

我们更新了启动命令处理程序，首先查询数据库以确定用户是否存在；如果存在，则立即发送问候消息；如果不存在，则保存用户并发送问候消息。接下来，我们添加一个命令，用于将项目添加到待办事项列表中。

在编写代码之前，我们需要创建一个索引来浏览我们之前创建的集合中的数据。

创建索引

完成此步骤后，让我们编写一个新的命令处理程序来添加待办任务并将其包含在我们的调度程序中。

def add_todo(update, context):
    chat_id = update.effective_chat.id

    user = client.query(q.get(q.match(q.index("users"), chat_id)))
    client.query(q.update(q.ref(q.collection("users"), user["ref"].id()), {"data": {"last_command": "add_todo"}}))
    context.bot.send_message(chat_id=chat_id, text="Enter the todo task you want to add 😁")


dispatcher.add_handler(CommandHandler("add_todo", add_todo))
Enter fullscreen mode Exit fullscreen mode

我在这里所做的是设置值，last_command然后add_todo向用户发送一条消息，告诉他们写下他们想要添加的内容。

添加待办事项图片

接下来我们需要为机器人添加一个消息处理器。消息处理器是一种 Telegram 调度程序，用于处理基于文本的消息。要创建消息处理器，我们需要导入所需的类。

from telegram.ext import Updater
from telegram.ext import CommandHandler
from telegram.ext import MessageHandler, Filters
Enter fullscreen mode Exit fullscreen mode

然后我们定义将待办任务保存到 Fauna 中的实际消息处理程序。

def echo(update, context):
    chat_id = update.effective_chat.id
    message = update.message.text

    user = client.query(q.get(q.match(q.index("users"), chat_id)))
    last_command = user["data"]["last_command"]

    if last_command == "add_todo":
        todo = client.query(q.create(q.collection("todo"), {
            "data": {
                "user_id": chat_id,
                "todo": message,
                "completed": False,
                "date": datetime.now(pytz.UTC)
            }
        }))
        client.query(q.update(q.ref(q.collection("users"), user["ref"].id()), {"data": {"last_command": ""}}))
        context.bot.send_message(chat_id=chat_id, text="Successfully added todo task 👍")
Enter fullscreen mode Exit fullscreen mode

我们不要忘记将消息处理程序也添加到调度程序中。

dispatcher.add_handler(MessageHandler(Filters.text, echo))
updater.start_polling()
Enter fullscreen mode Exit fullscreen mode

重启机器人并发送消息后，它应该会回复我一条成功消息。我们还可以从 Fauna 仪表板监控数据库。

添加了待办任务
在仪表板中查看数据

从 Fauna 读取数据

现在我们应该添加一个命令来检索所有待办任务及其状态并显示给用户。

def list_todo(update, context):
   chat_id = update.effective_chat.id

   task_message = ""
   tasks = client.query(q.paginate(q.match(q.index("todo"), chat_id)))
   for i in tasks["data"]:
       task = client.query(q.get(q.ref(q.collection("todo"), i.id())))
       if task["data"]["completed"]:
           task_status = "Completed"
       else:
           task_status = "Not Completed"
       task_message += "{}\nStatus: {}\n\n".format(task["data"]["todo"], task_status)
   if task_message == "":
       task_message = "You have not added any task, do that with /add_todo 😇"
   context.bot.send_message(chat_id=chat_id, text=task_message)
Enter fullscreen mode Exit fullscreen mode

然后将其添加到我们的调度程序

dispatcher.add_handler(CommandHandler("list_todo", list_todo))
Enter fullscreen mode Exit fullscreen mode

重新启动我们的脚本并/list_todo在 Telegram 上触发命令后，我们应该得到类似下图的响应。

列出任务图像

更新 Fauna 中的数据

现在是时候添加一个更新待办事项数据的函数了。在本例中，更新是指将任务状态从未完成更改为已完成，反之亦然。我们将使用待办事项的参考 ID 来跟踪要更新的任务。为了方便用户操作，我们创建一个包含在/list_todo消息中的命令文本。

def list_todo(update, context):
   chat_id = update.effective_chat.id

   task_message = ""
   tasks = client.query(q.paginate(q.match(q.index("todo"), chat_id)))
   for i in tasks["data"]:
       task = client.query(q.get(q.ref(q.collection("todo"), i.id())))
       if task["data"]["completed"]:
           task_status = "Completed"
       else:
           task_status = "Not Completed"
       task_message += "{}\nStatus: {}\nUpdate Link: /update_{}\n\n".format(task["data"]["todo"], task_status, i.id())
   if task_message == "":
       task_message = "You have not added any task, do that with /add_todo 😇"
   context.bot.send_message(chat_id=chat_id, text=task_message)
Enter fullscreen mode Exit fullscreen mode

现在，当我们触发/list_todo命令时，我们会得到类似下图的响应。

添加更新

在此之后，我们需要创建一个名为的函数，update_todo该函数将使用正则表达式查找以开头的任何文本，/update_{todo ID}然后更新它。

dispatcher.add_handler(MessageHandler(Filters.regex("/update_[0-9]*"), update_todo))
dispatcher.add_handler(MessageHandler(Filters.text, echo))
Enter fullscreen mode Exit fullscreen mode

我们使用了与之前相同的消息处理程序，但这次使用了正则表达式处理程序。/update_[0-9]*如果字符串以 开头，/update_后跟任意给定长度的数字，则正则表达式字符串将返回 true。附言：我注意到，如果将正则表达式处理程序放在接受文本过滤器的 echo 处理程序下方，它将不会被触发。

def update_todo(update, context):
   chat_id = update.effective_chat.id
   message = update.message.text
   todo_id = message.split("_")[1]

   task = client.query(q.get(q.ref(q.collection("todo"), todo_id)))
   if task["data"]["completed"]:
       new_status = False
   else:
       new_status = True
   client.query(q.update(q.ref(q.collection("todo"), todo_id), {"data": {"completed": new_status}}))
   context.bot.send_message(chat_id=chat_id, text="Successfully updated todo task status 👌")
Enter fullscreen mode Exit fullscreen mode

在 update_todo 函数中，我们首先提取用户的 chat_id 和他们想要更新的任务的 todo_id，然后检查任务的状态，如果未完成则将其更改为已完成，反之亦然。完整的用户流程如下图所示。

更新待办任务

从 Fauna 中删除数据

我们还需要从数据库中删除任务的功能，毕竟，没有 D 的 CRUD 应用又算什么呢？Fauna 提供了这方面的 API，让我们直接开始写代码吧。我们也将使用与更新函数相同的样式来执行删除操作。

task_message += "{}\nStatus: {}\nUpdate Link: /update_{}\nDelete Link: /delete_{}\n\n".format(task["data"]["todo"], task_status, i.id(), i.id())
Enter fullscreen mode Exit fullscreen mode

首先，我们将删除命令添加到我们的 list_todo 渲染中，以便我们得到类似于下面的响应。

新增删除任务

/delete_然后，我们定义一个正则表达式处理程序来跟踪以待办事项 ID开头并后跟该 ID 的消息。别忘了把它放在 echo 处理程序的上方。

dispatcher.add_handler(MessageHandler(Filters.regex("/delete_[0-9]*"), delete_todo))
dispatcher.add_handler(MessageHandler(Filters.text, echo))
Enter fullscreen mode Exit fullscreen mode

然后我们创建delete_todo与更新具有相同风格但具有删除功能的函数。

def delete_todo(update, context):
   chat_id = update.effective_chat.id
   message = update.message.text
   todo_id = message.split("_")[1]

   client.query(q.delete(q.ref(q.collection("todo"), todo_id)))
   context.bot.send_message(chat_id=chat_id, text="Successfully deleted todo task status 👌")
Enter fullscreen mode Exit fullscreen mode

完成删除任务

现在，我们的待办事项应用已经以 Python 和 Fauna 编写的 Telegram 机器人的形式完成了。整个应用的用户流程如下图所示。我还对机器人消息做了一些修改，Python 源代码可在教程末尾找到。

完整的机器人用户界面

结论

通过将 Fauna 与 Python 集成，我们能够使用 Telegram Bot API 轻松构建一个待办事项应用。我们还体验了使用 Fauna 创建和管理无服务器数据库是多么简单。

该应用的完整源代码可以在 GitHub 上找到。试用 Fauna 的过程非常有趣，我迫不及待地想看到你用它构建的精彩作品。如有任何疑问，欢迎在 Twitter 上联系我：@LordGhostX

鏂囩珷鏉ユ簮锛�https://dev.to/lordghostx/building-a-telegram-bot-with-python-and-fauna-494i
Devto 翻译
PREV
是否需要 Codar Fofo？ GenAI 现场直播！ | 2025 年 6 月 4 日
NEXT
网络安全简介：什么是网络安全？网络威胁？网络攻击的形式 网络攻击的影响 网络安全领域 黑客分类 红军 vs. 蓝军 渗透测试概述 渗透测试阶段 网络安全路线图 结论