一、數據準備：配置數據庫、創建數據庫引擎、創建基類、創建session

from sqlalchemy import create_engine, Column, Integer, ForeignKey, String, TEXT, Boolean, DATE, DECIMAL from sqlalchemy.ext.declarative import declarative_base from datetime import date from sqlalchemy.orm import sessionmaker,relationship,backref#配置數據庫 HOSTNAME = '127.0.0.1' PORT = '3306' DATABASE = 'test' USERNAME = 'root' PASSWORD = 'root' DB_URI = 'mysql+pymysql://{}:{}@{}:{}/{}?charset=utf8mb4'.format(USERNAME, PASSWORD, HOSTNAME, PORT, DATABASE)#創建數據庫引擎 engine=create_engine(DB_URI)#創建基類，所有的模型類繼承基類 Base=declarative_base(engine)#創建session session=sessionmaker(engine)()

二、創建模型類（部門表和員工表）

#創建模型類 class Dept(Base):__tablename__='t_dept2'dept_no=Column(name='dept_no',type_=Integer,primary_key=True,autoincrement=True)dept_name=Column(name='dept_name',type_=String(20))city=Column(name='city',type_=String(20))#代表當前部門下的所有員工列表，這種寫法不是最好的，最優的寫法只要在一個對象中關聯就可以了#emp=relationship('Emp') #參數必須是另一個關聯模型類的類名def __str__(self):return f'部門編號:{self.dept_no}部門:{self.dept_name}城市:{self.city}'class Emp(Base):__tablename__='t_emp2'emp_no=Column(name='emp_no',type_=Integer,primary_key=True,autoincrement=True)emp_name=Column(name='emp_name',type_=String(20))hire_date=Column(name='hire_date',type_=DATE)sal=Column(name='sal',type_=DECIMAL(10,2))#todo 設置外鍵關聯，在從表中增加一個字段，指定這個字段外鍵的是哪個表的哪個字段就可以了。從表中外鍵的字段，必須和主表的主鍵字段類型保持一致。dept_no=Column(ForeignKey('t_dept2.dept_no',ondelete='CASCADE'),name='dept_no',type_=Integer)#代表員工所屬的部門信息，backref:反向關聯的屬性名,lazy='dynamic:懶加載dept=relationship('Dept',backref=backref('emps',lazy='dynamic')) #參數必須是另一個關聯模型類的類名def __str__(self):return f'員工編號:{self.emp_no}員工姓名:{self.emp_name}員工入職時間:{self.hire_date}員工薪資:{self.sal}'#根據模型類建表 #Base.metadata.create_all() #Base.metadata.drop_all()

group_by
根據某個字段進行分組
having
having是對分組查找結果作進一步過濾

案例

def test_group():# 統計每個工資級別下有多少員工result=session.query(Emp.sal,func.count(Emp.emp_no)).group_by(Emp.sal).all()print(result)#統計每個部門下有多少員工result1=session.query(Emp.dept_no,func.count(Emp.emp_no)).group_by(Emp.dept_no).all()print(result1)# 統計每個工資級別下有多少員工,只統計7000以上的result2 = session.query(Emp.sal, func.count(Emp.emp_no)).group_by(Emp.sal).having(Emp.sal>7000).all()print(result2)

執行結果：

[(Decimal('8888.88'), 1), (Decimal('5555.88'), 1), (Decimal('6000.00'), 1), (Decimal('10000.00'), 2), (Decimal('7777.77'), 1), (Decimal('8000.00'), 1)] [(1, 5), (2, 2)] [(Decimal('8888.88'), 1), (Decimal('10000.00'), 2), (Decimal('7777.77'), 1), (Decimal('8000.00'), 1)]

子查詢
子查詢即select語句中還有select。
那么在sqlalchemy中，要實現一個子查詢，需以下幾個步驟：

1. 將子查詢按照傳統的方式寫好查詢代碼，然后在query對象后面執行subquery方法，將這個查詢變成一個子查詢。

2. 在子查詢中，將以后需要用到的字段通過label方法，取個別名。

3. 在父查詢中，如果想要使用子查詢的字段，那么可以通過子查詢的返回值上的c屬性拿到（c=Column）。

案例

def test_subquery():#查詢和django這個員工，入職時間、工資都相同的其他員工#第一步：寫子查詢result3=session.query(Emp.hire_date.label('h_d'),Emp.sal.label('sal')).filter(Emp.emp_name=='django').subquery()print(result3)#第二步：寫父查詢result4=session.query(Emp).filter(Emp.hire_date==result3.c.h_d,Emp.sal==result3.c.sal).all()print(result4) #查看sql語句for i in result4:print(i) SELECT t_emp2.hire_date AS h_d, t_emp2.sal AS sal FROM t_emp2 WHERE t_emp2.emp_name = %(emp_name_1)s[<__main__.Emp object at 0x000001BDEDEEBAF0>, <__main__.Emp object at 0x000001BDEDEEBB50>]員工編號:6員工姓名:django員工入職時間:2021-12-12員工薪資:10000.00 員工編號:13員工姓名:mysql員工入職時間:2021-12-12員工薪資:10000.00

總結

以上是生活随笔為你收集整理的【Flask】ORM高级操作之分组、过滤和子查询的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。