chapter13.2、SQLAlchemy
ORM
ORM,物件關係對映,物件和關係之間的對映,使用面向物件的方式來操作資料庫。 關係模型和Python物件之間的對映- table => class ,表對映為類
- row => object ,行對映為例項
- column => property ,欄位對映為屬性
from sqlalchemy.ext.declarative import declarative_base # 建立基類,便於實體類繼承。SQLAlchemy大量使用了超程式設計 Base = declarative_base()
建立實體類
# 建立實體類 class Student(Base): # 指定表名 __tablename__ = 'student' # 定義類屬性對應欄位 id = Column(Integer, primary_key=True, autoincrement=True) name = Column(String(64), nullable=False) age = Column(Integer) # 第一引數是欄位名,如果和屬性名不一致,一定要指定 # age = Column('age', Integer) def __repr__(self): return "{} id={} name={} age={}".format( self.__class__.__name__, self.id, self.name, self.age)
__tablename__ 指定表名 Column類指定對應的欄位,必須指定 例項化
s = Student(name='tom') print(s.name) s.age = 20 print(s.age)
建立表 可以使用SQLAlchemy來建立、刪除表
# 刪除繼承自Base的所有表 Base.metadata.drop_all(engine) # 建立繼承自Base的所有表 Base.metadata.create_all(engine)
metadata 記錄的是繼承自Base的表,
生產環境很少這樣建立表,都是系統上線的時候由指令碼生成。 生成環境很少刪除表, 寧可廢棄都不能刪除。建立會話session
在一個會話中操作資料庫,會話建立在連線上,連線被引擎管理。 當第一次使用資料庫時,從引擎維護的連線池中獲取一個連線使用。# 建立session Session = sessionmaker(bind=engine) # 返回類 session = Session() # 例項化 # 依然在第一次使用時連線資料庫
session物件執行緒不安全。所以不同執行緒應該使用不用的session物件。
Session類和engine有一個就行了。import sqlalchemy from sqlalchemy import create_engine,Column,String,Integer from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import sessionmaker USER = "root" PWD = "root" HOST = "172.16.34.34" PORT = "3306" DB = "test" conn_str = 'mysql+pymysql://{}:{}@{}:{}/{}'.format( USER,PWD,HOST,PORT,DB ) engine = create_engine(conn_str, echo=True) Base = declarative_base() # 建立實體類 class Student(Base): # 指定表名,必須指定 __tablename__ = 'student' # 定義類屬性對應欄位 id = Column(Integer,autoincrement=True,primary_key=True) name = Column(String(64),nullable=False) age = Column("age",Integer,nullable=False) gen = Column("gender",String(20)) # 第一引數是欄位名,如果和屬性名不一致,一定要指定 # age = Column('age', Integer) def __repr__(self): return "{} id={} name={} age={} gender={}".format( self.__class__.__name__, self.id, self.name, self.age, self.gen) print(Student) print(repr(Student.__table__)) s = Student(name='tom',age=20) print(s.name) s.gender = "M" print(s.gender) # Base.metadata.drop_all(bind=engine) # Base.metadata.create_all(bind=engine) # metadata 記錄的是繼承自Base的表,刪除也是 Session = sessionmaker(bind=engine) # 不做操作不會處理連線 session =Session() # 建立session物件,也不會在此時連線資料庫
CRUD操作
增
add():增加一個物件
add_all():可迭代物件,元素是物件
session.add(s) #新增一次物件s,未提交 print(s) session.commit() #提交一次物件s print(s) try: session.add_all([s]) print(s) session.commit() #在同一個執行緒內,再次提交同一個未修改過的物件不會執行 except: session.rollback() # 使用要加日誌記錄 raise
add_all()方法不會提交成功的,不是因為它不對,而是s,s成功提交後,s的主鍵就有了值,所以,只要s沒有修改過,就認為沒有改動。如下,s變化了,就可以提交修改了。
s.name = 'jerry' # 修改 session.add_all([s])
s主鍵沒有值,就是新增;主鍵有值,就是找到主鍵對應的記錄修改。
簡單查詢
使用query()方法,返回一個Query物件
students = session.query(Student) # 無條件,相當於select * from student for student in students: print(student) print('~~~~~~~~~~~~~') student = session.query(Student).get(3) # 通過主鍵查詢,相當於select * from student where id=3 print(student)
query方法將實體類傳入,返回類的物件可迭代物件,這時候並不查詢。迭代它就執行SQL來查詢資料庫,封裝資料到指定類的例項。
get方法使用主鍵查詢,返回一條傳入類的一個例項。物件不存在,返回None
改
先查後改
student = session.query(Student).get(3) print(student) student.name = 'sam' student.age = 30 print(student) session.add(student) session.commit()
刪
先查後刪 嘗試執行以下程式碼,會發現產生一個異常,未持久的異常try: student = Student(id=2, name="serry", age=10) session.delete(student) session.commit() except Exception as e: session.rollback() print('~~~~~~~~') print(e)
Instance '<Student at 0x26edf10b438>' is not persisted
資料庫查詢資料慢且忙,最好在容器中就記錄下來,不要重複查詢相同的資料
狀態
每一個實體,都有一個狀態屬性_sa_instance_state,其型別是sqlalchemy.orm.state.InstanceState,可以使用sqlalchemy.inspect(entity)函式檢視狀態。
常見的狀態值有transient、pending、persistent、deleted、detached。
狀態 | 說明 |
transient | 實體類尚未加入到session中,同時並沒有儲存到資料庫中 |
pending | transient的實體被add()到session中,狀態切換到pending,但它還沒有flush到資料庫中 |
persistent | session中的實體物件對應著資料庫中的真實記錄。pending狀態在提交成功後可以變成persistent狀態,或者查詢成功返回的實體也是persistent狀態 |
deleted | 實體被刪除且已經flush但未commit完成。事務提交成功了,實體變成detached,事務失敗, |
detached | 刪除成功的實體進入這個狀態 |
新建一個實體,狀態是transient臨時的。
一旦add()後從transient變成pending狀態。
成功commit()後從pending變成persistent狀態。
成功查詢返回的實體物件,也是persistent狀態。
persistent狀態的實體,修改依然是persistent狀態。
persistent狀態的實體,刪除後,flush後但沒有commit,就變成deteled狀態,成功提交,變為detached狀態,提交失敗,還原到persistent狀態。flush方法,主動把改變應用到資料庫中去。
刪除、修改操作,需要對應一個真實的記錄,所以要求實體物件是persistent狀態。
import sqlalchemy from sqlalchemy import create_engine,Column,String,Integer from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import sessionmaker USER = "root" PWD = "root" HOST = "172.16.34.34" PORT = "3306" DB = "test" conn_str = 'mysql+pymysql://{}:{}@{}:{}/{}'.format( USER,PWD,HOST,PORT,DB ) engine = create_engine(conn_str, echo=True) Base = declarative_base() # 建立實體類 class Student(Base): # 指定表名,必須指定 __tablename__ = 'student' # 定義類屬性對應欄位 id = Column(Integer,autoincrement=True,primary_key=True) name = Column(String(64),nullable=False) age = Column("age",Integer,nullable=False) gen = Column("gender",String(20)) # 第一引數是欄位名,如果和屬性名不一致,一定要指定 # age = Column('age', Integer) def __repr__(self): return "{} id={} name={} age={} gender={}".format( self.__class__.__name__, self.id, self.name, self.age, self.gen) print(Student) print(repr(Student.__table__)) Session = sessionmaker(bind=engine) session =Session() from sqlalchemy.orm.state import InstanceState def getstate(entity,i): insp = sqlalchemy.inspect(entity) state = "session={}, attached={},\ntransient={},persistent={},\npending={},deleted={}.detached={}".format( insp.session_id, insp._attached, insp.transient, insp.persistent, insp.pending, insp.deleted, insp.detached ) print(i,state) print(insp.key) print("_"*30) # student = session.query(Student).get(3) # getstate(student,1) try: student = Student(name="Tony", age=30) getstate(student, 2) # transit student = Student(name="sammy", age=30) getstate(student, 3) # transit session.add(student) # add後變成pending getstate(student, 4) # pending # session.delete(student) # 刪除的前提是persistent,否則拋異常 # getstate(student, 5) session.commit() getstate(student, 6) # persistent session.delete(student) # 刪除的前提是persistent,否則拋異常 getstate(student, 7) session.flush() getstate(student,8) session.commit() getstate(student, 9) except Exception as e: session.rollback() print('~~~~~~~~') print(e)
返回結果
2 session=None, attached=False, transient=True,persistent=False, pending=False,deleted=False.detached=False None ______________________________ 3 session=None, attached=False, transient=True,persistent=False, pending=False,deleted=False.detached=False None ______________________________ 4 session=1, attached=True, transient=False,persistent=False, pending=True,deleted=False.detached=False None ______________________________ 2018-11-12 18:04:27,682 INFO sqlalchemy.engine.base.Engine SHOW VARIABLES LIKE 'sql_mode' 2018-11-12 18:04:27,682 INFO sqlalchemy.engine.base.Engine {} 2018-11-12 18:04:27,683 INFO sqlalchemy.engine.base.Engine SHOW VARIABLES LIKE 'lower_case_table_names' 2018-11-12 18:04:27,683 INFO sqlalchemy.engine.base.Engine {} 2018-11-12 18:04:27,685 INFO sqlalchemy.engine.base.Engine SELECT DATABASE() 2018-11-12 18:04:27,685 INFO sqlalchemy.engine.base.Engine {} 2018-11-12 18:04:27,686 INFO sqlalchemy.engine.base.Engine show collation where `Charset` = 'utf8mb4' and `Collation` = 'utf8mb4_bin' 2018-11-12 18:04:27,686 INFO sqlalchemy.engine.base.Engine {} 2018-11-12 18:04:27,687 INFO sqlalchemy.engine.base.Engine SELECT CAST('test plain returns' AS CHAR(60)) AS anon_1 2018-11-12 18:04:27,687 INFO sqlalchemy.engine.base.Engine {} 2018-11-12 18:04:27,688 INFO sqlalchemy.engine.base.Engine SELECT CAST('test unicode returns' AS CHAR(60)) AS anon_1 2018-11-12 18:04:27,688 INFO sqlalchemy.engine.base.Engine {} 2018-11-12 18:04:27,689 INFO sqlalchemy.engine.base.Engine SELECT CAST('test collated returns' AS CHAR CHARACTER SET utf8mb4) COLLATE utf8mb4_bin AS anon_1 2018-11-12 18:04:27,689 INFO sqlalchemy.engine.base.Engine {} 2018-11-12 18:04:27,690 INFO sqlalchemy.engine.base.Engine BEGIN (implicit) 2018-11-12 18:04:27,691 INFO sqlalchemy.engine.base.Engine INSERT INTO student (name, age, gender) VALUES (%(name)s, %(age)s, %(gender)s) 2018-11-12 18:04:27,691 INFO sqlalchemy.engine.base.Engine {'name': 'sammy', 'age': 30, 'gender': None} 2018-11-12 18:04:27,692 INFO sqlalchemy.engine.base.Engine COMMIT 6 session=1, attached=True, transient=False,persistent=True, pending=False,deleted=False.detached=False (<class '__main__.Student'>, (11,), None) ______________________________ 7 session=1, attached=True, transient=False,persistent=True, pending=False,deleted=False.detached=False (<class '__main__.Student'>, (11,), None) ______________________________ 2018-11-12 18:04:27,693 INFO sqlalchemy.engine.base.Engine BEGIN (implicit) 2018-11-12 18:04:27,694 INFO sqlalchemy.engine.base.Engine SELECT student.gender AS student_gender, student.id AS student_id, student.name AS student_name, student.age AS student_age FROM student WHERE student.id = %(param_1)s 2018-11-12 18:04:27,694 INFO sqlalchemy.engine.base.Engine {'param_1': 11} 2018-11-12 18:04:27,695 INFO sqlalchemy.engine.base.Engine DELETE FROM student WHERE student.id = %(id)s 2018-11-12 18:04:27,695 INFO sqlalchemy.engine.base.Engine {'id': 11} 8 session=1, attached=True, transient=False,persistent=False, pending=False,deleted=True.detached=False (<class '__main__.Student'>, (11,), None) ______________________________ 2018-11-12 18:04:27,696 INFO sqlalchemy.engine.base.Engine COMMIT 9 session=None, attached=False, transient=False,persistent=False, pending=False,deleted=False.detached=True (<class '__main__.Student'>, (11,), None) ______________________________
複雜查詢
實體類
import enum class GenderEnum(enum.Enum): M = "M" F = "F" class Employee(Base): __tablename__ = "employees" emp_no = Column(Integer, primary_key=True) birth_date = Column(Date, nullable=False) first_name = Column(String(14), nullable=False) last_name = Column(String(16), nullable=False) gender = Column(Enum(GenderEnum), nullable=False) hire_date = Column(Date, nullable=False) def __repr__(self): return "{} no={} name={} {} gender={}".format( self.__class__.__name__, self.emp_no, self.first_name, self.last_name, self.gender.value )
# 返回的迭代器,檢視內容 def show(emps): for x in emps: print(x)
以下語句為條件
最簡單的查詢:
emps = session.query(Employee).filter(Employee.emp_no > 10015)
show(emps)
與,或,非
and條件可以使用兩個filter實現,也可以使用and_,也可使用運算子過載 &
emps = session.query(Employee).filter(Employee.emp_no > 10015).filter(Employee.gender ==GenderEnum.F) show(emps) emps = session.query(Employee).filter(and_(Employee.emp_no > 10015, Employee.gender ==GenderEnum.M)) show(emps) #運算子過載注意表示式要加括號 emps = session.query(Employee).filter((Employee.emp_no > 10015) & (Employee.gender == GenderEnum.M)) show(emps)
or 條件可以使用or_ 或者運算子 |
emps = session.query(Employee).filter(or_(Employee.emp_no > 10018, Employee.emp_no < 10003)) show(emps) #加括號 emps = session.query(Employee).filter((Employee.emp_no > 10018) | (Employee.emp_no < 10003)) show(emps)
not 條件使用not_ 或者運算子 ~
emps = session.query(Employee).filter(not_(Employee.emp_no < 10018)) show(emps) #加括號 emps = session.query(Employee).filter(~(Employee.emp_no < 10018)) show(emps)
總之,與或非的運算子&、|、~,一定要在表示式上加上括號
in
emplist = [10010, 10015, 10018] emps = session.query(Employee).filter(Employee.emp_no.in_(emplist)) show(emps)
not in
emplist = [10010, 10015, 10018] emps = session.query(Employee).filter(~Employee.emp_no.in_(emplist)) show(emps) emps = session.query(Employee).filter(Employee.emp_no.notin_(emplist)) show(emps)
like ,少用
emps = session.query(Employee).filter(Employee.last_name.like('P%')) show(emps)
not like ,少用
emps = session.query(Employee).filter(Employee.last_name.notlike('P%'))
ilike 忽略大小寫
emps = session.query(Employee).filter(Employee.last_name.ilike('P%')) show(emps)
排序
升序
emps = session.query(Employee).filter(Employee.emp_no > 10010).order_by(Employee.emp_no) emps = session.query(Employee).filter(Employee.emp_no > 10010).order_by(Employee.emp_no.asc()) show(emps)
降序
emps = session.query(Employee).filter(Employee.emp_no > 10010).order_by(Employee.emp_no.desc())
show(emps)
多列排序
emps = session.query(Employee).filter(Employee.emp_no > 10010).order_by(Employee.last_name).order_by(Employee.emp_no.desc()) show(emps) emps = session.query(Employee).filter(Employee.emp_no > 10010).order_by(Employee.last_name.desc(), Employee.emp_no.asc()) show(emps)
分頁
emps = session.query(Employee).limit(4) show(emps)
emps = session.query(Employee).limit(4).offset(18) show(emps)
消費者方法
總共的行數
emps = session.query(Employee) print(len(list(emps))) # 返回大量的結果集,然後轉換list print(emps.count()) # 聚合函式count(*)的查詢
取所有資料
emps = session.query(Employee) print(emps.all()) # 返回列表,查不到返回空列表
取首行 : first方法本質上就是limit語句
emps = session.query(Employee) print(emps.limit(1).one()) #返回一行 print(emps.one()) #如果查詢結果是多行拋異常
刪除
# 刪除 delete by query session.query(Employee).filter(Employee.emp_no > 10018).delete() #session.commit() # 提交則刪除
聚合,分組
聚合
from sqlalchemy import func query = session.query(func.count(Employee.emp_no)) print(query.one()) # 只能有一行結果 print(query.scalar()) # 取one()返回元組的第一個元素
max,min,avg
print(session.query(func.max(Employee.emp_no)).scalar()) print(session.query(func.min(Employee.emp_no)).scalar()) print(session.query(func.avg(Employee.emp_no)).scalar())
分組
print(session.query(Employee.gender,func.count(Employee.emp_no)).group_by(Employee.gender).all())
關聯查詢
有兩張表,其中的物件多對多關聯,就要建立第三張表。
如果是一對多關聯,就在多的一端建立外來鍵。
有一個員工,即屬於A部門,又屬於B部門,同時每個部門都有許多員工,這就是多對多
先把這些表的Model類和欄位屬性建立起來。
class Employee(Base): # 指定表名 __tablename__ = 'employees' # 定義屬性對應欄位 emp_no = Column(Integer, primary_key=True) birth_date = Column(Date, nullable=False) first_name = Column(String(14), nullable=False) last_name = Column(String(16), nullable=False) gender = Column(Enum(GenderEnum), nullable=False) hire_date = Column(Date, nullable=False) # 第一引數是欄位名,如果和屬性名不一致,一定要指定 # age = Column('age', Integer) def __repr__(self): return "{} no={} name={} {} gender={}".format( self.__class__.__name__, self.emp_no, self.first_name, self.last_name, self.gender.value ) class Department(Base): __tablename__ = 'departments' dept_no = Column(String(4), primary_key=True) dept_name = Column(String(40), nullable=False, unique=True) def __repr__(self): return "{} no={} name={}".format(type(self).__name__, self.dept_no, self.dept_name) class Dept_emp(Base): __tablename__ = "dept_emp" emp_no = Column(Integer, ForeignKey('employees.emp_no', ondelete='CASCADE'), primary_key=True) dept_no = Column(String(4), ForeignKey('departments.dept_no', ondelete='CASCADE'), primary_key=True) from_date = Column(Date, nullable=False) to_date = Column(Date, nullable=False) def __repr__(self): return "{} empno={} deptno={}".format(type(self).__name__, self.emp_no, self.dept_no)
查詢10010員工的所在部門和標號資訊
results = session.query(Employee,Dept_emp).filter((Employee.emp_no == Dept_emp.emp_no) & (Employee.emp_no == 10010)).all()
show(results)
這種方式產生隱式連線的語句
使用join
results = session.query(Employee).join(Dept_emp).filter(Employee.emp_no == 10010).all()
results = session.query(Employee).join(Dept_emp, Employee.emp_no == Dept_emp.emp_no).filter(Employee.emp_no == 10010).all() print(results)
返回的都只有一行資料。它們生成的SQL語句是一樣的,執行該SQL語句返回確實是2行記錄,可以Python中的返回值列表中只有一個元素。
原因在於 query(Employee) 這個只能返回一個實體物件中去,為了解決這個問題,需要修改實體類Employee,增加屬性用來存放部門資訊
sqlalchemy.orm.relationship(實體類名字串)
from sqlachemy import relationship class Employee(Base): # 指定表名 __tablename__ = 'employees' # 定義屬性對應欄位 emp_no = Column(Integer, primary_key=True) birth_date = Column(Date, nullable=False) first_name = Column(String(14), nullable=False) last_name = Column(String(16), nullable=False) gender = Column(Enum(GenderEnum), nullable=False) hire_date = Column(Date, nullable=False) # 第一引數是欄位名,如果和屬性名不一致,一定要指定 # age = Column('age', Integer) departments = relationship("Dept_emp") def __repr__(self): return "{} no={} name={} {} gender={} depts={}".format( self.__class__.__name__, self.emp_no, self.first_name, self.last_name, self.gender.value,self.departments )
查詢
# 第一種 # results = session.query(Employee).join(Dept_emp).filter(Employee.emp_no == Dept_emp.emp_no).filter(Employee.emp_no == 10010) # 第二種 # results = session.query(Employee).join(Dept_emp, Employee.emp_no == Dept_emp.emp_no).filter(Employee.emp_no == 10010) # 第三種 results = session.query(Employee).join(Dept_emp, (Employee.emp_no == Dept_emp.emp_no) & (Employee.emp_no == 10010)) show(results.all())
第一種方法join(Dept_emp)中沒有等值條件,會自動生成一個等值條件,如果後面有filter,哪怕是filter(Employee.emp_no == Dept_emp.emp_no),這個條件會在where中出現。第一種這種自動增加join的等值條件的方式不好,不要這麼寫
第二種方法在join中增加等值條件,阻止了自動的等值條件的生成。這種方式推薦
第三種方法就是第二種,這種方式也可以
只要不訪問departments屬性,就不會查dept_emp這張表。
總結
在開發中,一般都會採用ORM框架,這樣就可以使用物件操作表了。
定義表對映的類,使用Column的描述器定義類屬性,使用ForeignKey來定義外來鍵約束。
如果在一個物件中,想檢視其它表對應的物件的內容,就要使用relationship來定義關係。
是否使用外來鍵約束?
1、力挺派
能使資料保證完整性一致性
2、嫌棄派
開發難度增加,大資料的時候影響插入、修改、刪除的效率。
在業務層保證資料的一致性。