使用 SQLAlchemy 连接 TiDB

TiDB 是一个兼容 MySQL 的数据库，SQLAlchemy 是一个流行的 Python SQL 工具包和对象关系映射（ORM）库。

在本教程中，你可以学习如何使用 TiDB 和 SQLAlchemy 完成以下任务：

搭建你的开发环境。
使用 SQLAlchemy 连接到你的 TiDB 集群。
构建并运行你的应用程序。你还可以参考基本 CRUD 操作的示例代码片段。

Note

本教程适用于平凯云 Starter , 平凯云 Essential 以及 TiDB 自建集群。

前置条件

完成本教程，你需要：

Python 3.8 或更高版本。
Git。
一个 TiDB 集群。

如果你还没有 TiDB 集群，可以按如下方式创建：

（推荐）参照创建平凯云 Starter 集群创建你自己的平凯数据库云服务集群。
参照部署本地测试 TiDB 集群或部署生产环境 TiDB 集群创建本地集群。

运行示例应用并连接 TiDB

本节演示如何运行示例应用代码并连接到 TiDB。

步骤 1：克隆示例应用仓库

在终端窗口中运行以下命令，克隆示例代码仓库：

git clone https://github.com/tidb-samples/tidb-python-sqlalchemy-quickstart.git
cd tidb-python-sqlalchemy-quickstart

步骤 2：安装依赖

运行以下命令，为示例应用安装所需的依赖包（包括 SQLAlchemy 和 PyMySQL）：

pip install -r requirements.txt

为什么使用 PyMySQL？

SQLAlchemy 是一个支持多种数据库的 ORM 库。它为数据库提供了高级抽象，帮助开发者以更面向对象的方式编写 SQL 语句。但 SQLAlchemy 并不自带数据库驱动。要连接数据库，你需要安装相应的数据库驱动。本示例应用使用 PyMySQL 作为数据库驱动，它是一个纯 Python 的 MySQL 客户端库，兼容 TiDB，并可在所有平台上安装。

你也可以使用其他数据库驱动，如 mysqlclient 和 mysql-connector-python。但它们不是纯 Python 库，需要相应的 C/C++ 编译器和 MySQL 客户端进行编译。更多信息请参考 SQLAlchemy 官方文档。

步骤 3：配置连接信息

根据你选择的 TiDB 部署方式，连接到你的 TiDB 集群。

Note

目前，平凯云 Starter 集群有一个限制：如果 5 分钟内没有活跃连接，集群会自动关闭，所有连接会被断开。因此，当你在平凯云 Starter 集群上使用 SQLAlchemy 时，连接池可能会遇到 OperationalError，如 Lost connection to MySQL server during query 或 MySQL Connection not available。为避免此类错误，你可以将 pool_recycle 参数设置为 300。更多信息请参见 SQLAlchemy 文档中的 Dealing with Disconnects。

进入 Clusters 页面，点击目标集群名称，进入集群概览页面。
点击右上角的 Connect，弹出连接对话框。
确保连接对话框中的配置与你的操作环境一致。
- Connection Type 设置为 Public
- Branch 设置为 main
- Connect With 设置为 General
- Operating System 与你的环境一致
Tip

如果你的程序运行在 Windows Subsystem for Linux (WSL) 中，请切换到对应的 Linux 发行版。
点击 Generate Password 生成随机密码。

Tip

如果你之前已创建过密码，可以继续使用原密码，或点击 Reset 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 文件。

运行以下命令，复制 .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 sqlalchemy_example.py
```
检查 Expected-Output.txt 文件，确认输出是否一致。

示例代码片段

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

完整示例代码及运行方法请参见 tidb-samples/tidb-python-sqlalchemy-quickstart 仓库。

连接 TiDB

from sqlalchemy import create_engine, URL
from sqlalchemy.orm import sessionmaker

def get_db_engine():
    connect_args = {}
    if ${ca_path}:
        connect_args = {
            "ssl_verify_cert": True,
            "ssl_verify_identity": True,
            "ssl_ca": ${ca_path},
        }
    return create_engine(
        URL.create(
            drivername="mysql+pymysql",
            username=${tidb_user},
            password=${tidb_password},
            host=${tidb_host},
            port=${tidb_port},
            database=${tidb_db_name},
        ),
        connect_args=connect_args,
    )

engine = get_db_engine()
Session = sessionmaker(bind=engine)

使用该函数时，你需要将 ${tidb_host}、${tidb_port}、${tidb_user}、${tidb_password}、${tidb_db_name} 和 ${ca_path} 替换为你 TiDB 集群的实际值。

定义数据表

from sqlalchemy import Column, Integer, String
from sqlalchemy.orm import declarative_base

Base = declarative_base()

class Player(Base):
    id = Column(Integer, primary_key=True)
    name = Column(String(32), unique=True)
    coins = Column(Integer)
    goods = Column(Integer)

    __tablename__ = "players"

更多信息请参考 SQLAlchemy 文档：Mapping classes with declarative。

插入数据

with Session() as session:
    player = Player(name="test", coins=100, goods=100)
    session.add(player)
    session.commit()

更多信息请参考插入数据。

查询数据

with Session() as session:
    player = session.query(Player).filter_by(name == "test").one()
    print(player)

更多信息请参考查询数据。

更新数据

with Session() as session:
    player = session.query(Player).filter_by(name == "test").one()
    player.coins = 200
    session.commit()

更多信息请参考更新数据。

删除数据

with Session() as session:
    player = session.query(Player).filter_by(name == "test").one()
    session.delete(player)
    session.commit()

更多信息请参考删除数据。

后续步骤

通过 SQLAlchemy 官方文档学习更多 SQLAlchemy 的用法。
通过开发者指南各章节，学习 TiDB 应用开发最佳实践，例如插入数据、更新数据、删除数据、单表读取、事务以及 SQL 性能优化。
通过专业的 TiDB 开发者课程，并在通过考试后获得 TiDB 认证。

需要帮助？

在 Discord 或 Slack 社区提问，或提交支持工单。