Tortoise ORM 连接 OceanBase 数据库示例程序

Tortoise ORM 是一个基于 asyncio 的异步对象关系映射（ORM）库，专为现代 Python 应用程序设计。它的异步特性使其特别适合以下场景：

• 处理大量并发数据库连接
• 构建高性能的 Web 服务器
• 需要同时处理多个网络请求的应用

在 Tortoise ORM 中，异步通信使得数据库操作不会阻塞整个应用程序，可以同时处理其他任务，显著提高应用程序的整体性能和响应速度。

本文将介绍如何使用 Tortoise ORM 连接 OceanBase 数据库，实现基本的数据库操作，包括创建表、插入数据、更新数据和查询数据等。

前提条件

• 您已安装 Python 3.7 或更高版本和 pip。
• 您已安装 OceanBase 数据库并且创建了 MySQL 模式租户。

操作步骤

1. 获取 OceanBase 数据库连接串。
2. 安装 Tortoise ORM 和 MySQL 客户端库。
3. 编写 test_tortoise.py 文件，填入数据库连接信息。
4. 运行 test_tortoise.py 文件。

步骤一：获取 OceanBase 数据库连接串

联系 OceanBase 数据库部署人员或者管理员获取相应的数据库连接串。

obclient -h$host -P$port -u$user_name -p$password -D$database_name

参数说明：

• $host：提供 OceanBase 数据库连接 IP。OceanBase 数据库代理（OceanBase Database Proxy，ODP）连接方式使用的是一个 ODP 地址；直连方式使用的是 OBServer 节点的 IP 地址。

• $port：提供 OceanBase 数据库连接端口。ODP 连接的方式默认是 2883，在部署 ODP 时可自定义；直连方式默认是 2881，在部署 OceanBase 数据库时可自定义。

• $database_name：需要访问的数据库名称。

  注意

  连接租户的用户需要拥有该数据库的 CREATE、INSERT、UPDATE 和 SELECT 权限。更多有关用户权限的信息，请参见 MySQL 模式下的权限分类。

• $user_name：提供租户的连接账户。ODP 连接的常用格式：用户名@租户名#集群名 或者 集群名:租户名:用户名；直连方式格式：用户名@租户名。

• $password：提供账户密码。

更多连接串的信息，请参见 通过 OBClient 连接 OceanBase 租户。

示例如下：

obclient -hxxx.xxx.xxx.xxx -P2881 -utest_user001@mysql001 -p****** -Dtest

步骤二：安装 Tortoise ORM 和 MySQL 客户端库

Tortoise ORM 是一个异步的 Python ORM 库，专为 asyncio 设计，支持 MySQL、PostgreSQL 和 SQLite 等数据库。

打开命令提示符或 PowerShell 终端，运行以下命令，安装 Tortoise ORM 和 MySQL 客户端库。

pip install tortoise-orm aiomysql

安装完成后，可以通过以下命令验证安装是否成功：

pip list | grep tortoise

步骤三：编写 test_tortoise.py 文件，填入数据库连接信息

根据 步骤一：获取 OceanBase 数据库连接串 中的信息编写 test_tortoise.py 文件，填入数据库连接信息。

1. 创建一个名为 test_tortoise.py 的文件。

2. 在 test_tortoise.py 文件中填入以下内容，并根据实际情况修改数据库连接信息。

   test_tortoise.py 文件内容示例如下：

   from tortoise import Tortoise, run_async
   from tortoise.models import Model
   from tortoise import fields
   import asyncio
   
   # 定义模型
   from datetime import datetime
   
   class User(Model):
       id = fields.IntField(pk=True)
       username = fields.CharField(max_length=50, null=False)
       email = fields.CharField(max_length=100, unique=True, null=False)
       password_hash = fields.CharField(max_length=255, null=False)
       created_at = fields.DatetimeField(auto_now_add=True, null=False)
       updated_at = fields.DatetimeField(auto_now=True, null=True)
   
       def __str__(self):
           return f"{self.username} ({self.email})"
   
       class Meta:
           table = "users"
   
   # 数据库配置
   DB_CONFIG = {
       'connections': {
           'oceanbase': {
               'engine': 'tortoise.backends.mysql',
               'credentials': {
                   'host': 'xxx.xxx.xxx.xxx',
                   'port': 2881,
                   'user': 'test_user001@mysql001',
                   'password': '******',
                   'database': 'test',
               }
           },
       },
       'apps': {
           'models': {
               'models': ['__main__'],  # 包含当前文件的模型
               'default_connection': 'oceanbase',
           }
       },
       'use_tz': False,
       'timezone': 'Asia/Shanghai',
   }
   
   async def init_db():
       # 初始化数据库连接
       await Tortoise.init(config=DB_CONFIG)
       # 创建表
       await Tortoise.generate_schemas()
   
   async def main():
       # 初始化数据库
       await init_db()
   
       # 插入数据
       from datetime import datetime
   
       await User.create(
           username='john_doe',
           email='john@example.com',
           password_hash='hashed_password_123',
           created_at=datetime.now(),
           updated_at=datetime.now()
       )
   
       # 更新数据
       user = await User.filter(email='john@example.com').first()
       if user:
           user.username = 'john_updated'
           user.updated_at = datetime.now()
           await user.save()
   
       # 查询数据
       users = await User.all()
       for user in users:
           print(user)
   
       # 关闭数据库连接
       await Tortoise.close_connections()
   
   if __name__ == '__main__':
       run_async(main())    
   

步骤四：运行 test_tortoise.py 文件

打开命令提示符或 PowerShell 终端，运行 test_tortoise.py 文件，查询数据并输出结果。

1. 进入到 test_tortoise.py 文件所在的目录。

   示例如下：

   cd ~/orm-demo
   

2. 运行 test_tortoise.py 文件。

   示例如下：

   python test_tortoise.py
   

   返回结果如下：

   john_updated (john@example.com)
   

相关文档

• 更多连接 OceanBase 数据库的信息，请参见 连接方式概述。

• 更多创建数据库的信息，请参见 CREATE DATABASE。

• Tortoise ORM 官方文档：https://tortoise-orm.readthedocs.io/