使用 SQLAlchemy 连接到平凯数据库

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

本文档将展示如何使用平凯数据库和 SQLAlchemy 来完成以下任务：

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

前置需求

如果你还没有 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 并不提供数据库驱动，因此需要单独安装用于连接 TiDB 的驱动。本示例项目使用 PyMySQL 作为数据库驱动。PyMySQL 是一个与 TiDB 兼容的纯 Python 实现的 MySQL 客户端库，并可以在所有平台上安装。

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

第 3 步：配置连接信息

  1. 运行以下命令，将 .env.example 复制并重命名为 .env：

    cp .env.example .env

  2. 复制并粘贴对应 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，密码为空。

  3. 保存 .env 文件。

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

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

    python sqlalchemy_example.py

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

示例代码片段

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

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

连接到平凯数据库

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 声明式映射表。

插入数据

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

更多信息参考插入数据以及 SQLAlchemy Query。

查询数据

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

更多信息参考查询数据以及 SQLAlchemy Query。

更新数据

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

更多信息参考更新数据以及 SQLAlchemy Query。

删除数据

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

更多信息参考删除数据以及 SQLAlchemy Query。

下一步

需要帮助?

如果在开发的过程中遇到问题，可以在 AskTUG 上进行提问，寻求帮助。

下载 PDF