SQLAlchemy 连接 OB Cloud 云数据库示例程序

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

前提条件

• 您已安装 Python 3.x 和 pip。
• 您已注册 OB Cloud 云数据库账号,并创建集群实例和 OB Cloud MySQL 兼容模式租户。详情请参见 创建集群实例 和 创建租户。

操作步骤

1. 获取 OB Cloud 云数据库连接串。
2. 安装 SQLAlchemy 库。
3. 编写 test.py 文件，填入数据库连接信息。
4. 运行 test.py 文件。

步骤一：获取 OB Cloud 云数据库连接串

1. 登录 OB Cloud 云数据库控制台。在实例列表页面，展开目标实例的信息，在目标租户下，选择 连接 > 获取连接串。

   更多详细操作，请参见 获取连接串。

2. 根据已创建的 OB Cloud 云数据库填写下面 URL 的对应信息。

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

   参数说明：

   • $host：提供 OB Cloud 云数据库连接地址，例如 t********.********.oceanbase.cloud。
   • $port：提供 OB Cloud 云数据库连接端口，默认为 3306。
   • $database_name：需要访问的数据库名称。

   注意

   连接租户的用户需要拥有该数据库的 CREATE、INSERT、UPDATE 和 SELECT 权限。更多有关账号权限的信息，请参见 创建并管理账号。

   • $user_name：提供访问数据库的账号。
   • $password：提供账户密码。

   示例如下：

   obclient -h t********.********.oceanbase.cloud -P3306 -u mysql001 -p****** -Dtest
   

步骤二：安装 SQLAlchemy 库

SQLAlchemy 是 Python 中最流行的 ORM（对象关系映射）库之一，它提供了 SQL 工具包和 ORM 系统，可以简化数据库操作。SQLAlchemy 支持多种数据库后端，包括 MySQL、PostgreSQL、SQLite 等。

打开命令提示符或 PowerShell 终端，运行以下命令，安装 SQLAlchemy 库。

pip install sqlalchemy mysqlclient

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

pip list | grep SQLAlchemy

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

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

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

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

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

   from sqlalchemy import create_engine, Column, Integer, String, MetaData, Table
   from sqlalchemy.ext.declarative import declarative_base
   from sqlalchemy.orm import sessionmaker
   
   # 数据库连接信息
   DB_USER = 'mysql001'
   DB_PASSWORD = '******'
   DB_HOST = 't********.********.oceanbase.cloud'
   DB_PORT = 3306
   DB_NAME = 'test'
   
   # 创建数据库连接 URL
   DATABASE_URL = f"mysql+mysqldb://{DB_USER}:{DB_PASSWORD}@{DB_HOST}:{DB_PORT}/{DB_NAME}"
   
   # 创建引擎
   engine = create_engine(DATABASE_URL)
   
   # 创建基类
   Base = declarative_base()
   
   # 定义模型
   class User(Base):
       __tablename__ = 'users'
   
       id = Column(Integer, primary_key=True)
       name = Column(String(50), nullable=False)
       age = Column(Integer)
   
       def __repr__(self):
           return f"<User(id={self.id}, name='{self.name}', age={self.age})>"
   
   # 创建表
   Base.metadata.create_all(engine)
   
   # 创建会话
   Session = sessionmaker(bind=engine)
   session = Session()
   
   # 插入数据
   new_users = [
       User(name='John', age=20),
       User(name='Lucy', age=25),
       User(name='Tom', age=30)
   ]
   session.add_all(new_users)
   session.commit()
   
   # 更新数据
   user = session.query(User).filter_by(name='Lucy').first()
   if user:
       user.age = 26
       session.commit()
   
   # 查询数据
   users = session.query(User).all()
   for user in users:
       print(user)
   
   # 关闭会话
   session.close()
   

步骤四：运行 test.py 文件

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

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

   示例如下：

   cd D:\demo\demo
   

2. 运行 test.py 文件。

   示例如下：

   python test.py
   

   返回结果如下：

   <User(id=1, name='John', age=20)>
   <User(id=2, name='Lucy', age=26)>
   <User(id=3, name='Tom', age=30)>
   

错误处理

在使用 SQLAlchemy 连接 OB Cloud 云数据库时，可能会遇到各种错误。以下是一些常见的错误及其处理方法：

1. 连接错误：如果无法连接到数据库，请检查连接参数是否正确，包括主机名、端口、用户名、密码和数据库名。

2. 权限错误：如果遇到权限相关的错误，请确保用户拥有足够的权限来执行所需的操作。

3. SQL 语法错误：如果 SQL 语句有语法错误，请检查 SQL 语句的语法是否正确。

4. 数据类型错误：如果插入的数据类型与表定义不匹配，请确保插入的数据类型正确。

在代码中，我们可以使用 try-except 语句来捕获和处理这些错误，确保程序在遇到错误时能够优雅地处理，而不是直接崩溃。同时，我们可以使用 logging 模块记录错误信息，方便调试和问题排查。

相关文档

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