SQLAlchemy框架支持

SQLAlchemy框架支持

本文档描述了 SeaboxSQL 数据库对于SQLAlchemy 框架的支持情况和配置说明，面向所有使用SeaboxSQL数据库的用户，主要是数据库管理员和应用程序开发人员。

有关 SQLAlchemy 的更多信息，请参阅以下资源：SQLAlchemy 1.3 Documentation https://docs.sqlalchemy.org/en/13/dialects/postgresql.html

概述

​SQLAlchemy 是一个基于Python实现的ORM框架，该框架建立在 Python DB API 之上，使用关系对象映射进行数据库操作。 ​ SQLAlchemy 本身无法操作数据库，其必须使用第三方插件，Dialect用于和数据库交互，根据配置的不同而调用不同的数据库 API 从而实现对数据库的操作。

​说明：可使用postgresql的方式连接集群。包括相关的方言包。

SQLAlchemy 使用说明

配置信息

​安装sqlalchemy和psycopg2插件

pip install sqlalchemy
pip install psycopg2

可以使用以下语句连接到数据库：

engine_url = URL.create( 'postgresql+psycopg2', username='用户名', password='密码', host='192.168.2.168', port=7300, database='ssbm' ) engine = create_engine(engine_url)

连接串格式为dialect+driver://username:password@host:port/database

执行普通SQL

连接数据库执行普通sql

from sqlalchemy import create_engine, URL,text

# 使用URL对象来安全地传递数据库连接信息
engine_url = URL.create(
    'postgresql+psycopg2',
    username='用户名',
    password='密码',
    host='192.168.2.168',
    port=7300,
    database='ssbm'
)
engine = create_engine(engine_url)
 
# 之后的操作相同
with engine.connect() as conn:
    result = conn.execute(text("SELECT * FROM public.sjltest"))
    for row in result:
        print(row)

ORM 基本操作

from sqlalchemy import Table, Column, Integer, String, MetaData,URL,create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import Session

#创建连接
engine_url = URL.create(
    'postgresql+psycopg2',
    username='sjl',
    password='sjl',
    host='192.168.2.168',
    port=7300,
    database='ssbm'
)
engine = create_engine(engine_url)
metadata = MetaData()
Base = declarative_base()

#表映射实体
class User(Base):
   __tablename__ = 'users'

   id = Column(Integer, primary_key=True)
   name = Column(String)
   email = Column(String)


# drop 表
Base.metadata.drop_all(engine)

# create 表
Base.metadata.create_all(engine)


# 创建一个新的 Session
session = Session(engine)
# 创建一个新的 User 对象
new_user = User(name='John', email='john@example.com')
# 添加到 session
session.add(new_user)

#select
results = session.query(User).all()

for r in results:
    print(r.id,r.name,r.email)

#update
session.query(User).filter(User.id == 1).update({"email": User.email + "(append data1)"},synchronize_session=False)


# 提交（保存）更改
session.commit()

#select
results = session.query(User).all()

for r in results:
    print(r.id,r.name,r.email)

#关闭session
session.close()