SQLAlchemy 框架

SQLAlchemy是Python編程語言下的一款開源軟件。提供了SQL工具包及對象關(guān)系映射（ORM）工具，使用MIT許可證發(fā)行。

SQLAlchemy采用簡單的Python語言，為高效和高性能的數(shù)據(jù)庫訪問設(shè)計，實現(xiàn)了完整的企業(yè)級持久模型。SQLAlchemy的理念是，SQL數(shù)據(jù)庫的量級和性能重要于對象集合；而對象集合的抽象又重要于表和行。因此，SQLAlchemy采用了類似于Java里Hibernate的數(shù)據(jù)映射模型，而不是其他ORM框架采用的Active Record模型。不過，Elixir和declarative等可選插件可以讓用戶使用聲明語法。

SQLAlchemy 是python中，通過ORM操作數(shù)據(jù)庫的框架。簡單點來說，就是幫助我們從煩冗的sql語句中解脫出來，從而不需要再去寫原生的sql語句，只需要用python的語法來操作對象，就能被自動映射為sql語句。

SQLAlchemy首次發(fā)行于2006年2月，并迅速地在Python社區(qū)中成長為最廣泛使用的ORM工具之一，不亞于Django的ORM框架。

它有幾個不同的組件，可以單獨使用或組合在一起。其主要組件依賴關(guān)系組織如下圖所示：

img

Schema / Types 類到表之間的映射規(guī)則
SQL Expression Language SQL 語句
Engine 引擎
Connection Pooling 連接池
Dialect 方言，調(diào)用不同的數(shù)據(jù)庫 API(Oracle, postgresql, Mysql) 并執(zhí)行對應(yīng)的 SQL語句

具體用法參見技術(shù)手冊

安裝插件

1、安裝psycopg2，參見上節(jié)：psycopg2 接口

2、安裝SQLAlchemy

pip install SQLAlchemy

#測試一下sqlalchemy安裝成功沒有
import sqlalchemy
sqlalchemy.__version__

? 輸出為

image-20220507162634496

示例項目

項目結(jié)構(gòu)圖如下：

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


# 創(chuàng)建連接信息:
host = "192.168.2.5"
db = "test" 
user = "test"
#在sqlalchemy鏈接數(shù)據(jù)庫時密碼中有@等特殊字符，
#會報錯“could not translate host name "123@192.168.2.5" to address: Unknown server error”
password = "test"  
pwd = parse.quote_plus(password) 
print(pwd)
# 創(chuàng)建對象的基類:
Base = declarative_base() 

# 初始化數(shù)據(jù)庫連接:
DB_CONNECT = f'postgresql+psycopg2://{user}:{pwd}@{host}:5866/{db}'
print(DB_CONNECT)

engine = create_engine(DB_CONNECT)  # 用戶名:密碼@localhost:端口/數(shù)據(jù)庫名

# 創(chuàng)建DBSession類型:
DBSession = sessionmaker(bind=engine) 

# 定義User對象:
class User(Base):
    # 表的名字:
    __tablename__ = 'users' 
    # 表的結(jié)構(gòu):
    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))

# 創(chuàng)建表
def createTable():
    try:
        Base.metadata.create_all(engine)     
        print("創(chuàng)建表成功!")
    except SQLAlchemyError as e:
        error = str(e.__dict__['orig'])
        print(error)
    
    
# 插入操作
def insertData():
    # 創(chuàng)建會話
    session = DBSession()
    # 創(chuàng)建新User對象:
    local_time = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
    new_user = User(name='hsh4', sex='男', nation='漢', birth='19991012', id_address='廣州大學(xué)', id_number='441242142142',
                    create_time=local_time)
    # 添加到session:
    session.add(new_user)
    # 提交即保存到數(shù)據(jù)庫:
    session.commit()
    # 關(guān)閉session:
    session.close()  
         
# 查詢操作
def selectData():     
    session = DBSession() # 創(chuàng)建Session
    # 創(chuàng)建Query查詢，filter是where條件，最后調(diào)用one()返回唯一行，如果調(diào)用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()  # 關(guān)閉Session
    
# 更新操作
def updateData():    
    session = DBSession()  # 創(chuàng)建會話
    users = session.query(User).filter_by(name="hsh4").first()  # 查詢條件
    users.id_number = "abcd"  # 更新操作
    session.add(users)  # 添加到會話
    session.commit()  # 提交即保存到數(shù)據(jù)庫
    session.close()  # 關(guān)閉會話

# 刪除操作
def deleteData():    
    session = DBSession()  # 創(chuàng)建會話
    delete_users = session.query(User).filter(User.id == "1").first()
    if delete_users:
        session.delete(delete_users)
        session.commit()
    session.close()  # 關(guān)閉會話
 
# 刪除表
def dropTable():    
    session = DBSession()  # 創(chuàng)建會話
    session.execute('drop table users')
    session.commit()
    session.close()
 
 
createTable()
# insertData()
# selectData()
# updateData()
# deleteData()
# dropTable()

執(zhí)行結(jié)果：

image-20220517160715939

在數(shù)據(jù)庫中查看，users表已經(jīng)創(chuàng)建成功

image-20220507173104321

注意事項

運行過程中會出現(xiàn)如下錯誤：

image-20220507173615467

1、根據(jù)提示進(jìn)入到site-packages\sqlalchemy\dialects\postgresql\base.py，定位到 _get_server_version_info函數(shù)，此函數(shù)的功能是獲取Postgresql的版本信息，而HGDB的版本信息不能被正則表達(dá)式獲取到，所以會出現(xiàn)如上的錯誤。

image-20220507173934756

2、解決：修改3585行將select pg_catalog.version()改為select 'PostgreSQL 12.7'

說明：最新的HGDB企業(yè)版V6.0.4 和安全版V4.5.7 的內(nèi)核基于PG12.7

image-20220507174030335