使用 SQLAlchemy 连接 TiDB

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

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

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

前置条件

完成本教程，你需要：

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

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

• （推荐）参照创建 TiDB Cloud Starter 集群 创建你自己的 TiDB Cloud 集群。
• 参照部署本地测试 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 集群。

步骤 4：运行代码并检查结果

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

   python sqlalchemy_example.py
   

2. 检查 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 社区提问，或提交支持工单。