使用 peewee 连接到 TiDB

TiDB 是一个兼容 MySQL 的数据库。peewee 为当前流行的开源 Python ORM (Object Relational Mapper) 之一。

本文档将展示如何使用 TiDB 和 peewee 来完成以下任务：

• 配置你的环境。
• 使用 peewee 连接到 TiDB 集群。
• 构建并运行你的应用程序。你也可以参考示例代码片段，完成基本的 CRUD 操作。

前置需求

• 推荐 Python 3.8 及以上版本。
• Git。
• TiDB 集群。如果你还没有 TiDB 集群，可以按照以下方式创建：
  • （推荐方式）参考创建 TiDB 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 并不提供数据库驱动，因此需要单独安装用于连接 TiDB 的驱动。本示例项目使用 PyMySQL 作为数据库驱动。PyMySQL 是一个与 TiDB 兼容的纯 Python 实现的 MySQL 客户端库，并可以在所有平台上安装。更多信息，参考 peewee 官方文档。

第 3 步：配置连接信息

根据不同的 TiDB 部署方式，使用不同的方法连接到 TiDB 集群。

• TiDB Serverless
• TiDB Dedicated
• 本地部署 TiDB

第 4 步：运行代码并查看结果

1. 运行下述命令，执行示例代码：

   python peewee_example.py
   

2. 查看 Expected-Output.txt，并与你的程序输出进行比较。结果近似即为连接成功。

示例代码片段

你可参考以下关键代码片段，完成自己的应用开发。

完整代码及其运行方式，见代码仓库 tidb-samples/tidb-python-peewee-quickstart。

连接到 TiDB

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 上进行提问，寻求帮助。