| title | summary |
|---|---|
使用 peewee 连接到 TiDB |
了解如何使用 peewee 连接到 TiDB。本文提供了使用 peewee 与 TiDB 交互的 Python 示例代码片段。 |
TiDB 是一个兼容 MySQL 的数据库。peewee 为当前流行的开源 Python ORM (Object Relational Mapper) 之一。
本文档将展示如何使用 TiDB 和 peewee 来完成以下任务：
- 配置你的环境。
- 使用 peewee 连接到 TiDB 集群。
- 构建并运行你的应用程序。你也可以参考示例代码片段，完成基本的 CRUD 操作。
注意
本文档适用于 TiDB Cloud Serverless、TiDB Cloud Dedicated 和本地部署的 TiDB。
- 推荐 Python 3.8 及以上版本。
- Git。
- TiDB 集群。如果你还没有 TiDB 集群，可以按照以下方式创建：
- （推荐方式）参考创建 TiDB Cloud Serverless 集群，创建你自己的 TiDB Cloud 集群。
- 参考部署本地测试 TiDB 集群或部署正式 TiDB 集群，创建本地集群。
本小节演示如何运行示例应用程序的代码，并连接到 TiDB。
运行以下命令，将示例代码仓库克隆到本地：
git clone https://github.com/tidb-samples/tidb-python-peewee-quickstart.git
cd tidb-python-peewee-quickstart运行以下命令，安装示例代码所需要的依赖（包括 peewee 和 PyMySQL）：
pip install -r requirements.txtpeewee 是一个支持多种数据库的 ORM 库。它是对数据库的高层抽象，可以帮助开发者以更面向对象的方式编写 SQL 语句。但 peewee 并不提供数据库驱动，因此需要单独安装用于连接 TiDB 的驱动。本示例项目使用 PyMySQL 作为数据库驱动。PyMySQL 是一个与 TiDB 兼容的纯 Python 实现的 MySQL 客户端库，并可以在所有平台上安装。更多信息，参考 peewee 官方文档。
根据不同的 TiDB 部署方式，使用不同的方法连接到 TiDB 集群。
-
在 TiDB Cloud 的 Clusters 页面中，选择你的 TiDB Cloud Serverless 集群，进入集群的 Overview 页面。
-
点击右上角的 Connect 按钮，将会弹出连接对话框。
-
确认对话框中的配置和你的运行环境一致。
- Connection Type 为
Public。 - Branch 选择
main。 - Connect With 选择
General。 - Operating System 为你的运行环境。
Tip:
如果你在 Windows Subsystem for Linux (WSL) 中运行，请切换为对应的 Linux 发行版。
- Connection Type 为
-
如果你还没有设置密码，点击 Generate Password 生成一个随机密码。
Tip:
如果你之前已经生成过密码，可以直接使用原密码，或点击 Reset Password 重新生成密码。
-
运行以下命令，将
.env.example复制并重命名为.env：cp .env.example .env
-
复制并粘贴对应连接字符串至
.env中。示例结果如下：TIDB_HOST='{host}' # e.g. xxxxxx.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文件。
-
在 TiDB Cloud 的 Clusters 页面中，选择你的 TiDB Cloud Dedicated 集群，进入集群的 Overview 页面。
-
点击右上角的 Connect 按钮，将会出现连接对话框。
-
在连接对话框中，从 Connection Type 下拉列表中选择 Public，并点击 CA cert 下载 CA 文件。
如果你尚未配置 IP 访问列表，请在首次连接前点击 Configure IP Access List 或按照配置 IP 访问列表（英文）中的步骤进行配置。
除 Public 连接类型外，TiDB Cloud Dedicated 还支持 Private Endpoint 和 VPC Peering 连接类型。详情请参阅连接 TiDB Cloud Dedicated 集群（英文）。
-
运行以下命令，将
.env.example复制并重命名为.env：cp .env.example .env
-
复制并粘贴对应的连接字符串至
.env中。示例结果如下：TIDB_HOST='{host}' # e.g. xxxxxx.aws.tidbcloud.com TIDB_PORT='4000' TIDB_USER='{user}' # e.g. xxxxxx.root TIDB_PASSWORD='{password}' TIDB_DB_NAME='test' CA_PATH='{your-downloaded-ca-path}'
注意替换
{}中的占位符为连接对话框中获得的值，并配置前面步骤中下载好的证书路径。 -
保存
.env文件。
-
运行以下命令，将
.env.example复制并重命名为.env：cp .env.example .env
-
复制并粘贴对应 TiDB 的连接字符串至
.env中。示例结果如下：TIDB_HOST='{host}' TIDB_PORT='4000' TIDB_USER='root' TIDB_PASSWORD='{password}' TIDB_DB_NAME='test'
注意替换
{}中的占位符为你的 TiDB 对应的值，并删除CA_PATH这行。如果你在本机运行 TiDB，默认 Host 地址为127.0.0.1，密码为空。 -
保存
.env文件。
-
运行下述命令，执行示例代码：
python peewee_example.py
-
查看
Expected-Output.txt，并与你的程序输出进行比较。结果近似即为连接成功。
你可参考以下关键代码片段，完成自己的应用开发。
完整代码及其运行方式，见代码仓库 tidb-samples/tidb-python-peewee-quickstart。
from peewee import MySQLDatabase
def get_db_engine():
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"
def __str__(self):
return f"Player(name={self.name}, coins={self.coins}, goods={self.goods})"更多信息参考 peewee 模型与字段。
# 插入单个对象
Player.create(name="test", coins=100, goods=100)
# 插入多个对象
data = [
{"name": "test1", "coins": 100, "goods": 100},
{"name": "test2", "coins": 100, "goods": 100},
]
Player.insert_many(data).execute()更多信息参考插入数据。
# 查询所有对象
players = Player.select()
# 查询单个对象
player = Player.get(Player.name == "test")
# 查询多个对象
players = Player.select().where(Player.coins == 100)更多信息参考查询数据。
# 更新单个对象
player = Player.get(Player.name == "test")
player.coins = 200
player.save()
# 批量更新多个对象
Player.update(coins=200).where(Player.coins == 100).execute()更多信息参考更新数据。
# 删除单个对象
player = Player.get(Player.name == "test")
player.delete_instance()
# 批量删除多个对象
Player.delete().where(Player.coins == 100).execute()更多信息参考删除数据。
- 关于 peewee 的更多使用方法，可以参考 peewee 官方文档。
- 你可以继续阅读开发者文档，以获取更多关于 TiDB 应用开发的最佳实践。例如：插入数据、更新数据、删除数据、单表读取、事务、SQL 性能优化等。
- 如果你更倾向于参与课程进行学习，我们也提供专业的 TiDB 开发者课程支持，并在考试后提供相应的资格认证。
如果在开发的过程中遇到问题，可以在 AskTUG 上进行提问，寻求帮助。