SQLAlchemy 使用手册

📄字数 3.6K

👁️阅读量 加载中...

本文档将介绍 XuguDB 对 SQLAlchemy 框架的适配情况，并提供使用 SQLAlchemy 开发 XuguDB 应用的 Demo 指南。

框架名称	适配起始版本	官网地址	源框架下载地址	XuguDB框架下载	demo程序地址
SQLAlchemy	1.4.36	SQLAlchemy官网	SQLAlchemy官方下载	版本发布列表	xugu-sqlalchemy

提示

其他适配版本可详见版本发布列表

一、方言包使用说明

获取 XuguDB 方言压缩包后，将解压获取到的 xg 目录放置到 SQLAlchemy 应用同级目录中，并在 SQLAlchemy 应用入口导入 XuguDB 方言包，即可正常开发 XuguDB SQLAlchemy 应用。

import xg
import xg.xgPython
from sqlalchemy.dialects import registry
# 导入 XuguDB SQLAlchemy 方言包
registry.register("xg", "xg.xgPython", "dialect")

二、连接示例程序

本章节将介绍在 SQLAlchemy 应用连接 XuguDB ，实现基本的数据库操作，包括创建表、插入数据、更新数据和查询数据等。

2.1 前提条件

• 已安装 Python 3.x 。
• 计算机已部署 XuguDB 12.0.0 或有能连接访问的虚谷数据库实例。

2.2 操作步骤

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

提示

本文是在Windows环境中进行操作说明的，如果使用的是其它操作环境，那么与本文中的步骤可能存在不同。

2.2.1 修改 main.py 文件，编写数据库连接信息

提示

SQLAlchemy 是一个功能强大的 ORM 库，它提供了两种使用方式：Core 和 ORM。Core 提供了 SQL 表达式语言，而 ORM 提供了对象关系映射功能。本文示例主要使用 ORM 方式。

根据 XuguDB 连接信息修改 Demo 项目 main.py 文件，填入数据库连接信息。

from sqlalchemy.dialects import registry
from sqlalchemy import create_engine, Column, Integer, String, BLOB, CLOB
from sqlalchemy.orm import sessionmaker
from sqlalchemy.orm import declarative_base
import xg
import xg.xgPython
from sqlalchemy import ForeignKey
# 导入虚谷 SQLAlchemy 方言包
registry.register("xg", "xg.xgPython", "dialect")
# 创建数据库连接
# 连接串结构为'xg://用户名:密码@ip:端口/库' xg为虚谷标识
engine = create_engine('xg://SYSDBA:SYSDBA@10.28.10.18,127.0.0.1:5139/SYSTEM')


# 创建一个基类
Base = declarative_base()

# 定义一个表模型
class User(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True)
    name = Column(String(50))
    age = Column(Integer)

    def __repr__(self):
        return "<User(name='%s', age='%s')>" % (self.name, self.age)

#JOIN 查询表结构
class Author(Base):
    __tablename__ = 'author'
    id = Column(Integer, primary_key=True)
    name = Column(String(50))

class Book(Base):
    __tablename__ = 'book'
    id = Column(Integer, primary_key=True)
    title = Column(String(50))
    author_id = Column(Integer, ForeignKey('author.id'))

class LobData(Base):
    __tablename__ = 'lobdata'
    id = Column(Integer, primary_key=True)
    c1 = Column(CLOB)


# 删除
Base.metadata.drop_all(engine)
# 创建表
Base.metadata.create_all(engine)

# 创建一个 Session 类型的实例
Session = sessionmaker(bind=engine)
session = Session()

# 添加记录
def add_user(name,age):
    user = User(name=name,age=age)
    session.add(user)
    session.commit()
    return user.id

# 删除记录
def delete_user(user_id):
    user = session.get(User,user_id)
    session.delete(user)
    session.commit()

# 更新记录
def update_user(user_id, new_name):
    user = session.get(User,user_id)
    user.name = new_name
    session.commit()

# 查询记录
def get_user(user_id):
    user = session.get(User,user_id)
    return user

# 查询所有记录
def get_user_all():
    users = session.query(User).all()
    return users

def filter(name):
    user = session.query(User).filter(User.name == name).first()
    return user



print('-------------普通类型的增删改查------------')


print('-------------主键自增插入')
id = add_user("Alice",17)
id2 = add_user("Bob",18)
id2 = add_user("USER1",19)
id2 = add_user("USER2",20)

print('--------------条件查询-----------')

alice = get_user(id)
print(alice.name)  # 输出: Alice

id1 = filter('Alice')
print(id1.id, id1.name, id1.age)

print('--------------更新操作')
update_user(id2, "Bobby")

print('--------------删除操作')
delete_user(id)

print("\nafter update&delete\n")
users = get_user_all()
for user in users:
    print(user.id, user.name, user.age)

print('--------------查询所有数据')

users = get_user_all()
for user in users:
    print(user.id, user.name, user.age)

print('--------------union')

def union_demo():
    user1 = session.query(User).filter(User.name == 'Bobby')
    user2 = session.query(User).filter(User.name == 'USER1')
    user = user1.union(user2).all()
    return user

users = union_demo()
for user in users:
    print(user.id, user.name, user.age)

# join 查询

print('--------------join 查询')

author1 = Author(name='John Doe')
session.add(author1)
session.commit()
book1 = Book(title='Book 1', author_id=author1.id)
session.add(book1)
book2 = Book(title='Book 2', author_id=author1.id)
session.add(book2)
session.commit()

# JOIN查询
query = session.query(Author, Book).join(Book)
results = query.all()

# 输出结果
for author, book in results:
    print(author.name, book.title)


# 大对象插入
def add_lob():
    with open('clob.txt', 'r',encoding='utf-8') as file1:
        blob = file.read()
        clob = file1.read()
        record = LobData(c1=clob)
        session.add(record)
        session.commit()

2.2.2 运行 main.py 文件

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

python main.py

返回结果如下：

-------------普通类型的增删改查------------
-------------主键自增插入
1
2
--------------查询所有数据
1 Alice 17
2 Bob 18
3 xxxxxxx 22
4 USER2 20
--------------条件查询-----------
Alice
1 Alice 17
--------------更新操作
--------------删除操作

after update&delete

2 Bob 18
3 xxxxxxx 22
4 Bobby 20
--------------查询所有数据
2 Bob 18
3 xxxxxxx 22
4 Bobby 20
--------------union
4 Bobby 20
--------------join 查询
John Doe Book 1
John Doe Book 2

三、常见问题&解决办法

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

• 连接错误：如果无法连接到数据库，请检查连接参数是否正确，包括主机名、端口、用户名、密码和数据库名，可以用其他客户端工具测试连接信息。

• 表不存在错误：确定是否已存在相关表结构，当前用户是否有权限检索到，模式是否指定正确，create_all 操作是否成功执行。