ORM 全稱 Object Relational Mapping, 即物件關係對映。簡單的說，ORM 將資料庫中的表與面嚮物件語言中的類建立了一種對應關係。這樣，我們要操作資料庫，資料庫中的表或者表中的一條記錄就可以直接透過操作類或者類例項來完成。

SQLAlchemy 是Python 社群最知名的 ORM 工具之一，為高效和高效能的資料庫訪問設計，實現了完整的企業級持久模型。

SQLAlchemy 優點：

簡潔易讀：將資料表抽象為物件（資料模型），更直觀易讀。可移植：封裝了多種資料庫引擎，面對多個數據庫，操作基本一致，程式碼易維護。更安全：有效避免SQL注入。

本文透過介紹Sqlite資料庫的常用操作，來介紹一下SQLAlchemy 的使用方法。SQLAlchemy 具體的實現方式是將資料庫錶轉換為Python類，其中資料列作為屬性，資料庫操作作為方法。

SQLAlchem安裝

Sqlite3是Python3標準庫不需要另外安裝，只需要安裝SQLAlchemy即可。

pip install sqlalchemy

ORM 建立資料庫連線

Sqlite3 建立資料庫連線就是建立資料庫，而其他MySQL等資料庫，需要資料庫已存在，才能建立資料庫連線。

SQLite

以相對路徑形式，在當前目錄下建立資料庫格式如下：

from sqlalchemy import create_engineengine = create_engine('sqlite:///AiTestOps.db')

以絕對路徑形式建立資料庫，格式如下：

from sqlalchemy import create_engineengine = create_engine('sqlite:///G:\python_sql\AiTestOps.db')

其它常用資料庫的建立資料庫連線方法

SQLAlchemy用一個字串表示連線資訊：

'資料庫型別+資料庫驅動名稱://使用者名稱:密碼@IP地址:埠號/資料庫名'

PostgreSQL資料庫

from sqlalchemy import create_engine# default, 連線串格式為 "資料庫型別+資料庫驅動://資料庫使用者名稱:資料庫密碼@IP地址:埠/資料庫"engine = create_engine('postgresql://username:[email protected]:9527/AiTestOps')# psycopg2engine = create_engine('postgresql+psycopg2://username:[email protected]:9527/AiTestOps')# pg8000engine = create_engine('postgresql+pg8000://username:[email protected]:9527/AiTestOps')

MySQL資料庫

from sqlalchemy import create_engine# default,連線串格式為 "資料庫型別+資料庫驅動://資料庫使用者名稱:資料庫密碼@IP地址:埠/資料庫"engine = create_engine('mysql://username:[email protected]:9527/AiTestOps')# mysql-pythonengine = create_engine('mysql+mysqldb://username:[email protected]:9527/AiTestOps')# MySQL-connector-pythonengine = create_engine('mysql+mysqlconnector://username:[email protected]:9527/AiTestOps')

Oracle資料庫

from sqlalchemy import create_engine# default,連線串格式為 "資料庫型別+資料庫驅動://資料庫使用者名稱:資料庫密碼@IP地址:埠/資料庫"engine = create_engine('oracle://username:[email protected]:9527/AiTestOps')# cx_oracleengine = create_engine('oracle+cx_oracle://username:[email protected]:9527/AiTestOps')

我們以在當前目錄下建立SQLite資料庫為例，後續各步同使用此資料庫。我們在create_engine方法中補充了兩個引數。如下：

from sqlalchemy import create_engineengine = create_engine('sqlite:///AiTestOps.db?check_same_thread=False', echo=True)

echo：echo預設為False，表示不列印執行的SQL語句等較詳細的執行資訊，改為Ture表示讓其列印。check_same_thread：check_same_thread預設為 False，sqlite預設建立的物件只能讓建立該物件的執行緒使用，而sqlalchemy是多執行緒的，所以我們需要指定check_same_thread=False來讓建立的物件任意執行緒都可使用。定義對映（類與表的對映）

首先，我們建立基本對映類，後邊具體的對映類（表）需要繼承它。

from sqlalchemy.ext.declarative import declarative_baseBase = declarative_base()

然後，建立具體的對映類，我們這裡以Person對映類為例，我們把Person類對映到Person表。

from sqlalchemy import Column, Integer, Stringfrom sqlalchemy.ext.declarative import declarative_baseBase = declarative_base()# 定義對映類Person，並繼承 Baseclass Person(Base):    # 指定本類對映到 Person 表    __tablename__ = 'Person'    # 若有多個類指向同一張表，那麼在後邊的類需要把 extend_existing設為True，表示在已有列基礎上進行擴充套件    # 或者換句話說，sqlalchemy 允許類是表的字集，如下：    # __table_args__ = {'extend_existing': True}    # 若表在同一個資料庫服務（datebase）的不同資料庫中（schema），可使用schema引數進一步指定資料庫    # __table_args__ = {'schema': 'AiTestOps_database'}    # sqlalchemy 強制要求必須要有主鍵欄位不然會報錯，sqlalchemy在接收到查詢結果後還會自己根據主鍵進行一次去重，因此不要隨便設定非主鍵欄位設為primary_key    # 各變數名一定要與表的各欄位名一樣，因為相同的名字是他們之間的唯一關聯關係，指定 person_id 對映到 person_id 欄位; person_id 欄位為整型，為主鍵，自動增長（其實整型主鍵預設就自動增長）    person_id = Column(Integer, primary_key=True, autoincrement=True)    # 指定 username 對映到 username 欄位; username 欄位為字串類形，    # 指定 username 對映到 username 欄位; username 欄位為字串類形，    username = Column(String(20), nullable=False, index=True)    password = Column(String(32))    desc = Column(String(32))    # __repr__方法用於輸出該類的物件被print()時輸出的字串    def __repr__(self):        return "<User(username='%s', password='%s', desc='%s')>" % (            self.username, self.password, self.desc)

首先要明確下，ORM中一般情況下表是不需要先存在的，我們看到，在 Person 類中，用 __tablename__ 指定在 SQLite 中表的名字。

我們在Person中建立了三個欄位，類中的每一個 Column 代表資料庫中的一列（欄位），在 Colunm中，指定該列的一些屬性。第一個欄位代表資料型別，上面我們使用 String, Integer 兩個最常用的型別，其他常用的包括：Text、Boolean、SmallInteger、DateTime。

nullable=False 代表這一列不可以為空，index=True 表示在該列建立索引。另外，定義 __repr__ 是為了方便除錯。

在上面的Person類對映定義中，__tablename__屬性是靜態的，但有時我們可能想透過外部動態的給類傳遞表名，此時可以透過定義內部類進行傳參的方式來實現，如下：

from sqlalchemy import Column, Integer, Stringfrom sqlalchemy.ext.declarative import declarative_basedef table_name_model_class(table_name, Base=declarative_base()):    # 定義一個內部類    class User_Model(Base):        # 給表名賦值        __tablename__ = table_name        __table_args__ = {'extend_existing': True}        person_id = Column(Integer, primary_key=True, autoincrement=True)        # 指定 username 對映到 username 欄位; username 欄位為字串類形，        username = Column(String(20), nullable=False, index=True)        password = Column(String(32))        desc = Column(String(32))                def __repr__(self):            return "<User(username='%s', password='%s', desc='%s')>" % (                self.username, self.password, self.desc)    # 把動態設定表名的類返回去    return User_Modelif __name__ == '__main__':    TestModel = table_name_model_class("Person_Info")    print(TestModel.__table__)

建立資料表

檢視對映對應的表

Person.__table__

建立所有繼承於Base的類對應的表

Base.metadata.create_all(engine, checkfirst=True)

checkfirst預設為True，表示建立表前先檢查該表是否存在，若同名表已存在，則不再建立。

建立指定表

Base.metadata.create_all(engine, tables=[Base.metadata.tables['Person']], checkfirst=True)# 或者是Person.__table__.create(engine, checkfirst=True)

建立會話

from sqlalchemy.orm import sessionmakerfrom sqlalchemy import create_engine# 建立連結 engine = create_engine(r'sqlite:///AiTestOps.db?check_same_thread=False', echo=True)# 建立Session類物件Session = sessionmaker(bind=engine)# 建立Session類例項session = Session()

插入資料

from sqlalchemy import Column, Integer, Stringfrom sqlalchemy.ext.declarative import declarative_basefrom sqlalchemy.orm import sessionmakerfrom sqlalchemy import create_enginedef table_name_model_class(table_name, Base = declarative_base()):    # 定義一個內部類    class User_Model(Base):        # 給表名賦值        __tablename__ = table_name        __table_args__ = {'extend_existing': True}        person_id = Column(Integer, primary_key=True, autoincrement=True)        # 指定 username 對映到 username 欄位; username 欄位為字串類形，        username = Column(String(20))        password = Column(String(32))        desc = Column(String(32))                def __repr__(self):            return "<User(username='%s', password='%s', desc='%s')>" % (                self.username, self.password, self.desc)    # 把動態設定表名的類返回去    return User_Modelif __name__ == '__main__':    Person = table_name_model_class("Person")    # 建立連結    engine = create_engine(r'sqlite:///AiTestOps.db?check_same_thread=False', echo=True)    # 建立 Person 表    Person.__table__.create(engine, checkfirst=True)    # 建立Session類物件    Session = sessionmaker(bind=engine)    # 建立Session類例項    session = Session()    # 建立User類例項    jon_info = Person(username='Jon', password='123456', desc='活潑')    # 將該例項插入到 Person 表    session.add(jon_info)    # 一次插入多條記錄形式    session.add_all(        [        Person(username='Mark', password='123456', desc='活潑'),        Person(username='Tony', password='123456', desc='活潑')        ]    )    # 當前更改只是在session中，需要使用commit確認更改才會寫入資料庫    session.commit()

查詢資料

from sqlalchemy import Column, Integer, Stringfrom sqlalchemy.ext.declarative import declarative_basefrom sqlalchemy.orm import sessionmakerfrom sqlalchemy import create_enginedef table_name_model_class(table_name, Base = declarative_base()):    # 定義一個內部類    class User_Model(Base):        # 給表名賦值        __tablename__ = table_name        __table_args__ = {'extend_existing': True}        person_id = Column(Integer, primary_key=True, autoincrement=True)        # 指定 username 對映到 username 欄位; username 欄位為字串類形，        username = Column(String(20))        password = Column(String(32))        desc = Column(String(32))                def __repr__(self):            return "<User(username='%s', password='%s', desc='%s')>" % (                self.username, self.password, self.desc)    # 把動態設定表名的類返回去    return User_Modelif __name__ == '__main__':    Person = table_name_model_class("Person")    # 建立連結    engine = create_engine(r'sqlite:///AiTestOps.db?check_same_thread=False', echo=True)    # 建立 Person 表    Person.__table__.create(engine, checkfirst=True)    # 建立Session類物件    Session = sessionmaker(bind=engine)    # 建立Session類例項    session = Session()    # 一次插入多條記錄形式    session.add_all(        [        Person(username='Mark', password='123456', desc='活潑'),        Person(username='Tony', password='123456', desc='活潑')        ]    )    # 當前更改只是在session中，需要使用commit確認更改才會寫入資料庫    session.commit()        # 查詢 username='Mark' 的所有結果，返回結果物件    mark = session.query(Person).filter_by(username='Mark').all()    print(mark)    # 如果只獲取部分欄位，那麼返回的就是元組而不是物件了    mark_desc = session.query(Person.desc).filter_by(username='Mark').all()    print(mark_desc)

為了更好的理解 SQL 與 SQLalchemy 的寫法區別，可以參照以下內容：

query ：對應 SELECT xxx FROM xxxfilter/filter_by ：對應 WHERE ，fillter 可以進行比較運算(==, >, < ...)來對條件進行靈活的運用，不同的條件用逗號分割，fillter_by 只能指定引數傳參來獲取查詢結果。limit ：對應 limit()order by ：對應 order_by()group by ：對應 group_by()

like查詢

# like data_like = session.query(Person).filter(Person.desc.like("活%")).all()# not likedata_like = session.query(Person).filter(Person.desc.notlike("活%")).all()

is查詢

# is_ 相當於 ==result = session.query(Person).filter(Person.username.is_(None)).all()result = session.query(Person).filter(Person.username == None).all()# isnot 相當於 !=result = session.query(Person).filter(Person.username.isnot(None)).all()result = session.query(Person).filter(Person.username != None).all()

正則查詢

data_regexp = session.query(Person).filter(Person.password.op("regexp")(r"^[\u4e00-\u9fa5]+")).all()

統計數量

data_like_count = session.query(Person).filter(Person.desc.like("活%")).count()

IN 查詢

more_person = session.query(Person).filter(Person.username.in_(['Mark', 'Tony'])).all()

NOT IN 查詢

# ~代表取反,轉換成sql就是關鍵字notmore_person = session.query(Person).filter(~Person.username.in_(['Mark', 'Tony'])).all()# 或 notin_more_person = session.query(Person).filter(~Person.username.notin_(['Mark', 'Tony'])).all()

AND 查詢

from sqlalchemy import and_more_person = session.query(Person).filter(and_(Person.password=='123456',Person.desc=="可愛'")).all()

OR 查詢

from sqlalchemy import or_more_person = session.query(Person).filter(or_(Person.password=='123456',Person.desc=="活潑'")).all()

分組查詢

std_group_by = session.query(Person).group_by(Person.desc).all()# 或是from sqlalchemy.sql import funcres = session.query(Person.desc,                    func.count(Person.desc),                   ).group_by(Person.desc).all()# 遍歷檢視，已無ed使用者記錄for person in res:    print(person)

排序查詢

std_order_by = session.query(Person).order_by(Person.username.desc()).all()

limit 查詢

# limit 限制數量查詢, limit裡傳入一個整型來約束檢視的數量, 當limit裡面的引數大於例項表中的數量時，會返回所有的查詢結果data_limit = session.query(Person).filter(Person.desc.notlike("活%")).limit(1).all()

偏移量查詢

# offset 偏移量查詢,offset中傳入一個整型，從表中的該位置開始查詢，offset可以和limit混用來進行限制data_like = session.query(Person).filter(Person.desc.like("活%")).offset(1).all()result = session.query(Person).offset(1).limit(6).all()

聚合函式

from sqlalchemy import func, extract# countresult = session.query(Person.password, func.count(Person.id)).group_by(Person.password).all()# sumresult = session.query(Person.password, func.sum(Person.id)).group_by(Person.password).all()# maxresult = session.query(Person.password, func.max(Person.id)).group_by(Person.password).all()# minresult = session.query(Person.password, func.min(Person.id)).group_by(Person.password).all()# havingresult = session.query(Person.password, func.count(Person.id)).group_by(Person.password).having(func.count(Person.id) > 1).all()

關於返回結果數量

all()- 查詢所有- 返回一個列表物件first()- 查詢第一個符合條件的物件- 返回一個物件

關於傳參

filter = (Person.username=='Mark')our_user = session.query(Person).filter(filter).first()print(our_user)

更新

from sqlalchemy import Column, Integer, Stringfrom sqlalchemy.ext.declarative import declarative_basefrom sqlalchemy.orm import sessionmakerfrom sqlalchemy import create_enginedef table_name_model_class(table_name, Base = declarative_base()):    # 定義一個內部類    class User_Model(Base):        # 給表名賦值        __tablename__ = table_name        __table_args__ = {'extend_existing': True}        person_id = Column(Integer, primary_key=True, autoincrement=True)        # 指定 username 對映到 username 欄位; username 欄位為字串類形，        username = Column(String(20))        password = Column(String(32))        desc = Column(String(32))                def __repr__(self):            return "<User(username='%s', password='%s', desc='%s')>" % (                self.username, self.password, self.desc)    # 把動態設定表名的類返回去    return User_Modelif __name__ == '__main__':    Person = table_name_model_class("Person")    # 建立連結    engine = create_engine(r'sqlite:///AiTestOps.db?check_same_thread=False', echo=True)    # 建立 Person 表    Person.__table__.create(engine, checkfirst=True)    # 建立Session類物件    Session = sessionmaker(bind=engine)    # 建立Session類例項    session = Session()    # 一次插入多條記錄形式    session.add_all(        [            Person(username='Mark', password='123456', desc='活潑'),            Person(username='Tony', password='123456', desc='活潑')        ]    )    # 當前更改只是在session中，需要使用commit確認更改才會寫入資料庫    session.commit()        # 要修改需要先將記錄查出來    person = session.query(Person).filter_by(username='Mark').first()    # 將 Mark 使用者的密碼修改為 654321    person.password = '654321'    # 確認修改    session.commit()        our_user = session.query(Person.password).filter_by(username='Mark').all()    print(our_user)

上邊的操作，先進行查詢再修改，相當於執行了兩條語句，我們可直接使用如下方法

session.query(Person).filter_by(username='Mark').update({Person.password: '6543210'})session.commit()

以同schema的一張表更新另一張表的寫法，在跨表的update/delete等函式中要註明synchronize_session=False，否則報錯：

session.query(Person).filter_by(Person.username=Person1.username).update({Person.password: Person1.password}, synchronize_session=False)session.commit()

以一個schema的表更新另一個schema的表的寫法，寫法與同一schema的一樣，只是定義model時需要使用table_args = {'schema': 'test_Person'}等形式指定表對應的schema。

刪除

from sqlalchemy import Column, Integer, Stringfrom sqlalchemy.ext.declarative import declarative_basefrom sqlalchemy.orm import sessionmakerfrom sqlalchemy import create_enginedef table_name_model_class(table_name, Base = declarative_base()):    # 定義一個內部類    class User_Model(Base):        # 給表名賦值        __tablename__ = table_name        __table_args__ = {'extend_existing': True}        person_id = Column(Integer, primary_key=True, autoincrement=True)        # 指定 username 對映到 username 欄位; username 欄位為字串類形，        username = Column(String(20))        password = Column(String(32))        desc = Column(String(32))                def __repr__(self):            return "<User(username='%s', password='%s', desc='%s')>" % (                self.username, self.password, self.desc)    # 把動態設定表名的類返回去    return User_Modelif __name__ == '__main__':    Person = table_name_model_class("Person_Info")    # 建立連結    engine = create_engine(r'sqlite:///AiTestOps.db?check_same_thread=False', echo=True)    # 建立 Person 表    Person.__table__.create(engine, checkfirst=True)    # 建立Session類物件    Session = sessionmaker(bind=engine)    # 建立Session類例項    session = Session()    # 一次插入多條記錄形式    session.add_all(        [            Person(username='Mark', password='123456', desc='活潑'),            Person(username='Tony', password='123456', desc='活潑')        ]    )    # 當前更改只是在session中，需要使用commit確認更改才會寫入資料庫    session.commit()    mark = session.query(Person).filter_by(username='Mark').first()    # 將 mark 使用者記錄刪除    session.delete(mark)    # 確認刪除    session.commit()    # 遍歷檢視，已無 Mark 資料     for person in session.query(Person):        print(person.username)

或者，直接一步到位 ，不需要像上面那樣，先查詢出來，再執行刪除操作。

session.query(Person).filter(Person.username == "Mark").delete()session.commit()# 刪除 in 操作查詢出來的記錄,需要傳synchronize_session=False，否則會丟擲 qlalchemy.exc.InvalidRequestErrorsession.query(Person).filter(Person.desc.in_(['可愛', '活潑'])).delete(synchronize_session=False)