SQLAlchemy 框架

SQLAlchemy是Python编程语言下的一款开源软件。提供了SQL工具包及对象关系映射（ORM）工具，使用MIT许可证发行。

SQLAlchemy采用简单的Python语言，为高效和高性能的数据库访问设计，实现了完整的企业级持久模型。SQLAlchemy的理念是，SQL数据库的量级和性能重要于对象集合；而对象集合的抽象又重要于表和行。因此，SQLAlchemy采用了类似于Java里Hibernate的数据映射模型，而不是其他ORM框架采用的Active Record模型。不过，Elixir和declarative等可选插件可以让用户使用声明语法。

SQLAlchemy 是python中，通过ORM操作数据库的框架。简单点来说，就是帮助我们从烦冗的sql语句中解脱出来，从而不需要再去写原生的sql语句，只需要用python的语法来操作对象，就能被自动映射为sql语句。

SQLAlchemy首次发行于2006年2月，并迅速地在Python社区中成长为最广泛使用的ORM工具之一，不亚于Django的ORM框架。

它有几个不同的组件，可以单独使用或组合在一起。其主要组件依赖关系组织如下图所示：

img

Schema / Types 类到表之间的映射规则
SQL Expression Language SQL 语句
Engine 引擎
Connection Pooling 连接池
Dialect 方言，调用不同的数据库 API(Oracle, postgresql, Mysql) 并执行对应的 SQL语句

具体用法参见技术手册

安装插件

1、安装psycopg2，参见上节：psycopg2 接口

2、安装SQLAlchemy

pip install SQLAlchemy

#测试一下sqlalchemy安装成功没有
import sqlalchemy
sqlalchemy.__version__

​ 输出为

image-20220507162634496

示例项目

项目结构图如下：

image-20220517160115566

sqlalchemy_conn.py

from sqlalchemy import Column, String, create_engine, Integer, Text
from sqlalchemy.exc import SQLAlchemyError
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from urllib import parse

import time


# 创建连接信息:
host = "192.168.2.5"
db = "test" 
user = "test"
#在sqlalchemy链接数据库时密码中有@等特殊字符，
#会报错“could not translate host name "123@192.168.2.5" to address: Unknown server error”
password = "test"  
pwd = parse.quote_plus(password) 
print(pwd)
# 创建对象的基类:
Base = declarative_base() 

# 初始化数据库连接:
DB_CONNECT = f'postgresql+psycopg2://{user}:{pwd}@{host}:5866/{db}'
print(DB_CONNECT)

engine = create_engine(DB_CONNECT)  # 用户名:密码@localhost:端口/数据库名

# 创建DBSession类型:
DBSession = sessionmaker(bind=engine) 

# 定义User对象:
class User(Base):
    # 表的名字:
    __tablename__ = 'users' 
    # 表的结构:
    id = Column(Integer, autoincrement=True, primary_key=True, unique=True, nullable=False)
    name = Column(String(50), nullable=False)
    sex = Column(String(4), nullable=False)
    nation = Column(String(20), nullable=False)
    birth = Column(String(8), nullable=False)
    id_address = Column(Text, nullable=False)
    id_number = Column(String(18), nullable=False)
    creater = Column(String(32))
    create_time = Column(String(20), nullable=False)
    updater = Column(String(32))
    update_time = Column(String(20), nullable=False, default=time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()), onupdate=time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()))
    comment = Column(String(200))

# 创建表
def createTable():
    try:
        Base.metadata.create_all(engine)     
        print("创建表成功!")
    except SQLAlchemyError as e:
        error = str(e.__dict__['orig'])
        print(error)
    
    
# 插入操作
def insertData():
    # 创建会话
    session = DBSession()
    # 创建新User对象:
    local_time = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
    new_user = User(name='hsh4', sex='男', nation='汉', birth='19991012', id_address='广州大学', id_number='441242142142',
                    create_time=local_time)
    # 添加到session:
    session.add(new_user)
    # 提交即保存到数据库:
    session.commit()
    # 关闭session:
    session.close()  
         
# 查询操作
def selectData():     
    session = DBSession() # 创建Session
    # 创建Query查询，filter是where条件，最后调用one()返回唯一行，如果调用all()则返回所有行:
    user = session.query(User).filter(User.id == '1' and User.name == 'hsh4').one()
    print('name:', user.name)
    print('id_address:', user.id_address)
    session.close()  # 关闭Session
    
# 更新操作
def updateData():    
    session = DBSession()  # 创建会话
    users = session.query(User).filter_by(name="hsh4").first()  # 查询条件
    users.id_number = "abcd"  # 更新操作
    session.add(users)  # 添加到会话
    session.commit()  # 提交即保存到数据库
    session.close()  # 关闭会话

# 删除操作
def deleteData():    
    session = DBSession()  # 创建会话
    delete_users = session.query(User).filter(User.id == "1").first()
    if delete_users:
        session.delete(delete_users)
        session.commit()
    session.close()  # 关闭会话
 
# 删除表
def dropTable():    
    session = DBSession()  # 创建会话
    session.execute('drop table users')
    session.commit()
    session.close()
 
 
createTable()
# insertData()
# selectData()
# updateData()
# deleteData()
# dropTable()

执行结果：

image-20220517160715939

在数据库中查看，users表已经创建成功

image-20220507173104321

注意事项

运行过程中会出现如下错误：

image-20220507173615467

1、根据提示进入到site-packages\sqlalchemy\dialects\postgresql\base.py，定位到 _get_server_version_info函数，此函数的功能是获取Postgresql的版本信息，而HGDB的版本信息不能被正则表达式获取到，所以会出现如上的错误。

image-20220507173934756

2、解决：修改3585行将select pg_catalog.version()改为select 'PostgreSQL 12.7'

说明：最新的HGDB企业版V6.0.4 和安全版V4.5.7 的内核基于PG12.7

image-20220507174030335