产品指南
生态对接
开发工具
Sqlalchemy

使用SQLAlchemy开发ProtonBase应用示例

ProtonBase 支持基于主流 ORM（Object-Relational Mapping）框架开发应用，SQLAlchemy 是 Python 语言下的对象关系映射解决方案。本文介绍基于 SqlAlchemy 如何快速开发。

准备工作

  1. 在 ProtonBase 上创建 database
CREATE DATABASE example;
  1. 在新创建的 database 上创建 table
CREATE TABLE customer (
    id bigint NOT NULL,
    name text NOT NULL,
    email text NOT NULL,
    create_time timestamp DEFAULT CURRENT_TIMESTAMP,
    PRIMARY KEY (id)
);

SQLAlchemy 示例

SQLAlchemy 是 Python 语言下的对象关系映射解决方案。

  1. 引入 SQLAlchemy 依赖。
pip install SQLAlchemy
  1. 定义实体类，每一张表都需要对应一个实体类。这里我们只有一张 customer 表，只需要编写一个 Customer 类即可。在定义实体类时，我们需要指明字段与数据库列的对应关系。
class Base(DeclarativeBase):
pass
 
class Customer(Base):
__tablename__ = "customer"
 
id: Mapped[int] = mapped_column(primary_key=True)
name: Mapped[str] = mapped_column(Text, nullable=False)
email: Mapped[str] = mapped_column(Text, nullable=False)
create_time: Mapped[datetime] = mapped_column(DATETIME, default=datetime.now)
 
def __str__(self):
    return f"id: {self.id}, name: {self.name}, email: {self.email}, create_time: {self.create_time}"
  1. 配置数据库连接参数
host = "localhost"
port = 5432
username = "platformadmin"
password = "pOoFOBA1o!P7DfJaLx!"
database = "example"
  1. 通过以下代码，先向 customer 表中插入（insert）两条新纪录，然后查询（select）cluster 表验证插入已生效；再修改（update）其中的一条记录，查询表中记录验证修改已生效；最后删除其中一条记录，查询表中记录验证删除已生效。
def insert_func(db_engine):
    with Session(db_engine) as session:
        first_customer = Customer(id=1, name="Jacob Emily", email="jacob.emily@protonbase.io")
        session.add(first_customer)
        second_customer = Customer(id=2, name="Michael Emma", email="michael.emma@protonbase.io")
        session.add(second_customer)
        session.commit()
 
def delete_func(db_engine):
    with Session(db_engine) as session:
        customer = session.get(Customer, 1)
        session.delete(customer)
        session.commit()
 
def update_func(db_engine):
    with Session(db_engine) as session:
        stmt = select(Customer).where(Customer.id == 2)
        customer = session.scalars(stmt).one()
        customer.email = "michael.emma@gmail.com"
        session.commit()
 
def select_func(db_engine):
    with Session(db_engine) as session:
        stmt = select(Customer).where(Customer.id.in_([1, 2]))
        for customer in session.scalars(stmt):
        print(customer)
 
if __name__ == '__main__':
    engine = create_engine(f"postgresql://{username}:{password}@{host}:{port}/{database}?sslmode=disable", echo=False)
 
# Insert
insert_func(engine)
 
# Select
print("Result of the first selection:")
select_func(engine)
 
# Update
update_func(engine)
 
# Select
print("Result of the second selection:")
select_func(engine)
 
# Delete
delete_func(engine)
 
# Select
print("Result of the third selection:")
select_func(engine)

运行结果如下：

Result of the first selection:
id: 1, name: Jacob Emily, email: jacob.emily@protonbase.io, create_time: 2023-10-29 00:46:18.951433
id: 2, name: Michael Emma, email: michael.emma@protonbase.io, create_time: 2023-10-29 00:46:18.951441
Result of the second selection:
id: 1, name: Jacob Emily, email: jacob.emily@protonbase.io, create_time: 2023-10-29 00:46:18.951433
id: 2, name: Michael Emma, email: michael.emma@gmail.com, create_time: 2023-10-29 00:46:18.951441
Result of the third selection:
id: 2, name: Michael Emma, email: michael.emma@gmail.com, create_time: 2023-10-29 00:46:18.951441
HibernateBI工具