openGauss-sqlalchemy

为SQLAlchemy框架适配的OpenGauss方言驱动库。

项目已通过SQLAlchemy官方测试套件验证。

运行环境要求

• 适用于openGauss的psycopg2驱动

  请从以下地址下载并安装至运行环境：

  https://gitcode.com/opengauss/openGauss-connector-python-psycopg2

  或

  https://github.com/opengauss-mirror/openGauss-connector-python-psycopg2

重要说明：

使用版本不匹配的psycopg2驱动将导致SQLAlchemy 2.0版本中部分autoescape相关测试用例失败。

如需使用autoescape相关功能，请确保：

1. 系统中已正确安装openGauss数据库
2. 通过配置LD_LIBRARY_PATH环境变量确保pg_config指向openGauss的安装路径

# 配置环境变量
$ export GAUSSHOME=/home/omm/openGauss-server/mppdb_temp_install
$ export LD_LIBRARY_PATH=$GAUSSHOME/lib:$LD_LIBRARY_PATH

$ which pg_config
# 预期结果
'/home/omm/openGauss-server/mppdb_temp_install/bin/pg_config'

# 删除psycopg2 for opengauss
$ pip install --force-reinstall psycopg2

# 若已经安装完成opengauss-sqlalchemy，重新安装
>>> python setup.py install

或者安装 psycopg2-binary：

$ pip uninstall psycopg2
$ pip install psycopg2-binary

# 修改数据库实例下postgresql.conf文件中password_encryption_type = 1
$ vim /datanode/postgresql.conf

# 重启数据库
$ gs_ctl restart -D /datanode

修改密码
$ gsql -d postgres -p <port_num>
openGauss=# alter user <usrname> with password <yourpassword>;

安装

# 从pypi安装
>>> python -m pip install opengauss-sqlalchemy

# 从本地源码安装
>>> python setup.py install

使用方法

在OpenGauss中创建一个数据库。

>>> import sqlalchemy as sa
# 访问集中式模式DB
>>> sa.create_engine('opengauss://username:password@host:port/database_name')
# 或
>>> sa.create_engine('opengauss+psycopg2://username:password@host:port/database_name')
# 访问分布式模式DB
>>> sa.create_engine('opengauss+dc_psycopg2://username:password@host:port/database_name')
# 或
>>> sa.create_engine('opengauss+dc_psycopg2://username:password@/database_name?host=hostA:portA&host=hostB:portB')

2.4.0版本新增了异步连接接口。需要注意的是，由于asyncpg的特殊性，当使用asyncpg连接并操作vector类型时，需手动注册该数据类型。具体实现示例如下：

from opengauss_sqlalchemy.register_async import register_vector
engine = create_async_engine("opengauss+asyncpg://username:password@host:port/database_name")

# 绑定事件监听器
@event.listens_for(engine.sync_engine, "connect")
def connect(dbapi_connection, connection_record):
    dbapi_connection.run_async(register_vector)

OpenGauss的数据库开发指南详见 OpenGauss DeveloperGuide。

OpenGauss特性的使用方式（集中式和分布式）

索引

• 使用USING method创建索引

tbl = Table("testtbl", m, Column("data", String))
Index("test_idx1", tbl.c.data, opengauss_using="btree")

• 带列表达式的索引

tbl = Table(
    "testtbl",
    m,
    Column("data", String),
    Column("data2", Integer, key="d2"),
)

Index(
    "test_idx1",
    tbl.c.data,
    tbl.c.d2,
    opengauss_ops={"data": "text_pattern_ops", "d2": "int4_ops"},
)

• 使用 LOCAL 创建索引，仅适用于分区表上的索引

tbl = Table(
    "testtbl",
    m,
    Column("data", Integer),
    opengauss_partition_by="RANGE (data) ..."
)
Index("test_idx1", tbl.c.data, opengauss_local=[""])

Index(
    "test_idx2",
    tbl.c.data,
    opengauss_local=[
        "PARTITION data_index1",
        "PARTITION data_index2 TABLESPACE example3",
    ]
)

• 使用 WITH 创建索引

tbl = Table("testtbl", m, Column("data", String))
Index("test_idx1", tbl.c.data, opengauss_with={"fillfactor": 50})

• 使用 TABLESPACE 创建索引

tbl = Table("testtbl", m, Column("data", String))
Index("test_idx1", tbl.c.data, opengauss_tablespace="sometablespace")

• 带 WHERE 条件的索引，不支持在分区表上创建

tbl = Table("testtbl", m, Column("data", Integer))
Index(
    "test_idx1",
    tbl.c.data,
    opengauss_where=and_(tbl.c.data > 5, tbl.c.data < 10),
)

表

• 使用 WITH ({storage_parameter = value}) 参数定义的表

Table("some_table", ..., opengauss_with={"storage_parameter": "value"})

• 带 ON COMMIT 选项的表

Table("some_talbe", ..., prefixes=["TEMPORARY"], opengauss_on_commit="PRESERVE ROWS")

• 带有 COMPRESS 属性的表

Table("some_talbe", ..., opengauss_with={"ORIENTATION": "COLUMN"}, opengauss_compress=True)

• 带有 TABLESPACE tablespace_name 的表

Table("some_talbe", ..., opengauss_tablespace="tablespace_name")

• 使用 PARTITION BY 子句的表

Table("some_talbe", ..., opengauss_partition_by="RANGE(column_name) "
        "(PARTITION P1 VALUES LESS THAN(10), "
        "PARTITION P2 VALUES LESS THAN(MAXVALUE))")

• 启用行迁移功能的表

Table("some_talbe", ..., opengauss_partition_by="RANGE(column_name) ...",
    opengauss_enable_row_movement=True)

向量数据类型

• 使用 select 查询向量数据类型

tbl = Table(
    "test",
    m,
    Column("id", Integer),
    Column("bit_embedding", BIT(3)),
    Column("sparsevec_embedding", SPARSEVEC(3)),
    Column("vector_embedding", VECTOR(3))
)

select(tbl.c.id).order_by(tbl.c.bit_embedding.jaccard_distance('110'))

select(tbl.c.id).order_by(tbl.c.sparsevec_embedding.l2_distance(SparseVector([1, 2, 3])))

select(tbl.c.id).order_by(tbl.c.vector_embedding.l2_distance([1,2,3]))

OpenGauss 特性的使用方式（集中式）

索引

• 使用 CONCURRENTLY 创建索引

tbl = Table("testtbl", m, Column("data", Integer))
Index("test_idx1", tbl.c.data, opengauss_concurrently=True)

OpenGauss特性的使用方式（分布式）

表

• 使用DISTRIBUTE BY创建分布式表

Table("some_table", ..., opengauss_distribute_by="HASH(column_name)")

注意：未指定分布列的表将默认设置为"DISTRIBUTE BY REPLICATION"（按复制分布）

• 使用TO GROUP语法创建的表

Table("some_table", ..., opengauss_to="GROUP group_name")

发布指南

构建 Python Wheel 包

>>> python -m pip install wheel
>>> python setup.py bdist_wheel

本地测试

1. 设置环境变量 export LD_LIBRARY_PATH= 和 export PYTHONPATH= 的值为测试环境中 psycopg2 包所在的目录。
2. 安装 OpenGauss 并修改数据库配置，具体步骤参见“安装并配置 OpenGauss 调测环境”。
3. 执行命令 tox -e py38。

安装并配置 OpenGauss 调测环境

1. 添加 OpenGauss 操作系统用户 >>> useradd omm -g dbgrp
2. 修改 OpenGauss 目录的用户和用户组 >>> chown omm:dbgrp ${db_dir} -R
3. 切换到新的系统用户 >>> su - omm
4. 安装 OpenGauss >>> sh install.sh -w ${db_password} -p 37200
5. 启动 OpenGauss >>> gs_ctl start -D ${db_dir}/data/single_node/
6. 登录 OpenGauss >>> gsql -d postgres -p 37200
7. 创建数据库用户、测试数据库和测试模式

openGauss=# create user scott identified by 'Tiger123';
openGauss=# create database test with owner=scott encoding='utf8' template=template0;
openGauss=# GRANT ALL PRIVILEGES TO scott;
openGauss=# ALTER DATABASE test SET default_text_search_config = 'pg_catalog.english';
openGauss=# \c test
test=# create schema test_schema AUTHORIZATION scott;
test=# create schema test_schema_2 AUTHORIZATION scott;
test=# \q

8. 配置OpenGauss参数调整

>>> gs_guc set -D ${db_dir}/data/single_node/ -c "ssl=off"
>>> gs_guc set -D ${db_dir}/data/single_node/ -c "max_prepared_transactions = 100"
>>> gs_guc reload -D ${db_dir}/data/single_node/ -h "local all scott sha256"
>>> gs_guc reload -D ${db_dir}/data/single_node/ -h "host all scott 127.0.0.1/32 sha256"
>>> gs_guc reload -D ${db_dir}/data/single_node/ -h "host all scott 0.0.0.0/0 sha256"
>>> gs_ctl stop -D ${db_dir}/data/single_node/
>>> gs_tl start -D ${db_dir}/data/single_node/

9. 启用SQL日志记录（可选）

>>> gs_guc set -D ${db_dir}/data/single_node/ -c "log_min_error_statement = error"
>>> gs_guc set -D ${db_dir}/data/single_node/ -c "log_statement = 'all'"