您现在的位置：程式師世界 >> 編程語言 > >> 更多編程語言 >> Python

Python SQLalchemy 基礎操作之數據庫增刪改查

編輯：Python

ORM全稱ObjectRelationalMapping,即對象關系映射。簡潔的說，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_engine
engine = create_engine('sqlite:///AiTestOps.db')

以絕對路徑形式創建數據庫，格式如下：

from sqlalchemy import create_engine
engine = 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')
# psycopg2
engine = create_engine('postgresql+psycopg2://username:[email protected]:9527/AiTestOps')
# pg8000
engine = 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-python
engine = create_engine('mysql+mysqldb://username:[email protected]:9527/AiTestOps')
# MySQL-connector-python
engine = 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_oracle
engine = create_engine('oracle+cx_oracle://username:[email protected]:9527/AiTestOps')

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

from sqlalchemy import create_engine
engine = 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_base
Base = declarative_base()

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

from sqlalchemy import Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
Base = declarative_base()
# 定義映射類Person，並繼承 Base
class 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, String
from sqlalchemy.ext.declarative import declarative_base
def 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_Model
if __name__ == '__main__':
TestModel = table_name_model_class("Person_Info")
print(TestModel.__table__)

創建數據表

查看映射對應的表

Person.__table__

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

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

[外鏈圖片轉存失敗,源站可能有防盜鏈機制,建議將圖片保存下來直接上傳(img-ASxifLgM-1657718256915)(https://juejin.cn/ “點擊並拖拽以移動”)]

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

創建指定表

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

建立會話

from sqlalchemy.orm import sessionmaker
from 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, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from sqlalchemy import create_engine
def 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_Model
if __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, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from sqlalchemy import create_engine
def 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_Model
if __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 xxx
filter/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 like
data_like = session.query(Person).filter(Person.desc.notlike("活%")).all()

[外鏈圖片轉存失敗,源站可能有防盜鏈機制,建議將圖片保存下來直接上傳(img-bauMzlRD-1657718256923)(https://juejin.cn/ “點擊並拖拽以移動”)]

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就是關鍵字not
more_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 func
res = 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
# count
result = session.query(Person.password, func.count(Person.id)).group_by(Person.password).all()
# sum
result = session.query(Person.password, func.sum(Person.id)).group_by(Person.password).all()
# max
result = session.query(Person.password, func.max(Person.id)).group_by(Person.password).all()
# min
result = session.query(Person.password, func.min(Person.id)).group_by(Person.password).all()
# having
result = 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, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from sqlalchemy import create_engine
def 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_Model
if __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, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from sqlalchemy import create_engine
def 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_Model
if __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)

[外鏈圖片轉存失敗,源站可能有防盜鏈機制,建議將圖片保存下來直接上傳(img-ks55u4T3-1657718256936)(https://juejin.cn/ “點擊並拖拽以移動”)]

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

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