摸一个基于fastapi的飞书bot
查看飞书的介绍文档,有两种机器人,一种是自定义机器人,只能推送消息,没办法和用户互动;应用机器人可以互动,功能也更多一些,不过需要企业的管理员审核。最好的方式是自己的企业创建一个企业应用,功能比个人的应用更多,而且 5 个人一下不用审核。
预计写一个打卡 bot,可以和数据库以及 deepseek 交互。因为涉及数据转换比较复杂,决定使用 python fastapi + sqlite 吧。用 python 写代码比较舒适区内…虽然官方有 sdk 封装好的,但是我寻思着,如果后续有写成前后端的需要的话,单单用 fastapi 不用 sdk 更好兼容一些,所以这次就没有使用官方的 sdk,而是自己封装了 bot 的类。
fastapi
用 fastapi 的体验还是很好,一些数据库的交互、测试等等过程都比较完善而且简洁,虽然在使用的过程中因为版本迭代的原因,有一些用法我感觉还挺好用,但是 deprecated 了()
python 包管理
既然要部署在服务器上,就不能偷懒不用包管理了(),所以这次选了 poetry(本来在 pdm 和 poetry 之间纠结,然后觉得 poetry 更好听就选它了)Poetry。创建程序用 poetry new project_name; poetry install
添加包用 poetry add pkg_name
方便嘞~
数据库交互
fastapi 的文档一看就懂,比如在 fastapi 里面和 sqlite 交互:SQL (Relational) Databases - FastAPI,这里用到了 SQLModel 使用 Python 对象和数据库交互,就不用写 sql 语言啦~~ 支持这些类型:Pydantic field
因为数据里面有日期时间,我为了这些东西怎么 json 序列化反序列化纠结了半天,自动类型转换成 json 是不行的,无法自动序列化 datetime.time
类型,导致返回 Checkin
对象时抛出错误:
TypeError: Object of type Checkin is not JSON serializable
文档写了一种自己定义反序列化、序列化器的方式: Serialization - Pydantic 有一个很好用的就功能删掉了…令人疑惑 Custom json encoder for datetime
后来发现原来不用多麻烦,直接 dump 就可以:
# json to model
mytype_model = Mytype.model_validate(mytype_json)
# model to json
print(mytype_model.model_dump_json())
mark 一个在网页查看 sqlite 数据库内容的网站 SQLite Viewer Web App
项目目录
python 的项目结构管理参考了 项目结构 - Python 项目工程化开发指南 (我终于会写 init 文件了…)最好使用绝对路径,在 init 里面使用相对路径的 import,比如
# ./main.py
from a.b import xxx
# ./a/__init__.py
from .b import xxx
pytest 测试
使用 pytest 来进行测试 pytest documentation,所有的测试文件用 test_
开头,函数也用 test_
开头。
如何测试?这里有一个比较优雅的方式:Test Applications with FastAPI and SQLModel - SQLModel。使用 pytest 的 fixture 可以节省不少代码,比如把测试的数据库安排在内存中,就可以不用总是来回折腾 database 了!
yiled
fastapi 里面用到了很多 yield…带有 yield 的函数在 python 里面是一个 generator,比如说这样可以实现一个斐波那契数列生成器
def fab(max):
n, a, b = 0, 0, 1
while n < max:
yield b
a, b = b, a + b
n = n + 1
for n in fab(5):
print(n)
接上 飞书
fastapi 也讲了 config 怎么配置,Pydantic Settings
可以明确 .env
文件里面的设置的字段类型 Settings and Environment Variables - FastAPI
在服务器里测试要用 uvicorn,因为 fastapi dev 只会在 127.0.0.1
上支持本地的连接
poetry run uvicorn checkbot.app:app --host '0.0.0.0' --port 8000 --reload
接上 ds
如果要进行多轮对话需要拼接历史消息多轮对话 | DeepSeek API Docs,每次调用对话不全的时候,把历史的消息也发送给 ds,因此打算用 dequeue 来记录数据了。