Python使用SQLAlchemy模塊實現(xiàn)操作數(shù)據(jù)庫
更新時間：2023年11月24日 11:35:07 作者：微軟技術(shù)分享
SQLAlchemy 是用Python編程語言開發(fā)的一個開源項目,它提供了SQL工具包和ORM對象關(guān)系映射工具,使用SQLAlchemy可以實現(xiàn)高效和高性能的數(shù)據(jù)庫訪問,下面我們就來學習一下SQLAlchemy模塊的具體應(yīng)用吧
SQLAlchemy 是用Python編程語言開發(fā)的一個開源項目，它提供了SQL工具包和ORM對象關(guān)系映射工具，使用MIT許可證發(fā)行，SQLAlchemy 提供高效和高性能的數(shù)據(jù)庫訪問，實現(xiàn)了完整的企業(yè)級持久模型。
ORM（對象關(guān)系映射）是一種編程模式，用于將對象與關(guān)系型數(shù)據(jù)庫中的表和記錄進行映射，從而實現(xiàn)通過面向?qū)ο蟮姆绞竭M行數(shù)據(jù)庫操作。ORM 的目標是在編程語言中使用類似于面向?qū)ο缶幊痰恼Z法，而不是使用傳統(tǒng)的 SQL 查詢語言，來操作數(shù)據(jù)庫。
主要思想是將數(shù)據(jù)庫表的結(jié)構(gòu)映射到程序中的對象，通過對對象的操作來實現(xiàn)對數(shù)據(jù)庫的操作，而不是直接編寫 SQL 查詢。ORM 工具負責將數(shù)據(jù)庫記錄轉(zhuǎn)換為程序中的對象，反之亦然。
ORM 的核心概念包括：
- 實體（Entity）： 在 ORM 中，實體是指映射到數(shù)據(jù)庫表的對象。每個實體對應(yīng)數(shù)據(jù)庫中的一條記錄。
- 屬性（Attribute）： 實體中的屬性對應(yīng)數(shù)據(jù)庫表中的列。每個屬性表示一個字段。
- 關(guān)系（Relationship）： ORM 允許定義實體之間的關(guān)系，例如一對多、多對一、多對多等。這種關(guān)系會映射到數(shù)據(jù)庫表之間的關(guān)系。
- 映射（Mapping）： ORM 負責將實體的屬性和方法映射到數(shù)據(jù)庫表的列和操作。
- 會話（Session）： ORM 提供了會話來管理對象的生命周期，包括對象的創(chuàng)建、更新和刪除。
- 查詢語言： ORM 通常提供一種查詢語言，允許開發(fā)者使用面向?qū)ο蟮姆绞骄帉懖樵儯皇侵苯邮褂?SQL。
對象映射ROM模型可連接任何關(guān)系數(shù)據(jù)庫，連接方法大同小異，以下總結(jié)了如何連接常用的幾種數(shù)據(jù)庫方式。
# sqlite 創(chuàng)建數(shù)據(jù)庫連接
engine = create_engine('sqlite:///database.db', echo=False)
# sqlite 創(chuàng)建內(nèi)存數(shù)據(jù)庫
engine = create_engine('sqlite://')
engine = create_engine('sqlite:///:memory:', echo=True)
# PostgreSQL 創(chuàng)建數(shù)據(jù)庫連接
engine = create_engine('postgresql://scott:tiger@localhost/mydatabase') # default
engine = create_engine('postgresql+psycopg2://scott:tiger@localhost/mydatabase') # psycopg2
engine = create_engine('postgresql+pg8000://scott:tiger@localhost/mydatabase') # pg8000
# MySQL 創(chuàng)建數(shù)據(jù)庫連接
engine = create_engine('mysql://scott:tiger@localhost/foo') # default
engine = create_engine('mysql+mysqldb://scott:tiger@localhost/foo') # mysql-python
engine = create_engine('mysql+mysqlconnector://scott:tiger@localhost/foo') # MySQL-connector-python
engine = create_engine('mysql+oursql://scott:tiger@localhost/foo') # OurSQL
# Oracle 創(chuàng)建數(shù)據(jù)庫連接
engine = create_engine('oracle://scott:tiger@127.0.0.1:1521/sidname')
engine = create_engine('oracle+cx_oracle://scott:tiger@tnsname')
# MSSQL 創(chuàng)建數(shù)據(jù)庫連接
engine = create_engine('mssql+pyodbc://scott:tiger@mydsn') # pyodbc
engine = create_engine('mssql+pymssql://scott:tiger@hostname:port/dbname') # pymssql
數(shù)據(jù)表創(chuàng)建
簡單的創(chuàng)建一個User映射類，映射到UserDB庫上，分別增加幾個常用的數(shù)據(jù)庫字段，并插入一些測試數(shù)據(jù)。
import sqlite3,datetime,time
from sqlalchemy.orm import sessionmaker
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, Float, DateTime, Date, Time, Boolean, DECIMAL, Enum, Text
# 建立基本映射類
Base = declarative_base()
# 創(chuàng)建SQLITE數(shù)據(jù)庫
engine = create_engine('sqlite:///:memory:', echo=False)
# 創(chuàng)建映射類User
class User(Base):
__tablename__ = 'UserDB'
# 主鍵 primary_key | 自動增長 autoincrement | 不為空 nullable | 唯一性約束 unique
id = Column(Integer, primary_key=True, autoincrement=True, nullable=True, unique=True)
# 字符串類型
username = Column(String(32), nullable=True, default="none")
password = Column(String(32), nullable=True, default="none")
# 姓名字段默認值是0
age = Column(Integer,nullable=False, default=0)
# 增加創(chuàng)建日期 [日期:時間]
create_time = Column(DateTime, default=datetime.datetime.now)
# onupdate=datetime.now 每次更新數(shù)據(jù)的時候都要更新該字段值
update_time = Column(DateTime, onupdate=datetime.datetime.now, default=datetime.datetime.now)
# 增加用戶分數(shù)
user_value = Column(Float, default=0.0)
# 枚舉類型定義
# tag = Column(Enum("python",'flask','django'))
# __repr__方法用于輸出該類的對象被print()時輸出的字符串，如果不想寫可以不寫
def __repr__(self):
return "<UserDB(username='%s',password='%s')>" % (self.username,self.password)
if __name__ == "__main__":
print("當前表名: {}".format(User.__table__))
# 創(chuàng)建會話
Session = sessionmaker(bind=engine)
session = Session()
# 創(chuàng)建數(shù)據(jù)表
Base.metadata.create_all(engine, checkfirst=True)
# 逐條增加新記錄
insert_user = User(username='lyshark', password='123456', age=24, user_value=12.5)
session.add(insert_user)
insert_user = User(username='sqlalchemy', password='123', age=34, user_value=45.8)
session.add(insert_user)
# 插入多條記錄
session.add_all(
[
User(username="admin", password="123123", age=54, user_value=66.9),
User(username="root", password="3456576", age=67, user_value=98.4)
]
)
# 提交事務(wù)
session.commit()
數(shù)據(jù)庫查詢
演示了通過ORM關(guān)系映射實現(xiàn)對單表的簡單查詢與篩選過濾功能。
import sqlite3,time,datetime
from sqlalchemy import func
from sqlalchemy import or_
from sqlalchemy.orm import sessionmaker
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, Float,DateTime
# 建立基本映射類
Base = declarative_base()
# 創(chuàng)建SQLITE數(shù)據(jù)庫
engine = create_engine('sqlite:///:memory:', echo=False)
# 創(chuàng)建映射類User
class User(Base):
__tablename__ = 'UserDB'
id = Column(Integer, primary_key=True, autoincrement=True, nullable=True, unique=True)
username = Column(String(32), nullable=True, default="none")
password = Column(String(32), nullable=True, default="none")
age = Column(Integer,nullable=False, default=0)
create_time = Column(DateTime, default=datetime.datetime.now)
update_time = Column(DateTime, onupdate=datetime.datetime.now, default=datetime.datetime.now)
user_value = Column(Float, default=0.0)
# __repr__方法用于輸出該類的對象被print()時輸出的字符串，如果不想寫可以不寫
def __repr__(self):
return "<UserDB(username='%s',password='%s')>" % (self.username,self.password)
if __name__ == "__main__":
# 創(chuàng)建會話
Session = sessionmaker(bind=engine)
session = Session()
# 創(chuàng)建數(shù)據(jù)表(存在則跳過)
Base.metadata.create_all(engine, checkfirst=True)
# 查詢所有字段
all_value = session.query(User).all()
for item in all_value:
print("ID: {} --> 用戶: {}".format(item.id, item.username))
# 查詢指定字段
key_value = session.query(User.username,User.password).all()
print(key_value)
# 查詢第一條
first_value = session.query(User).first()
print("第一條記錄: {} {}".format(first_value.username, first_value.password))
# 使用過濾器 [ 過濾出age>20的用戶,輸出其(id,username)字段 ]
filter_value = session.query(User.id,User.username).filter(User.age > 20).all()
print("過濾結(jié)果: {}".format(filter_value))
# 排序輸出 [ 正序/倒序 ]
sort_value = session.query(User.username,User.age).order_by(User.age).all()
print("正序排列: {}".format(sort_value))
sort_value = session.query(User.username,User.age).order_by(User.age.desc()).all()
print("倒序排列: {}".format(sort_value))
# 查詢計數(shù)
count_value = session.query(User).count()
print("記錄條數(shù): {}".format(count_value))
# and/or 條件過濾 默認為and 在filter()中用,分隔多個條件表示,如果是or則需增加or_連接多個條件
and_value = session.query(User.username,User.age).filter(User.age >= 20, User.age <= 40).all()
print("與查詢: {}".format(and_value))
or_value = session.query(User.username,User.age).filter(or_(User.age >= 20, User.age <= 40)).all()
print("或查詢: {}".format(or_value))
# 等于查詢
equal_value = session.query(User.username,User.password).filter(User.age == 67).all()
print("等于查詢: {}".format(equal_value))
not_equal_value = session.query(User.username,User.password).filter(User.age != 67).all()
print("不等于查詢: {}".format(not_equal_value))
# like模糊匹配
like_value = session.query(User.username,User.create_time).filter(User.username.like("%ly%")).all()
print("模糊匹配: {}".format(like_value))
# in查詢范圍
in_value = session.query(User.username,User.password).filter(User.age.in_([24,34])).all()
print("查詢兩者: {}".format(in_value))
not_in_value = session.query(User.username,User.password).filter(User.age.notin_([24,34])).all()
print("查詢非兩者: {}".format(not_in_value))
# op正則匹配查詢
op_value = session.query(User.username).filter(User.username.op("regexp")("^a")).all()
print("正則匹配: {}".format(op_value))
# 調(diào)用數(shù)據(jù)庫內(nèi)置函數(shù)
func_value = session.query(func.count(User.age)).one()
print("調(diào)用函數(shù): {}".format(func_value))
# 數(shù)據(jù)切片
cat_value = session.query(User.username).all()[:2]
print("輸出前兩條: {}".format(cat_value))
cat_value = session.query(User.username).offset(5).limit(3).all()
print("第6行開始顯示前3個: {}".format(cat_value))
cat_value = session.query(User.username).order_by(User.id.desc())[0:10]
print("輸出最后10個: {}".format(cat_value))
# 非空查詢
isnot_value = session.query(User).filter(User.username.isnot(None)).all()
print("非空顯示: {}".format(isnot_value))
null_value = session.query(User).filter(User.username.is_(None)).all()
print("為空顯示: {}".format(null_value))
# 分組測試
group_by = session.query(User.username,func.count(User.id)).group_by(User.age).all()
print("根據(jù)年齡分組: {}".format(group_by))
# 進一步過濾查詢
having_by = session.query(User.username,User.age).group_by(User.age).having(User.age > 30).all()
print("以age分組,并查詢age大于30的記錄: {}".format(having_by))
數(shù)據(jù)庫修改
演示了修改數(shù)據(jù)庫參數(shù)以及對數(shù)據(jù)庫指定記錄的刪除功能。
import sqlite3,time,datetime
from sqlalchemy import func
from sqlalchemy import or_
from sqlalchemy.orm import sessionmaker
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, Float,DateTime
# 建立基本映射類
Base = declarative_base()
# 創(chuàng)建SQLITE數(shù)據(jù)庫
engine = create_engine('sqlite:///:memory:', echo=False)
# 創(chuàng)建映射類User
class User(Base):
__tablename__ = 'UserDB'
id = Column(Integer, primary_key=True, autoincrement=True, nullable=True, unique=True)
username = Column(String(32), nullable=True, default="none")
password = Column(String(32), nullable=True, default="none")
age = Column(Integer,nullable=False, default=0)
create_time = Column(DateTime, default=datetime.datetime.now)
update_time = Column(DateTime, onupdate=datetime.datetime.now, default=datetime.datetime.now)
user_value = Column(Float, default=0.0)
# __repr__方法用于輸出該類的對象被print()時輸出的字符串，如果不想寫可以不寫
def __repr__(self):
return "<UserDB(username='%s',password='%s')>" % (self.username,self.password)
if __name__ == "__main__":
# 創(chuàng)建會話
Session = sessionmaker(bind=engine)
session = Session()
# 創(chuàng)建數(shù)據(jù)表(存在則跳過)
Base.metadata.create_all(engine, checkfirst=True)
# 修改數(shù)據(jù): 先查詢在修改
select_update = session.query(User).filter_by(username="lyshark").first()
select_update.password = "test1234"
session.commit()
# 修改數(shù)據(jù): 直接修改
session.query(User).filter_by(username="lyshark").update({User.password: 'abcd'})
session.commit()
session.query(User).filter_by(username="lyshark").update({"password": '123456'})
session.commit()
# 刪除數(shù)據(jù): 先查詢在刪除
del_ptr = session.query(User).filter_by(username="lyshark").first()
session.delete(del_ptr)
session.commit()
# 刪除數(shù)據(jù): 直接刪除
session.query(User).filter(User.username=="sqlalchemy").delete()
session.commit()
數(shù)據(jù)庫查詢轉(zhuǎn)字典
將從數(shù)據(jù)庫中過濾查詢指定的記錄，并將該記錄轉(zhuǎn)換為字典或JSON格式，利于解析。
import sqlite3,time,datetime,json
from sqlalchemy.orm import sessionmaker
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, Float,DateTime
# 建立基本映射類
Base = declarative_base()
# 創(chuàng)建SQLITE數(shù)據(jù)庫
engine = create_engine('sqlite:///:memory:', echo=False)
# 創(chuàng)建映射類User
class User(Base):
__tablename__ = 'UserDB'
id = Column(Integer, primary_key=True, autoincrement=True, nullable=True, unique=True)
username = Column(String(32), nullable=True, default="none")
password = Column(String(32), nullable=True, default="none")
age = Column(Integer,nullable=False, default=0)
create_time = Column(DateTime, default=datetime.datetime.now)
update_time = Column(DateTime, onupdate=datetime.datetime.now, default=datetime.datetime.now)
user_value = Column(Float, default=0.0)
# 查詢結(jié)果轉(zhuǎn)字典 (保留數(shù)據(jù)類型)
def single_to_dict(self):
return {c.name: getattr(self, c.name) for c in self.__table__.columns}
# 查詢結(jié)果轉(zhuǎn)字典 (全轉(zhuǎn)為字符串)
def dobule_to_dict(self):
result = {}
for key in self.__mapper__.c.keys():
if getattr(self, key) is not None:
result[key] = str(getattr(self, key))
else:
result[key] = getattr(self, key)
return result
# 將查詢結(jié)果轉(zhuǎn)為JSON
def to_json(all_vendors):
v = [ ven.dobule_to_dict() for ven in all_vendors ]
return v
if __name__ == "__main__":
# 創(chuàng)建會話
Session = sessionmaker(bind=engine)
session = Session()
# 創(chuàng)建數(shù)據(jù)表(存在則跳過)
Base.metadata.create_all(engine, checkfirst=True)
# 查詢結(jié)果轉(zhuǎn)為字典(保持數(shù)據(jù)庫格式)
key_value = session.query(User).first()
data = key_value.single_to_dict()
print("轉(zhuǎn)為字典: {}".format(data))
# 查詢結(jié)果轉(zhuǎn)為字典(字符串格式)
key_value = session.query(User).first()
data = key_value.dobule_to_dict()
print("轉(zhuǎn)為字符串字典: {}".format(data))
# 查詢結(jié)果轉(zhuǎn)為JSON格式
key_value = session.query(User)
data = to_json(key_value)
print("轉(zhuǎn)為JSON格式: {}".format(data))
數(shù)據(jù)庫類內(nèi)函數(shù)調(diào)用
用戶在使用ORM模型定義類時，可以同時在該映射類中定義各種針對類模型的處理函數(shù)，實現(xiàn)對數(shù)據(jù)的動態(tài)處理
from werkzeug.security import generate_password_hash,check_password_hash
import sqlite3,datetime,time
from sqlalchemy.orm import sessionmaker
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, Float, DateTime, Date, Time, Boolean, DECIMAL, Enum, Text
# 創(chuàng)建SQLITE數(shù)據(jù)庫
engine = create_engine("sqlite:///:memory:", encoding='utf-8')
Base = declarative_base() # 生成orm基類
# 創(chuàng)建會話
Session_class = sessionmaker(bind=engine) # 創(chuàng)建與數(shù)據(jù)庫的會話session
session = Session_class() # 生成session實例
# 創(chuàng)建映射類User
class User(Base):
__tablename__ = 'UserDB'
id = Column(Integer, primary_key=True, autoincrement=True, nullable=True, unique=True)
username = Column(String(64))
_password_hash_ = Column(String(256)) # 加下劃線作為私有函數(shù),無法被外部訪問
email = Column(String(64))
# 設(shè)置一個password字段用來設(shè)置密碼
@property
def password(self):
raise Exception("密碼不能被讀取")
# 賦值password字段時,則自動加密存儲
@password.setter
def password(self, value):
self._password_hash_ = generate_password_hash(value)
# 使用 check_password,進行密碼校驗 返回True False。
def check_password(self, pasword):
return check_password_hash(self._password_hash_, pasword)
# 設(shè)置輸出函數(shù)
def print_function(self):
# 密碼不可讀調(diào)用 self.password 會報錯
# print("用戶: {} 密碼: {}".format(self.username,self.password))
print("用戶: {} email: {}".format(self.username, self.email))
return True
if __name__ == "__main__":
print("當前表名: {}".format(User.__table__))
# 創(chuàng)建數(shù)據(jù)表
Base.metadata.create_all(engine, checkfirst=True)
# 插入測試數(shù)據(jù)
insert = User(username="lyshark",password="123123",email="lyshark@163.com")
session.add(insert)
insert = User(username="admin",password="556677",email="lyshark@163.com")
session.add(insert)
session.commit()
# 查詢測試
tag = session.query(User).filter_by(username="lyshark").first()
print("測試密碼是否正確: {}".format(tag.check_password("123123")))
# 調(diào)用函數(shù)驗證當前用戶
tag = session.query(User).filter_by(username="admin").first()
func = tag.print_function()
print("輸出測試: {}".format(func))
數(shù)據(jù)庫聚合函數(shù)
通過func庫調(diào)用數(shù)據(jù)庫內(nèi)的聚合函數(shù),實現(xiàn)統(tǒng)計最大最小平均數(shù)等數(shù)據(jù)。
import sqlite3,datetime,time
from sqlalchemy import func
from sqlalchemy.orm import sessionmaker
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, Float, DateTime, Date, Time, Boolean, DECIMAL, Enum, Text
# 建立基本映射類
Base = declarative_base()
# 創(chuàng)建SQLITE數(shù)據(jù)庫
engine = create_engine('sqlite:///:memory:', echo=False)
# 創(chuàng)建映射類User
class User(Base):
__tablename__ = 'UserDB'
# 主鍵 primary_key | 自動增長 autoincrement | 不為空 nullable | 唯一性約束 unique
id = Column(Integer, primary_key=True, autoincrement=True, nullable=True, unique=True)
# 字符串類型
username = Column(String(32), nullable=True, default="none")
password = Column(String(32), nullable=True, default="none")
# 姓名字段默認值是0
age = Column(Integer,nullable=False, default=0)
# 增加創(chuàng)建日期 [日期:時間]
create_time = Column(DateTime, default=datetime.datetime.now)
# onupdate=datetime.now 每次更新數(shù)據(jù)的時候都要更新該字段值
update_time = Column(DateTime, onupdate=datetime.datetime.now, default=datetime.datetime.now)
# 增加用戶分數(shù)
user_value = Column(Float, default=0.0)
# 枚舉類型定義
# tag = Column(Enum("python",'flask','django'))
# __repr__方法用于輸出該類的對象被print()時輸出的字符串，如果不想寫可以不寫
def __repr__(self):
return "<UserDB(username='%s',password='%s')>" % (self.username,self.password)
if __name__ == "__main__":
print("當前表名: {}".format(User.__table__))
# 創(chuàng)建會話
Session = sessionmaker(bind=engine)
session = Session()
# 創(chuàng)建數(shù)據(jù)表
Base.metadata.create_all(engine, checkfirst=True)
# 統(tǒng)計總數(shù)
count = session.query(func.count(User.id)).first()
print("總記錄: {}".format(count))
# age 字段平均值
age_avg = session.query(func.avg(User.age)).first()
print("平均值: {}".format(age_avg))
# age 字段最大值
age_max = session.query(func.max(User.age)).first()
print("最大值: {}".format(age_max))
# age 最小值
age_min = session.query(func.min(User.age)).one()
print("最小值: {}".format(age_min))
# age 求和 只求前三個的和
age_sum = session.query(func.sum(User.age)).one()
print("求總數(shù): {}".format(age_sum))
# 提交事務(wù)
session.commit()
ORM定義一對多關(guān)系
SQLAlchemy提供了一個relationship，這個類可以定義屬性，以后在訪問相關(guān)聯(lián)的表的時候就直接可以通過屬性訪問的方式就可以訪問得到。
from sqlalchemy.orm import sessionmaker
from sqlalchemy.orm import relationship
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import create_engine, Column, Integer, String, Text, func, ForeignKey
# 打開數(shù)據(jù)庫
Base = declarative_base()
engine = create_engine('sqlite:///:memory:', echo=False)
# 主表
class Author(Base):
__tablename__ = 'author'
id = Column(Integer, primary_key=True, autoincrement=True)
name = Column(String(10), nullable=False)
# 定義外鍵關(guān)聯(lián)到Book模型上面,主表是author
books = relationship('Book', backref='author')
def __repr__(self):
return '<Author:(id={}, name={})>'.format(self.id, self.name)
# 從表
class Book(Base):
__tablename__ = 'book'
id = Column(Integer, primary_key=True, autoincrement=True)
name = Column(String(20), nullable=False)
# 外鍵關(guān)聯(lián)到主表author的id字段上
author_id = Column(Integer, ForeignKey('author.id',ondelete="RESTRICT"))
def __repr__(self):
return '<Book:(id={}, name={}, author_id={})>'.format(self.id, self.name, self.author_id)
if __name__ == "__main__":
# 創(chuàng)建會話
Session = sessionmaker(bind=engine)
session = Session()
# 創(chuàng)建數(shù)據(jù)表(存在則跳過)
Base.metadata.create_all(engine, checkfirst=True)
# ------------------------------------------------------
# 創(chuàng)建數(shù)據(jù)并插入
author1 = Author(id=1, name="張三")
author2 = Author(id=2, name="李四")
book1 = Book(name="<Python 開發(fā)>", author_id=1)
book2 = Book(name="<C++ 開發(fā)教程>", author_id=1)
book3 = Book(name="<C# 從入門到精通>", author_id=1)
book4 = Book(name="<滲透測試指南>", author_id=2)
book5 = Book(name="<nmap 掃描工具指南>", author_id=2)
session.add_all([author1,book1,book2,book3])
session.add_all([author2,book4,book5])
session.commit()
# ------------------------------------------------------
# 關(guān)聯(lián)插入模式
author1 = Author(id=3, name="王五")
book1 = Book(name="<Python 開發(fā)>", author_id=1)
book2 = Book(name="<C++ 開發(fā)教程>", author_id=1)
book3 = Book(name="<C# 從入門到精通>", author_id=1)
author1.books.append(book1)
author1.books.append(book2)
author1.books.append(book3)
session.add(author1)
session.commit()
# ------------------------------------------------------
# 一對多查詢測試
book = session.query(Book).get(1)
print("書籍作者: {}".format(book.author.name))
author = session.query(Author).get(1)
print("書籍數(shù)量: {}".format(len(author.books)))
for book_name in author.books:
print("書籍: {}".format(book_name.name))
ORM定義一對一關(guān)系
如果想要將兩個模型映射成一對一的關(guān)系，那么應(yīng)該在父模型中，指定引用的時候，要傳遞一個uselist=False參數(shù)進去。就是告訴父模型，以后引用這個從模型的時候，不再是一個列表了，而是一個對象了。
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker,relationship,backref
from sqlalchemy import create_engine, Column, Integer, String, Text, func, ForeignKey
# 打開數(shù)據(jù)庫
Base = declarative_base()
engine = create_engine('sqlite:///:memory:', echo=False)
# 主表
class User(Base):
__tablename__ = 'user'
id = Column(Integer, primary_key=True, autoincrement=True)
username = Column(String(50), nullable=False)
def __repr__(self):
return 'User(username:%s)' % self.username
# 從表
class UserExtend(Base):
__tablename__ = 'user_extend'
id = Column(Integer,primary_key=True,autoincrement=True)
school = Column(String(50))
age = Column(String(32))
sex = Column(String(32))
uid = Column(Integer,ForeignKey('user.id'))
user = relationship('User',backref=backref('extend', uselist=False))
#uselist=False 告訴父模型 以后引用時不再是列表 而是對象
def __repr__(self):
return 'extend(school:%s)'%self.school
if __name__ == "__main__":
# 創(chuàng)建會話
Session = sessionmaker(bind=engine)
session = Session()
# 創(chuàng)建數(shù)據(jù)表(存在則跳過)
Base.metadata.create_all(engine, checkfirst=True)
# ------------------------------------------------------
# 插入測試數(shù)據(jù)
user = User(username="lyshark")
extend = UserExtend(school="<家里蹲大學>", age="22", sex="M")
extend.user = user
session.add(extend)
session.commit()
# ------------------------------------------------------
# 一對一關(guān)系測試
user_ptr = session.query(User).first()
print("用戶名: {} --> 學校: {}".format(user_ptr.username,user_ptr.extend.school))
extend_ptr = session.query(UserExtend).first()
print("用戶名: {} --> 學校: {} --> 年齡: {} --> 性別: {}".format(extend_ptr.user.username,extend_ptr.school,extend_ptr.age,extend_ptr.sex))
ORM定義多對多關(guān)系
多對多與上面的一對多，一對一不同，創(chuàng)建多對對必須使用中間表Table來解決查詢問題。
- 多對多的關(guān)系需要通過一張中間表來綁定他們之間的關(guān)系。
- 先把兩個需要做多對多的模型定義出來
- 使用Table定義一個中間表，中間表一般就是包含兩個模型的外鍵字段就可以了，并且讓他們兩個來作為一個“復合主鍵”。
- 在兩個需要做多對多的模型中隨便選擇一個模型，定義一個relationship屬性，來綁定三者之間的關(guān)系，在使用relationship的時候，需要傳入一個secondary=中間表。
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker,relationship,backref
from sqlalchemy import create_engine, Column, Integer, String, Text, func, ForeignKey, Table
from sqlalchemy import Column,INT,VARCHAR,ForeignKey
from sqlalchemy.orm import relationship
# 打開數(shù)據(jù)庫
Base = declarative_base()
engine = create_engine('sqlite:///:memory:', echo=False)
# 女孩表
class Girls(Base):
__tablename__ = "girl"
id = Column(Integer,primary_key=True, autoincrement=True)
name = Column(String(32))
# 建立多對多關(guān)系
g2b = relationship("Boys",backref="b2g",secondary="hotel")
# 男孩表
class Boys(Base):
__tablename__ = "boy"
id = Column(Integer,primary_key=True, autoincrement=True)
name = Column(String(32))
# 映射關(guān)系表
class Table(Base):
__tablename__ = "hotel"
id = Column(Integer, primary_key=True, autoincrement=True)
boy_id = Column(Integer,ForeignKey("boy.id"))
girl_id = Column(Integer,ForeignKey("girl.id"))
if __name__ == "__main__":
# 創(chuàng)建會話
Session = sessionmaker(bind=engine)
session = Session()
# 創(chuàng)建數(shù)據(jù)表(存在則跳過)
Base.metadata.create_all(engine, checkfirst=True)
# ------------------------------------------------------
# 增加數(shù)據(jù) - relationship 正向
girl_obj = Girls(name="女孩主")
girl_obj.g2b = [Boys(name="男孩從1"),Boys(name="男孩從2")]
session.add(girl_obj)
session.commit()
# 增加數(shù)據(jù) - relationship 反向
boy_obj = Boys(name="男孩主")
boy_obj.b2g = [Girls(name="女孩從1"),Girls(name="女孩從2")]
session.add(boy_obj)
session.commit()
# ------------------------------------------------------
# 正向查詢
girl_obj_list = session.query(Girls).all()
for girl_obj in girl_obj_list:
for boy in girl_obj.g2b:
print(girl_obj.name,boy.name)
# 反向查詢
boy_obj_list = session.query(Boys).all()
for boy in boy_obj_list:
for girl in boy.b2g:
print(girl.name,boy.name)
連接查詢與子查詢
連接查詢通過JOIN語句實現(xiàn)，子查詢則通過subquery實現(xiàn)，首先需要創(chuàng)建一對多關(guān)系然后才可使用子查詢。
from sqlalchemy.orm import sessionmaker
from sqlalchemy.orm import relationship
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import create_engine, Column, Integer, String, Text, func, ForeignKey
# 打開數(shù)據(jù)庫
Base = declarative_base()
engine = create_engine('sqlite:///:memory:', echo=False)
# 主表
class Author(Base):
__tablename__ = 'author'
id = Column(Integer, primary_key=True, autoincrement=True)
name = Column(String(10), nullable=False)
# 定義外鍵關(guān)聯(lián)到Book模型上面,主表是author
books = relationship('Book', backref='author')
def __repr__(self):
return '<Author:(id={}, name={})>'.format(self.id, self.name)
# 從表
class Book(Base):
__tablename__ = 'book'
id = Column(Integer, primary_key=True, autoincrement=True)
name = Column(String(20), nullable=False)
# 外鍵關(guān)聯(lián)到主表author的id字段上
author_id = Column(Integer, ForeignKey('author.id',ondelete="RESTRICT"))
def __repr__(self):
return '<Book:(id={}, name={}, author_id={})>'.format(self.id, self.name, self.author_id)
if __name__ == "__main__":
# 創(chuàng)建會話
Session = sessionmaker(bind=engine)
session = Session()
# 創(chuàng)建數(shù)據(jù)表(存在則跳過)
Base.metadata.create_all(engine, checkfirst=True)
# ------------------------------------------------------
# 創(chuàng)建數(shù)據(jù)并插入
author1 = Author(id=1, name="張三")
author2 = Author(id=2, name="李四")
book1 = Book(name="<Python 開發(fā)>", author_id=1)
book2 = Book(name="<C++ 開發(fā)教程>", author_id=1)
book3 = Book(name="<C# 從入門到精通>", author_id=1)
book4 = Book(name="<滲透測試指南>", author_id=2)
book5 = Book(name="<nmap 掃描工具指南>", author_id=2)
session.add_all([author1,book1,book2,book3])
session.add_all([author2,book4,book5])
session.commit()
# ------------------------------------------------------
# 連表查詢
no_join_select = session.query(Author).filter(Author.id == Book.id).filter(Author.name == "王五").all()
print("查詢主鍵==從鍵 并且 Author.name == 王五的記錄: {}".format(no_join_select))
# JOIN 連接查詢
join = session.query(Author).join(Book).filter(Book.name=="<nmap 掃描工具指南>").first().name
print("查詢主表Author中的Book書名的作者是: {}".format(join))
join = session.query(Book).join(Author).filter(Author.name=="李四").all()
for book in join:
print("查詢從表Book中的Author作者有哪些書: {}".format(book.name))
# subquery 子查詢
sbq = session.query(Book.author_id,func.count('*').label("book_count")).group_by(Book.author_id).subquery()
print("查詢出書籍編號計數(shù)(子語句): {}".format(sbq))
sub_join = session.query(Author.name,sbq.c.book_count).outerjoin(sbq,Author.id == sbq.c.author_id).all()
print("查詢用戶有幾本書(主語句): {}".format(sub_join))以上就是Python使用SQLAlchemy模塊實現(xiàn)操作數(shù)據(jù)庫的詳細內(nèi)容，更多關(guān)于Python SQLAlchemy的資料請關(guān)注腳本之家其它相關(guān)文章！
您可能感興趣的文章:
- 分析解決Python中sqlalchemy數(shù)據(jù)庫連接池QueuePool異常
- 3個Python?SQLAlchemy數(shù)據(jù)庫操作功能詳解
- Python?SQLAlchemy與數(shù)據(jù)庫交互操作完整指南
- Python使用sqlalchemy實現(xiàn)連接數(shù)據(jù)庫的幫助類
- Python中使用sqlalchemy操作數(shù)據(jù)庫的問題總結(jié)
- Python中SQLAlchemy庫的使用方法分析
- Python使用SQLAlchemy進行復雜查詢的操作代碼
- Python如何使用sqlalchemy實現(xiàn)動態(tài)sql
- python SQLAlchemy 數(shù)據(jù)庫連接池的實現(xiàn)
相關(guān)文章
這篇文章主要介紹了Python驗證碼識別的方法,涉及Python針對驗證碼圖片的相關(guān)分析與操作技巧,具有一定參考借鑒價值,需要的朋友可以參考下2015-07-07
關(guān)于tensorflow的幾種參數(shù)初始化方法小結(jié)
今天小編就為大家分享一篇關(guān)于tensorflow的幾種參數(shù)初始化方法小結(jié)，具有很好的參考價值，希望對大家有所幫助。一起跟隨小編過來看看吧2020-01-01
這篇文章主要介紹了Python按指定列的空值刪除行的操作代碼,本文通過實例代碼給大家介紹的非常詳細,感興趣的朋友跟隨小編一起看看吧2024-01-01
對于文件和目錄的處理，雖然可以通過操作系統(tǒng)命令來完成，但是Python語言為了便于開發(fā)人員以編程的方式處理相關(guān)工作，提供了許多處理文件和目錄的內(nèi)置函數(shù)。重要的是，這些函數(shù)無論是在Unix、Windows還是Macintosh平臺上，它們的使用方式是完全一致的。2011-03-03
這篇文章主要介紹了在Python中使用lambda高效操作列表的教程,結(jié)合了包括map、filter、reduce、sorted等函數(shù),需要的朋友可以參考下2015-04-04
python和websocket構(gòu)建實時日志跟蹤器的步驟
這篇文章主要介紹了python和websocket構(gòu)建實時日志跟蹤器的步驟，幫助大家更好的理解和學習使用python，感興趣的朋友可以了解下2021-04-04
Windows不是有個消息通知功能，挺喜歡這個功能的，但是不太方便使用，也懶得去研究，于是準備用Python自己寫一個，感興趣的可以了解一下2022-12-12
這篇文章主要為大家詳細介紹了Python實現(xiàn)圖片添加文字，文中示例代碼介紹的非常詳細，具有一定的參考價值，感興趣的小伙伴們可以參考一下2019-11-11
python thrift搭建服務(wù)端和客戶端測試程序
這篇文章主要為大家詳細介紹了python thrift搭建服務(wù)端和客戶端測試程序，具有一定的參考價值，感興趣的小伙伴們可以參考一下2018-01-01
這篇文章主要介紹了python如何實現(xiàn)圖片轉(zhuǎn)文字問題，具有很好的參考價值，希望對大家有所幫助。如有錯誤或未考慮完全的地方，望不吝賜教2023-03-03