Python中使用SQLAlchemy操作GBase 8s数据库

gbase8s-sqlalchemy 介绍     

SQLAlchemy是Python中非常流行的ORM（对象关系映射）框架,提供了高层抽象来管理数据库交互。通过SQLAlchemy可以省去很多手动管理数据库连接、资源、事务等重复工作,让开发者更加高效地使用数据库。

GBase 8s技术团队提供了SQLAlchemy方言包gbase8s-sqlalchemy，通过它，您可以使用SQLAlchemy管理GBase 8s数据库。它需要SQLAlchemy 2.0及以上版本，Python版本需要Python 3.7或更高版本，支持数据库GBase 8s V8.8_3.6.2版本及以上。

1 安装

 pip install gbase8s-sqlalchemy

2 使用

2.1 配置GSDK

方言包连接数据库依赖GBase 8s数据库的Python驱动gbase8sdb（安装方言包时会作为依赖安装），而驱动连接数据库依赖GSDK 1.1版本 ，所以您需要联系GBase 8s技术支持或通过官方渠道获取相应版本的GSDK，并安装到您的机器上， 并设置如下环境变量：

GSDK_PATH=/path/to/gsdk
export LD_LIBRARY_PATH=${GSDK_PATH}/lib:$LD_LIBRARY_PATH

2.2 创建引擎

from sqlalchemy import create_engine, text

username = 'gbasedbt'
password = 'xxxxxx'
ip = '192.168.xxx.xxx'
port = 9088
dbname = 'testdb'
servername = 'gbase8s01'

# 声明url
url = f'gbase8s+gbase8sdb://{username}:{password}@{ip}:{port}/{dbname}?GBASEDBTSERVER={servername}&DB_LOCALE=zh_CN.utf8'
# 或使用sqlhosts文件连接
# url = f'gbase8s+gbase8sdb://{username}:{password}@{dbname}?GBASEDBTSERVER={servername}&DB_LOCALE=zh_CN.utf8&SQLH_FILE=/path/to/sqlhosts'

# 创建引擎
engine = create_engine(url, echo=True)  # echo参数设置为True用于打印执行的sql语句

2.2 验证连接

with engine.connect() as connection:
    result = connection.execute(text("SELECT 1 FROM DUAL"))
    print(result.fetchone()) 	# 输出(1,)

2.3 声明模型

from typing import List
from typing import Optional
from sqlalchemy import ForeignKey
from sqlalchemy import String
from sqlalchemy.orm import DeclarativeBase
from sqlalchemy.orm import Mapped
from sqlalchemy.orm import mapped_column
from sqlalchemy.orm import relationship

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = "user_account"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(30))
    fullname: Mapped[Optional[str]] = mapped_column(String(100))
    addresses: Mapped[List["Address"]] = relationship(
        back_populates="user", cascade="all, delete-orphan"
    )
    def __repr__(self) -> str:
        return f"User(id={self.id!r}, name={self.name!r}, fullname={self.fullname!r})"

class Address(Base):
    __tablename__ = "address"
    id: Mapped[int] = mapped_column(primary_key=True)
    email_address: Mapped[str] = mapped_column(String(100))
    user_id: Mapped[int] = mapped_column(ForeignKey("user_account.id"))
    user: Mapped["User"] = relationship(back_populates="addresses")
    def __repr__(self) -> str:
        return f"Address(id={self.id!r}, email_address={self.email_address!r})"

2.4 创建模型

Base.metadata.create_all(engine)

2.5 插入数据

from sqlalchemy.orm import Session

with Session(engine) as session:
    spongebob = User(
        name="spongebob",
        fullname="Spongebob Squarepants",
        addresses=[Address(email_address="spongebob@sqlalchemy.org")],
    )
    sandy = User(
        name="sandy",
        fullname="Sandy Cheeks",
        addresses=[
            Address(email_address="sandy@sqlalchemy.org"),
            Address(email_address="sandy@squirrelpower.org"),
        ],
    )
    patrick = User(name="patrick", fullname="Patrick Star")
    session.add_all([spongebob, sandy, patrick])
    session.commit()

2.7 简单的查询

from sqlalchemy import select

session = Session(engine)
stmt = select(User).where(User.name.in_(["spongebob", "sandy"]))
for user in session.scalars(stmt):
    print(user)