Sqlalchemy Xugudb Adapter

SQLAlchemy 虚谷数据库适配指南

概述

本技能提供 SQLAlchemy 适配虚谷数据库(XuguDB)的完整流程。SQLAlchemy 是一个功能强大的 Python ORM 库，通过虚谷数据库专用的方言包 xugu-sqlalchemy，可以无缝集成 SQLAlchemy 框架的各种功能，包括模型定义、CRUD 操作、事务处理、关联关系等。

适配流程

在开始适配前，按以下顺序检查和修改：

1. 依赖安装 → 2. 连接配置 → 3. 模型定义 → 4. 数据库迁移 → 5. CRUD 操作 → 6. 事务处理 → 7. 性能优化 → 8. 测试验证

第一步：安装依赖

安装 SQLAlchemy

# 安装 SQLAlchemy
pip install sqlalchemy

# 或者使用 pip3
pip3 install sqlalchemy

安装虚谷数据库驱动

1. 下载方言包 - 从虚谷数据库官方下载 SQLAlchemy 方言压缩包
2. 解压方言包 - 解压压缩包获取 xg 目录
3. 放置方言文件 - 将 xg 目录放置在 Python 项目根目录下
4. 注册方言 - 在 Python 程序入口处注册方言

验证驱动安装

import xg
import xg.xgPython
from sqlalchemy.dialects import registry
from sqlalchemy import create_engine

# 注册方言
registry.register("xg", "xg.xgPython", "dialect")

# 创建数据库连接
engine = create_engine('xg://SYSDBA:SYSDBA@127.0.0.1:5138/SYSTEM')

# 测试连接
try:
    with engine.connect() as connection:
        print("Database connection established successfully!")
except Exception as e:
    print(f"Failed to connect to database: {e}")

第二步：配置连接

连接字符串格式

虚谷数据库连接字符串格式：

xg://用户名:密码@IP地址:端口/数据库名

连接参数说明

| 参数 | 说明 | 默认值 | |------|------|--------| | 用户名 | 数据库用户名 | SYSDBA | | 密码 | 数据库密码 | - | | IP地址 | 服务器地址 | 127.0.0.1 | | 端口 | 端口号 | 5138 | | 数据库名 | 数据库名 | SYSTEM |

基本连接配置

import xg
import xg.xgPython
from sqlalchemy.dialects import registry
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, declarative_base

# 注册方言
registry.register("xg", "xg.xgPython", "dialect")

# 创建数据库连接
engine = create_engine('xg://SYSDBA:SYSDBA@127.0.0.1:5138/SYSTEM')

# 创建会话工厂
Session = sessionmaker(bind=engine)

# 创建基类
Base = declarative_base()

连接池配置

import xg
import xg.xgPython
from sqlalchemy.dialects import registry
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, declarative_base
from sqlalchemy.pool import QueuePool

# 注册方言
registry.register("xg", "xg.xgPython", "dialect")

# 创建数据库连接（带连接池配置）
engine = create_engine(
    'xg://SYSDBA:SYSDBA@127.0.0.1:5138/SYSTEM',
    poolclass=QueuePool,
    pool_size=20,          # 连接池大小
    max_overflow=10,       # 超出连接池大小的额外连接数
    pool_timeout=30,       # 获取连接超时时间（秒）
    pool_recycle=1800,     # 连接回收时间（秒）
    pool_pre_ping=True     # 连接前检查
)

# 创建会话工厂
Session = sessionmaker(bind=engine)

# 创建基类
Base = declarative_base()

第三步：定义模型

基本模型定义

import xg
import xg.xgPython
from sqlalchemy.dialects import registry
from sqlalchemy import create_engine, Column, Integer, String, Boolean, DateTime
from sqlalchemy.orm import sessionmaker, declarative_base
from datetime import datetime

# 注册方言
registry.register("xg", "xg.xgPython", "dialect")

# 创建数据库连接
engine = create_engine('xg://SYSDBA:SYSDBA@127.0.0.1:5138/SYSTEM')

# 创建基类
Base = declarative_base()

# 定义用户模型
class User(Base):
    __tablename__ = 'users'
    
    id = Column(Integer, primary_key=True, autoincrement=True)
    username = Column(String(50), unique=True, nullable=False)
    email = Column(String(100), unique=True, nullable=False)
    age = Column(Integer, default=0)
    is_active = Column(Boolean, default=True)
    created_at = Column(DateTime, default=datetime.now)
    updated_at = Column(DateTime, default=datetime.now, onupdate=datetime.now)
    
    def __repr__(self):
        return f""

字段类型说明

| 字段类型 | 说明 | 示例 | |----------|------|------| | Integer | 整数 | Column(Integer) | | BigInteger | 大整数 | Column(BigInteger) | | String | 字符串 | Column(String(100)) | | Text | 文本 | Column(Text) | | Float | 浮点数 | Column(Float) | | Numeric | 数值 | Column(Numeric(10, 2)) | | Boolean | 布尔值 | Column(Boolean) | | Date | 日期 | Column(Date) | | DateTime | 日期时间 | Column(DateTime) | | Time | 时间 | Column(Time) | | UUID | UUID | Column(UUID) | | LargeBinary | 二进制大对象 | Column(LargeBinary) | | JSON | JSON | Column(JSON) |

关联关系

import xg
import xg.xgPython
from sqlalchemy.dialects import registry
from sqlalchemy import create_engine, Column, Integer, String, ForeignKey, Table
from sqlalchemy.orm import sessionmaker, declarative_base, relationship
from datetime import datetime

# 注册方言
registry.register("xg", "xg.xgPython", "dialect")

# 创建数据库连接
engine = create_engine('xg://SYSDBA:SYSDBA@127.0.0.1:5138/SYSTEM')

# 创建基类
Base = declarative_base()

# 定义用户模型
class User(Base):
    __tablename__ = 'users'
    
    id = Column(Integer, primary_key=True, autoincrement=True)
    username = Column(String(50), unique=True, nullable=False)
    email = Column(String(100), unique=True, nullable=False)
    age = Column(Integer, default=0)
    created_at = Column(DateTime, default=datetime.now)
    
    # 定义关联关系
    articles = relationship("Article", back_populates="author")
    
    def __repr__(self):
        return f""

# 定义文章模型
class Article(Base):
    __tablename__ = 'articles'
    
    id = Column(Integer, primary_key=True, autoincrement=True)
    title = Column(String(200), nullable=False)
    content = Column(String(1000))
    user_id = Column(Integer, ForeignKey('users.id'))
    created_at = Column(DateTime, default=datetime.now)
    
    # 定义关联关系
    author = relationship("User", back_populates="articles")
    tags = relationship("Tag", secondary="article_tags", back_populates="articles")
    
    def __repr__(self):
        return f""

# 定义标签模型
class Tag(Base):
    __tablename__ = 'tags'
    
    id = Column(Integer, primary_key=True, autoincrement=True)
    name = Column(String(50), unique=True, nullable=False)
    
    # 定义关联关系
    articles = relationship("Article", secondary="article_tags", back_populates="tags")
    
    def __repr__(self):
        return f""

# 定义文章标签关联表
article_tags = Table('article_tags', Base.metadata,
    Column('article_id', Integer, ForeignKey('articles.id'), primary_key=True),
    Column('tag_id', Integer, ForeignKey('tags.id'), primary_key=True)
)

第四步：数据库迁移

创建表

import xg
import xg.xgPython
from sqlalchemy.dialects import registry
from sqlalchemy import create_engine
from sqlalchemy.orm import declarative_base

# 注册方言
registry.register("xg", "xg.xgPython", "dialect")

# 创建数据库连接
engine = create_engine('xg://SYSDBA:SYSDBA@127.0.0.1:5138/SYSTEM')

# 创建基类
Base = declarative_base()

# 定义模型（略）

# 创建所有表
Base.metadata.create_all(engine)

使用迁移工具

# 安装迁移工具
# pip install alembic

# 初始化迁移
# alembic init alembic

# 生成迁移脚本
# alembic revision --autogenerate -m "create_users_table"

# 执行迁移
# alembic upgrade head

# 回滚迁移
# alembic downgrade -1

第五步：CRUD 操作

创建记录

import xg
import xg.xgPython
from sqlalchemy.dialects import registry
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, declarative_base
from datetime import datetime

# 注册方言
registry.register("xg", "xg.xgPython", "dialect")

# 创建数据库连接
engine = create_engine('xg://SYSDBA:SYSDBA@127.0.0.1:5138/SYSTEM')

# 创建会话工厂
Session = sessionmaker(bind=engine)

# 创建基类
Base = declarative_base()

# 定义用户模型
class User(Base):
    __tablename__ = 'users'
    
    id = Column(Integer, primary_key=True, autoincrement=True)
    username = Column(String(50), unique=True, nullable=False)
    email = Column(String(100), unique=True, nullable=False)
    age = Column(Integer, default=0)
    created_at = Column(DateTime, default=datetime.now)

# 创建表
Base.metadata.create_all(engine)

# 创建会话
session = Session()

# 创建单个记录
new_user = User(username='john', email='john@example.com', age=25)
session.add(new_user)
session.commit()
print(f"Created user ID: {new_user.id}")

# 批量创建
users_data = [
    User(username='user1', email='user1@example.com', age=20),
    User(username='user2', email='user2@example.com', age=25),
    User(username='user3', email='user3@example.com', age=30),
]
session.add_all(users_data)
session.commit()
print("Created users successfully")

# 关闭会话
session.close()

查询记录

import xg
import xg.xgPython
from sqlalchemy.dialects import registry
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, declarative_base

# 注册方言
registry.register("xg", "xg.xgPython", "dialect")

# 创建数据库连接
engine = create_engine('xg://SYSDBA:SYSDBA@127.0.0.1:5138/SYSTEM')

# 创建会话工厂
Session = sessionmaker(bind=engine)

# 创建会话
session = Session()

# 查询所有记录
users = session.query(User).all()
for user in users:
    print(f"User: {user.username}, Age: {user.age}")

# 查询单个记录
user = session.query(User).filter_by(id=1).first()
if user:
    print(f"Found user: {user.username}")

# 条件查询
users = session.query(User).filter(User.age > 18).all()
for user in users:
    print(f"User: {user.username}, Age: {user.age}")

# 查询特定字段
users = session.query(User.username, User.email).all()
for user in users:
    print(f"Username: {user.username}, Email: {user.email}")

# 分页查询
users = session.query(User).limit(10).offset(0).all()
for user in users:
    print(f"User: {user.username}")

# 排序查询
users = session.query(User).order_by(User.age.desc()).all()
for user in users:
    print(f"User: {user.username}, Age: {user.age}")

# 计数
count = session.query(User).filter(User.age > 18).count()
print(f"Found {count} users")

# 关闭会话
session.close()

更新记录

import xg
import xg.xgPython
from sqlalchemy.dialects import registry
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, declarative_base

# 注册方言
registry.register("xg", "xg.xgPython", "dialect")

# 创建数据库连接
engine = create_engine('xg://SYSDBA:SYSDBA@127.0.0.1:5138/SYSTEM')

# 创建会话工厂
Session = sessionmaker(bind=engine)

# 创建会话
session = Session()

# 更新单个记录
user = session.query(User).filter_by(id=1).first()
if user:
    user.age = 30
    session.commit()
    print(f"Updated user: {user.username}, Age: {user.age}")

# 批量更新
session.query(User).filter(User.age  :age'), {'age': 18})
    for row in result:
        print(f"User: {row.username}, Age: {row.age}")

# 执行原生 SQL 更新
with engine.connect() as connection:
    connection.execute(text('UPDATE users SET age = :age WHERE id = :id'), {'age': 30, 'id': 1})
    connection.commit()
    print("Updated user successfully")

关联查询

import xg
import xg.xgPython
from sqlalchemy.dialects import registry
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, joinedload

# 注册方言
registry.register("xg", "xg.xgPython", "dialect")

# 创建数据库连接
engine = create_engine('xg://SYSDBA:SYSDBA@127.0.0.1:5138/SYSTEM')

# 创建会话工厂
Session = sessionmaker(bind=engine)

# 创建会话
session = Session()

# 使用 joinedload 预加载关联查询
users = session.query(User).options(joinedload(User.articles)).all()
for user in users:
    print(f"User: {user.username}")
    for article in user.articles:
        print(f"  Article: {article.title}")

# 关闭会话
session.close()

第七步：性能优化

批量操作

import xg
import xg.xgPython
from sqlalchemy.dialects import registry
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, declarative_base

# 注册方言
registry.register("xg", "xg.xgPython", "dialect")

# 创建数据库连接
engine = create_engine('xg://SYSDBA:SYSDBA@127.0.0.1:5138/SYSTEM')

# 创建会话工厂
Session = sessionmaker(bind=engine)

# 创建会话
session = Session()

# 批量插入
users_data = []
for i in range(1000):
    users_data.append(User(
        username=f'user{i}',
        email=f'user{i}@example.com',
        age=20 + (i % 50),
    ))

# 分批插入，每批 100 条
batch_size = 100
for i in range(0, len(users_data), batch_size):
    batch = users_data[i:i + batch_size]
    session.add_all(batch)
    session.flush()

session.commit()
print("Bulk insert completed")

# 关闭会话
session.close()

查询优化

import xg
import xg.xgPython
from sqlalchemy.dialects import registry
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, joinedload, load_only

# 注册方言
registry.register("xg", "xg.xgPython", "dialect")

# 创建数据库连接
engine = create_engine('xg://SYSDBA:SYSDBA@127.0.0.1:5138/SYSTEM')

# 创建会话工厂
Session = sessionmaker(bind=engine)

# 创建会话
session = Session()

# 使用特定字段
users = session.query(User).options(load_only(User.id, User.username, User.email)).all()
for user in users:
    print(f"Username: {user.username}, Email: {user.email}")

# 使用关联查询优化
users = session.query(User).options(joinedload(User.articles)).limit(10).all()
for user in users:
    print(f"User: {user.username}")
    for article in user.articles:
        print(f"  Article: {article.title}")

# 关闭会话
session.close()

第八步：测试验证

运行测试

import xg
import xg.xgPython
from sqlalchemy.dialects import registry
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, declarative_base
import unittest

# 注册方言
registry.register("xg", "xg.xgPython", "dialect")

# 创建数据库连接
engine = create_engine('xg://SYSDBA:SYSDBA@127.0.0.1:5138/SYSTEM')

# 创建基类
Base = declarative_base()

# 定义用户模型
class User(Base):
    __tablename__ = 'users'
    
    id = Column(Integer, primary_key=True, autoincrement=True)
    username = Column(String(50), unique=True, nullable=False)
    email = Column(String(100), unique=True, nullable=False)
    age = Column(Integer, default=0)

# 创建表
Base.metadata.create_all(engine)

# 创建会话工厂
Session = sessionmaker(bind=engine)

class TestUserModel(unittest.TestCase):
    def setUp(self):
        self.session = Session()
    
    def tearDown(self):
        self.session.rollback()
        self.session.close()
    
    def test_create_user(self):
        user = User(username='testuser', email='test@example.com', age=25)
        self.session.add(user)
        self.session.commit()
        
        self.assertIsNotNone(user.id)
        self.assertEqual(user.username, 'testuser')
        self.assertEqual(user.email, 'test@example.com')
        self.assertEqual(user.age, 25)
    
    def test_find_user_by_id(self):
        user = User(username='testuser', email='test@example.com', age=25)
        self.session.add(user)
        self.session.commit()
        
        found_user = self.session.query(User).filter_by(id=user.id).first()
        self.assertIsNotNone(found_user)
        self.assertEqual(found_user.username, 'testuser')
    
    def test_update_user(self):
        user = User(username='testuser', email='test@example.com', age=25)
        self.session.add(user)
        self.session.commit()
        
        user.age = 30
        self.session.commit()
        
        updated_user = self.session.query(User).filter_by(id=user.id).first()
        self.assertEqual(updated_user.age, 30)
    
    def test_delete_user(self):
        user = User(username='testuser', email='test@example.com', age=25)
        self.session.add(user)
        self.session.commit()
        
        self.session.delete(user)
        self.session.commit()
        
        deleted_user = self.session.query(User).filter_by(id=user.id).first()
        self.assertIsNone(deleted_user)

if __name__ == '__main__':
    unittest.main()

验证数据库连接

import xg
import xg.xgPython
from sqlalchemy.dialects import registry
from sqlalchemy import create_engine, text

# 注册方言
registry.register("xg", "xg.xgPython", "dialect")

# 创建数据库连接
engine = create_engine('xg://SYSDBA:SYSDBA@127.0.0.1:5138/SYSTEM')

# 验证数据库连接
try:
    with engine.connect() as connection:
        print("Database connection established successfully!")
        
        # 验证表结构
        result = connection.execute(text('DESCRIBE users'))
        columns = result.fetchall()
        print("Table structure:")
        for column in columns:
            print(f"  {column}")
        
except Exception as e:
    print(f"Failed to connect to database: {e}")

常见问题排查

1. 方言包找不到

现象： ModuleNotFoundError: No module named 'xg'

解决： 确保 xg 目录已正确放置在 Python 项目根目录下

2. 连接错误

现象： 无法连接到虚谷数据库

解决：

• 检查连接字符串中的主机、端口、用户名、密码、库名是否正确
• 确保虚谷数据库服务已启动
• 检查网络连接和防火墙设置
• 建议先用其他数据库客户端（如 XGConsole）测试连接信息

3. 表不存在错误

现象： 提示目标表不存在

解决：

• 确认是否已执行 Base.metadata.create_all(engine) 来创建表
• 检查当前连接的数据库用户是否有访问该表的权限
• 如果数据库有模式（Schema）概念，请确认模型中是否正确指定了 __table_args__ = {'schema': '模式名'}

#

…

Source & license

This open-source skill is cataloged on AgentStack and links to its original source — we do not rehost the code.

• Author: kourou25
• Source: kourou25/xugudb-dev-skills
• License: Apache-2.0

Install and usage instructions live in the source repository linked above.