使用 peewee 连接 TiDB
TiDB 是一个兼容 MySQL 的数据库,peewee 是 Python 中流行的对象关系映射(ORM)工具。
在本教程中,你可以学习如何使用 TiDB 和 peewee 完成以下任务:
- 搭建你的开发环境。
- 使用 peewee 连接到你的 TiDB 集群。
- 构建并运行你的应用程序。你还可以找到基本 CRUD 操作的示例代码片段。
前置条件
完成本教程,你需要:
- Python 3.8 或更高版本。
- Git。
- 一个 TiDB 集群。
如果你还没有 TiDB 集群,可以按如下方式创建:
- (推荐)参考创建 TiDB Cloud Serverless 集群来创建你自己的 TiDB Cloud 集群。
- 参考部署本地测试 TiDB 集群或部署生产环境 TiDB 集群来创建本地集群。
运行示例应用并连接 TiDB
本节演示如何运行示例应用代码并连接到 TiDB。
第 1 步:克隆示例应用仓库
在终端窗口中运行以下命令,克隆示例代码仓库:
git clone https://github.com/tidb-samples/tidb-python-peewee-quickstart.git
cd tidb-python-peewee-quickstart
第 2 步:安装依赖
运行以下命令,安装示例应用所需的依赖包(包括 peewee 和 PyMySQL):
pip install -r requirements.txt
为什么要使用 PyMySQL?
peewee 是一个支持多种数据库的 ORM 库。它为数据库操作提供了高级抽象,帮助开发者以更面向对象的方式编写 SQL 语句。然而,peewee 并不自带数据库驱动。要连接数据库,你需要安装相应的数据库驱动。本示例应用使用 PyMySQL 作为数据库驱动,它是一个纯 Python 实现的 MySQL 客户端库,兼容 TiDB,并可在所有平台上安装。更多信息请参考 peewee 官方文档。
第 3 步:配置连接信息
根据你选择的 TiDB 部署方式,连接到你的 TiDB 集群。
进入 Clusters 页面,点击目标集群名称进入集群概览页。
点击右上角的 Connect,弹出连接对话框。
确保连接对话框中的配置与你的操作环境一致。
- Connection Type 设置为
Public
- Branch 设置为
main
- Connect With 设置为
General
- Operating System 与你的环境一致
- Connection Type 设置为
点击 Generate Password 生成随机密码。
运行以下命令,复制
.env.example
并重命名为.env
:cp .env.example .env将对应的连接字符串复制粘贴到
.env
文件中。示例结果如下:TIDB_HOST='{host}' # e.g. gateway01.ap-northeast-1.prod.aws.tidbcloud.com TIDB_PORT='4000' TIDB_USER='{user}' # e.g. xxxxxx.root TIDB_PASSWORD='{password}' TIDB_DB_NAME='test' CA_PATH='{ssl_ca}' # e.g. /etc/ssl/certs/ca-certificates.crt (Debian / Ubuntu / Arch)请务必将
{}
占位符替换为连接对话框中获取的连接参数。保存
.env
文件。
进入 Clusters 页面,点击目标集群名称进入集群概览页。
点击右上角的 Connect,弹出连接对话框。
在连接对话框中,从 Connection Type 下拉列表选择 Public,然后点击 CA cert 下载 CA 证书。
如果你还未配置 IP 访问列表,请点击 Configure IP Access List,或参考 Configure an IP Access List 进行配置后再首次连接。
除了 Public 连接类型,TiDB Cloud Dedicated 还支持 Private Endpoint 和 VPC Peering 连接类型。更多信息请参见 Connect to Your TiDB Cloud Dedicated Cluster。
运行以下命令,复制
.env.example
并重命名为.env
:cp .env.example .env将对应的连接字符串复制粘贴到
.env
文件中。示例结果如下:TIDB_HOST='{host}' # e.g. tidb.xxxx.clusters.tidb-cloud.com TIDB_PORT='4000' TIDB_USER='{user}' # e.g. root TIDB_PASSWORD='{password}' TIDB_DB_NAME='test' CA_PATH='{your-downloaded-ca-path}'请务必将
{}
占位符替换为连接对话框中获取的连接参数,并将CA_PATH
配置为上一步下载的证书路径。保存
.env
文件。
运行以下命令,复制
.env.example
并重命名为.env
:cp .env.example .env将对应的连接字符串复制粘贴到
.env
文件中。示例结果如下:TIDB_HOST='{tidb_server_host}' TIDB_PORT='4000' TIDB_USER='root' TIDB_PASSWORD='{password}' TIDB_DB_NAME='test'请务必将
{}
占位符替换为连接参数,并删除CA_PATH
这一行。如果你在本地运行 TiDB,默认主机地址为127.0.0.1
,密码为空。保存
.env
文件。
第 4 步:运行代码并检查结果
执行以下命令运行示例代码:
python peewee_example.py检查 Expected-Output.txt 文件,确认输出是否一致。
示例代码片段
你可以参考以下示例代码片段,完成你自己的应用开发。
完整示例代码及运行方法请参考 tidb-samples/tidb-python-peewee-quickstart 仓库。
连接 TiDB
from peewee import MySQLDatabase
def get_db_engine():
config = Config()
connect_params = {}
if ${ca_path}:
connect_params = {
"ssl_verify_cert": True,
"ssl_verify_identity": True,
"ssl_ca": ${ca_path},
}
return MySQLDatabase(
${tidb_db_name},
host=${tidb_host},
port=${tidb_port},
user=${tidb_user},
password=${tidb_password},
**connect_params,
)
使用该函数时,你需要将 ${tidb_host}
、${tidb_port}
、${tidb_user}
、${tidb_password}
、${tidb_db_name}
和 ${ca_path}
替换为你 TiDB 集群的实际值。
定义数据表
from peewee import Model, CharField, IntegerField
db = get_db_engine()
class BaseModel(Model):
class Meta:
database = db
class Player(BaseModel):
name = CharField(max_length=32, unique=True)
coins = IntegerField(default=0)
goods = IntegerField(default=0)
class Meta:
table_name = "players"
更多信息请参考 peewee 文档:Models and Fields。
插入数据
# Insert a single record
Player.create(name="test", coins=100, goods=100)
# Insert multiple records
Player.insert_many(
[
{"name": "test1", "coins": 100, "goods": 100},
{"name": "test2", "coins": 100, "goods": 100},
]
).execute()
更多信息请参考 插入数据。
查询数据
# Query all records
players = Player.select()
# Query a single record
player = Player.get(Player.name == "test")
# Query multiple records
players = Player.select().where(Player.coins == 100)
更多信息请参考 查询数据。
更新数据
# Update a single record
player = Player.get(Player.name == "test")
player.coins = 200
player.save()
# Update multiple records
Player.update(coins=200).where(Player.coins == 100).execute()
更多信息请参考 更新数据。
删除数据
# Delete a single record
player = Player.get(Player.name == "test")
player.delete_instance()
# Delete multiple records
Player.delete().where(Player.coins == 100).execute()
更多信息请参考 删除数据。
后续步骤
- 通过 peewee 官方文档 学习更多 peewee 的用法。
- 通过 开发者指南 各章节,学习 TiDB 应用开发最佳实践,例如 插入数据、更新数据、删除数据、单表读取、事务 以及 SQL 性能优化。
- 通过专业的 TiDB 开发者课程,并在通过考试后获得 TiDB 认证。