Python SQLALchemy框架

時間  2020-12-18
標籤 html python mysql sql 數據庫 flask session 架構 oracle 框架 欄目 Python 简体版
原文   原文鏈接

SQLALchemy

　　 SQLALchemy是Python中的一款優秀的ORM框架，它能夠做用於任何第三方Web框架，如flask，tornado等框架。html

　　 SQLALchemy相較於DjangoORM來講更加的貼近原生SQL語句，所以學習難度較低。python

　　 img

　　

組成部分 描述
Engine 框架引擎
Connection Pooling 數據庫連接池
Dialect 數據庫DB API種類
Schema/Types 架構&類型
SQL Exprression Language SQL表達式語言

　　 下載SQLALchemy模塊：mysql

pip install sqlalchemy

　　 值得注意的是SQLALchemy必須依賴其餘操縱數據的模塊，Dialect用於和數據API進行交流，根據配置文件的不一樣調用不一樣的數據庫API，從而實現對數據庫的操做，如：sql

MySQL-Python
    mysql+mysqldb://<user>:<password>@<host>[:<port>]/<dbname>
    
pymysql
    mysql+pymysql://<username>:<password>@<host>/<dbname>[?<options>]
    
MySQL-Connector
    mysql+mysqlconnector://<user>:<password>@<host>[:<port>]/<dbname>
    
cx_Oracle
    oracle+cx_oracle://user:pass@host:port/dbname[?key=value&key=value...]
    
更多：http://docs.sqlalchemy.org/en/latest/dialects/index.html

表操做

　　 SQLALchemy中不容許修改表結構，若是修改表結構則須要刪除舊錶，再建立新表：數據庫

#!/usr/bin/env python
# -*- coding:utf-8 -*-

import datetime

from sqlalchemy import Column, Integer, String, DateTime, UniqueConstraint, Index
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base

# 基礎類
Base = declarative_base()

# 建立引擎
engine = create_engine(
    "mysql+pymysql://root@127.0.0.1:3306/db1?charset=utf8",
    # "mysql+pymysql://root:123@127.0.0.1:3306/db1?charset=utf8",  # 有密碼時
    max_overflow=0,  # 超過鏈接池大小外最多建立的鏈接
    pool_size=5,  # 鏈接池大小
    pool_timeout=30,  # 池中沒有線程最多等待的時間，不然報錯
    pool_recycle=-1  # 多久以後對線程池中的線程進行一次鏈接的回收（重置）
)


class Users(Base):
    __tablename__ = 'users'

    id = Column(Integer, primary_key=True)
    name = Column(String(32), index=True, nullable=False)
    age = Column(Integer,nullable=False)
    phone = Column(String(11))
    addr = Column(String(64), nullable=True)
    create_time = Column(DateTime, default=datetime.datetime.now)  # 必定不要加括號

    __table_args__ = (
        UniqueConstraint("id", "name"),  # 建立聯合惟一 可指定name給個別名
        Index("phone", "addr", unique=True),  # 建立聯合惟一索引  可指定name給個別名
    )

    def __str__(self):
        return "object:<id:%s name:%s>" % (self.id, self.name)


def create_tb():
    """
    建立表
    :return:
    """
    Base.metadata.create_all(engine)


def drop_tb():
    """
    刪除表
    :return:
    """
    Base.metadata.drop_all(engine)


if __name__ == '__main__':
    drop_tb()
    create_tb()

連接庫

　　 表建立好以後，開始連接庫。flask

from sqlalchemy.orm import sessionmaker
from sqlalchemy.orm import scoped_session

# 導入引擎，模型表等
from models import *

# 經過Session綁定引擎和數據庫創建關係
Session = sessionmaker(bind=engine)
# 建立連接池，使用session便可爲當前線程拿出一個連接對象。內部採用threading.local進行隔離
session = scoped_session(Session)

單表記錄

新增記錄

　　 新增單條記錄：session

from sqlalchemy.orm import scoped_session
from sqlalchemy.orm import sessionmaker

# 導入引擎，模型表等
from models import *

# 經過Session綁定引擎和數據庫創建關係
Session = sessionmaker(bind=engine)
# 建立連接池，使用session便可爲當前線程拿出一個連接對象。內部採用threading.local進行隔離
session = scoped_session(Session)


user_obj = Users(name="user001", phone="15125352333",age=23, addr="China")
session.add(user_obj)

# 提交
session.commit()

# 關閉連接（可以使用session.remove()）
session.close()

修改記錄

　　 修改記錄：架構

from sqlalchemy.orm import scoped_session
from sqlalchemy.orm import sessionmaker

# 導入引擎，模型表等
from models import *

# 經過Session綁定引擎和數據庫創建關係
Session = sessionmaker(bind=engine)
# 建立連接池，使用session便可爲當前線程拿出一個連接對象。內部採用threading.local進行隔離
session = scoped_session(Session)

# 修更名字
session.query(Users).filter_by(id=1).update({"name": "USER001"})
# 修改年齡，使用+號，默認爲"fetch"，表明只容許int類型使用+號
session.query(Users).filter_by(id=1).update({"age": Users.age + 1},synchronize_session="fetch")
# 修改地址,使用+號，因爲是字符類型，因此要修改synchronize_session=False
session.query(Users).filter_by(id=1).update({"addr":Users.addr + "BeiJing"},synchronize_session=False)

# 提交
session.commit()

# 關閉連接
session.close()

刪除記錄

　　 刪除案例：oracle

from sqlalchemy.orm import scoped_session
from sqlalchemy.orm import sessionmaker

# 導入引擎，模型表等
from models import *

# 經過Session綁定引擎和數據庫創建關係
Session = sessionmaker(bind=engine)
# 建立連接池，使用session便可爲當前線程拿出一個連接對象。內部採用threading.local進行隔離
session = scoped_session(Session)

session.query(Users).filter_by(id=2).delete()

# 提交
session.commit()

# 關閉連接
session.close()

批量增長

　　 批量增長：框架

from sqlalchemy.orm import scoped_session
from sqlalchemy.orm import sessionmaker

# 導入引擎，模型表等
from models import *

# 經過Session綁定引擎和數據庫創建關係
Session = sessionmaker(bind=engine)
# 建立連接池，使用session便可爲當前線程拿出一個連接對象。內部採用threading.local進行隔離
session = scoped_session(Session)

# 批量增長
session.add_all([
    Users(name="user002",age=21,phone="13269867233",addr="ShangHai"),
    Users(name="user003",age=18,phone="13269867234",addr="GuangZhou"),
    Users(name="user003",age=24,phone="13269867235",addr="ChongQing"),
])

# 提交
session.commit()

# 關閉連接
session.close()

單表查詢

基本查詢

　　 基本查詢：

from sqlalchemy.orm import scoped_session
from sqlalchemy.orm import sessionmaker

# 導入引擎，模型表等
from models import *

# 經過Session綁定引擎和數據庫創建關係
Session = sessionmaker(bind=engine)
# 建立連接池，使用session便可爲當前線程拿出一個連接對象。內部採用threading.local進行隔離
session = scoped_session(Session)

# 查詢
# -- 查全部 --
result_01 = session.query(Users).all()
# -- 過濾 --
result_02 = session.query(Users).filter(Users.name == "USER001").all()  # Python表達式的形式過濾
result_03 = session.query(Users).filter_by(name="user002").all()  # ORM形式過濾
result_04 = session.query(Users).filter_by(name="user003").first()  # ORM形式過濾 取第一個

print(result_01) # [<models.Users>,<models.Users>,<models.Users>]
print(result_02)
print(result_03)
print(result_04) # object:<id:3 name:user003>  經過__str__拿到結果

# 提交
session.commit()

# 關閉連接
session.close()

其餘過濾

　　 條件查詢：

from sqlalchemy.orm import scoped_session
from sqlalchemy.orm import sessionmaker

# 導入引擎，模型表等
from models import *

# 經過Session綁定引擎和數據庫創建關係
Session = sessionmaker(bind=engine)
# 建立連接池，使用session便可爲當前線程拿出一個連接對象。內部採用threading.local進行隔離
session = scoped_session(Session)

# 只拿某字段
result_00 = session.query(Users.name,Users.age).first()
print(result_00)

# and(用逗號或者用and_)
result_01 = session.query(Users).filter( Users.id > 1,Users.age < 23).all()
print(result_01)

from sqlalchemy import and_
result_02 = session.query(Users).filter(and_( Users.id > 1,Users.age < 23)).all()
print(result_02)

# or
from sqlalchemy import  or_
result_03 = session.query(Users).filter(or_(Users.id > 3,Users.age < 23)).all()
print(result_03)

# and與or的組合使用
result_04 = session.query(Users).filter(or_(
    Users.id > 1,
    and_(Users.id > 2, Users.age < 24)
)).all()
print(result_04)

# 範圍
result_05 = session.query(Users).filter(Users.age.between(18,24)).all()
print(result_05)

# 包含
result_06 = session.query(Users).filter(Users.age.in_([18,21,24])).all()
print(result_06)

# 取反 ~
result_07 = session.query(Users).filter(~Users.age.in_([18,21,24])).all()
print(result_07)

# 通配符
result_08 = session.query(Users).filter(Users.name.like("us%")).all()
print(result_08)

# 分頁
result_09 = session.query(Users).all()[0:1]
print(result_09)

# 排序
result_10 = session.query(Users).order_by(Users.id.desc()).all()  # 倒序
print(result_10)

result_11 = session.query(Users).order_by(Users.id.asc()).all()  # 正序
print(result_11)

# 分組
result_12 = session.query(Users).group_by(Users.id).all()
print(result_12)


# 聚合函數
from sqlalchemy.sql import func
result_13 = session.query(
    func.max(Users.age),
    func.sum(Users.age),
    func.min(Users.age),
).group_by(Users.name).having(func.max(Users.age > 12)).all()
print(result_13)

# 提交
session.commit()

# 關閉連接
session.close()

多表相關

一對多

　　 首先是創建一對多的關係，使用relationship作邏輯一對多，不會在物理表中建立關係，可是能夠經過該字段進行增刪改查：

#!/usr/bin/env python
# -*- coding:utf-8 -*-

from sqlalchemy import Column, Integer, String, ForeignKey
from sqlalchemy.orm import relationship
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base

# 基礎類
Base = declarative_base()

# 建立引擎
engine = create_engine(
    "mysql+pymysql://root@127.0.0.1:3306/db1?charset=utf8",
    # "mysql+pymysql://root:123@127.0.0.1:3306/db1?charset=utf8",  # 有密碼時
    max_overflow=0,  # 超過鏈接池大小外最多建立的鏈接
    pool_size=5,  # 鏈接池大小
    pool_timeout=30,  # 池中沒有線程最多等待的時間，不然報錯
    pool_recycle=-1  # 多久以後對線程池中的線程進行一次鏈接的回收（重置）
)


class Classes(Base):
    __tablename__ = "classes"

    id = Column(Integer, primary_key=True)
    name = Column(String(32), nullable=False)


class Students(Base):
    __tablename__ = "students"

    id = Column(Integer, primary_key=True)
    name = Column(String(32), nullable=False)
    # 真實約束字段：避免髒數據寫入，在物理表中會建立真實字段關係
    # 可選級聯操做：CASCADE，DELETE、RESTRICT
    fk_class = Column(Integer, ForeignKey("classes.id",ondelete="CASCADE",onupdate="CASCADE"))
    # 邏輯關係字段：不會在真實物理表中建立字段，可是能夠經過該邏輯字段進行增刪改查
    # backref:反向查詢的名字
    re_class = relationship("Classes",backref="students")

def create_tb():
    """
    建立表
    :return:
    """
    Base.metadata.create_all(engine)


def drop_tb():
    """
    刪除表
    :return:
    """
    Base.metadata.drop_all(engine)


if __name__ == '__main__':
    drop_tb()
    create_tb()

　　 經過邏輯字段進行增長：

from sqlalchemy.orm import scoped_session
from sqlalchemy.orm import sessionmaker

# 導入引擎，模型表等
from models import *

# 經過Session綁定引擎和數據庫創建關係
Session = sessionmaker(bind=engine)
# 建立連接池，使用session便可爲當前線程拿出一個連接對象。內部採用threading.local進行隔離
session = scoped_session(Session)


session.add_all(
    [
        Students(name="學生01", re_class=Classes(name="一年級一班")),  # 自動填入fk_class
        Students(name="學生02", re_class=Classes(name="一年級二班")),
    ]
)

# 提交
session.commit()

# 關閉連接
session.close()

多對多

　　 多對多也使用relationship作邏輯多對多，不會在物理表中建立關係，可是能夠經過該字段進行增刪改查。

　　 使用relationship時，傳入指定手動生成的第三張表，表明這是多對多關係：

#!/usr/bin/env python
# -*- coding:utf-8 -*-

from sqlalchemy import Column, Integer, String, ForeignKey, UniqueConstraint
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import relationship

# 基礎類
Base = declarative_base()

# 建立引擎
engine = create_engine(
    "mysql+pymysql://root@127.0.0.1:3306/db1?charset=utf8",
    # "mysql+pymysql://root:123@127.0.0.1:3306/db1?charset=utf8",  # 有密碼時
    max_overflow=0,  # 超過鏈接池大小外最多建立的鏈接
    pool_size=5,  # 鏈接池大小
    pool_timeout=30,  # 池中沒有線程最多等待的時間，不然報錯
    pool_recycle=-1  # 多久以後對線程池中的線程進行一次鏈接的回收（重置）
)


class Classes(Base):
    __tablename__ = "classes"

    id = Column(Integer, primary_key=True)
    name = Column(String(32), nullable=False)


class Students(Base):
    __tablename__ = "students"

    id = Column(Integer, primary_key=True)
    name = Column(String(32), nullable=False)
    # 可選級聯操做：CASCADE，DELETE、RESTRICT
    fk_class = Column(Integer, ForeignKey("classes.id", ondelete="CASCADE", onupdate="CASCADE"))
    # 邏輯關係字段：不會在真實物理表中建立字段，可是能夠經過該邏輯字段進行增刪改查
    # backref:反向查詢的名字
    re_class = relationship("Classes", backref="students")


class Teachers(Base):
    __tablename__ = "teachers"

    id = Column(Integer, primary_key=True)
    name = Column(String(32), nullable=False)
    # 邏輯字段M2M:指定第三張表，secondary參數爲__tablename__，反向查詢爲teachers
    re_class = relationship("Classes", secondary="teachersm2mclasses", backref="teachers")


class TeachersM2mClasses(Base):
    __tablename__ = "teachersm2mclasses"

    id = Column(Integer, primary_key=True)
    teacher_id = Column(Integer, ForeignKey("teachers.id"))
    class_id = Column(Integer, ForeignKey("classes.id"))

    __table_args__ = (
        UniqueConstraint("teacher_id", "class_id"),  # 建立聯合惟一 可指定name給個別名
    )


def create_tb():
    """
    建立表
    :return:
    """
    Base.metadata.create_all(engine)


def drop_tb():
    """
    刪除表
    :return:
    """
    Base.metadata.drop_all(engine)


if __name__ == '__main__':
    drop_tb()
    create_tb()

　　 用一個列表，將班級的記錄對象放進去，你能夠用多種增長方式，使用邏輯字段添加或本身操縱第三張表：

from sqlalchemy.orm import scoped_session
from sqlalchemy.orm import sessionmaker

# 導入引擎，模型表等
from models import *

# 經過Session綁定引擎和數據庫創建關係
Session = sessionmaker(bind=engine)
# 建立連接池，使用session便可爲當前線程拿出一個連接對象。內部採用threading.local進行隔離
session = scoped_session(Session)

session.add_all(
    [
        Teachers(name="老師01",re_class=[ 
            session.query(Classes).filter_by(id=1).first()
        ]),
        Teachers(name="老師02",re_class=[
            session.query(Classes).filter_by(id=1).first()
        ]),
        Teachers(name="老師03",re_class=[
            session.query(Classes).filter_by(id=2).first()
        ]),
    ]
)

# 提交
session.commit()

# 關閉連接
session.close()

組合查詢

　　 組合查詢將兩張表用笛卡爾積的效果顯現出來：

from sqlalchemy.orm import scoped_session
from sqlalchemy.orm import sessionmaker

# 導入引擎，模型表等
from models import *

# 經過Session綁定引擎和數據庫創建關係
Session = sessionmaker(bind=engine)
# 建立連接池，使用session便可爲當前線程拿出一個連接對象。內部採用threading.local進行隔離
session = scoped_session(Session)

# 必須用filter，獲取所有也是，不可使用all由於他會返回一個list，list不具有union_all
# 使用filter返回的對象是：<class 'sqlalchemy.orm.query.Query'>
# 而且query中必須單拿某一個字段，若是不指定字段就直接返回對象

s = session.query(Students.name).filter()
t = session.query(Teachers.name).filter()
c = session.query(Classes.name).filter()
ret = s.union_all(t).union_all(c).all()  # 用列表顯示
print(ret)
# [('學生01',), ('學生02',), ('老師01',), ('老師02',), ('老師03',), ('一年級一班',), ('一年級二班',)]

# 提交
session.commit()

# 關閉連接
session.close()

連表查詢

　　 使用join進行連表查詢：

from sqlalchemy.orm import scoped_session
from sqlalchemy.orm import sessionmaker

# 導入引擎，模型表等
from models import *

# 經過Session綁定引擎和數據庫創建關係
Session = sessionmaker(bind=engine)
# 建立連接池，使用session便可爲當前線程拿出一個連接對象。內部採用threading.local進行隔離
session = scoped_session(Session)

# 手動指定條件查詢
result = session.query(Students.name, Classes.name).filter(Students.id == Classes.id).all()
for i in result:
    print(i)

# 鏈接查詢,同上，內部自動指定 Students.fk_class == Classes.id 的條件
result = session.query(Students.name, Classes.name).join(Classes).all()
# 至關於：result = session.query(Students.name,Classes.name).join(Classes, Students.fk_class == Classes.id).all()
for i in result:
    print(i)

# 左連接查詢,即便有同窗沒有班級也拿出來
result = session.query(Students.name, Classes.name).join(Classes, isouter=True).all()
for i in result:
    print(i)

# 若是想查看有哪些班級沒有同窗，就換一個位置
result = session.query(Students.name, Classes.name).join(Students, isouter=True).all()
for i in result:
    print(i)

# 三表查詢,須要本身指定條件
result = session.query(Teachers.name, Classes.name, TeachersM2mClasses.id) \
    .join(Teachers, TeachersM2mClasses.teacher_id == Teachers.id) \
    .join(Classes, TeachersM2mClasses.class_id == Classes.id) \
    .filter()  # 查看原生語句
print(result)
for i in result:
    print(i)

# 提交
session.commit()

# 關閉連接
session.close()

正反查詢

　　 上面是使用join進行的連表查詢，其實也可使用邏輯字段relationship查詢。

　　

from sqlalchemy.orm import scoped_session
from sqlalchemy.orm import sessionmaker

# 導入引擎，模型表等
from models import *

# 經過Session綁定引擎和數據庫創建關係
Session = sessionmaker(bind=engine)
# 建立連接池，使用session便可爲當前線程拿出一個連接對象。內部採用threading.local進行隔離
session = scoped_session(Session)

# 正向：查看第一個老師都在哪些班級（經過邏輯字段的名字）
result = session.query(Teachers).first()
# result.re_class是一個列表，存了有關該老師所在的班級 <class 'sqlalchemy.orm.collections.InstrumentedList'>

for class_obj in result.re_class:  # 查看其全部的班級
    print(class_obj.name)

# 反向：查看第一個班級下都有哪些老師，都有哪些學生（經過邏輯字段中的backref參數進行反向查詢）
result = session.query(Classes).first()

# 看老師
for teacher_obj in result.teachers:
    print(teacher_obj.name)

# 看學生
for student_obj in result.students:
    print(student_obj.name)

# 提交
session.commit()

# 關閉連接
session.close()

正反方法

　　 使用邏輯字段relationship可擁有一些方法執行增刪改。

　　 因爲邏輯字段是一個相似列表的存在，因此列表的方法都能用。

print(type(老師對象.班級列表))

# <class 'sqlalchemy.orm.collections.InstrumentedList'>

　　 好比使用extend方法增長老師的班級：

# 給老師增長班級

result = session.query(Teachers).first()


# extend方法：
result.re_class.extend([
    Classes(name="三年級一班",),
    Classes(name="三年級二班",),
])

　　 使用remove方法刪除老師的班級：

# 減小老師所在的班級
result = session.query(Teachers).first()

# 待刪除的班級對象,集合查找比較快
delete_class_set = {
    session.query(Classes).filter_by(id=7).first(),
    session.query(Classes).filter_by(id=8).first(),
}

# 循換老師所在的班級
# remove方法：
for class_obj in result.re_class:
    if class_obj in delete_class_set:
        result.re_class.remove(class_obj)

　　 使用clear清空老師所對應的班級：

# 拿出一個老師
result = session.query(Teachers).first()

result.re_class.clear()

原生SQL

查看SQL命令

　　 若是一條查詢語句是以filter結尾，則返回結果對象的__str__方法中都是SQL語句：

result = session.query(Teachers).filter()
print(result)

# SELECT teachers.id AS teachers_id, teachers.name AS teachers_name 
# FROM teachers

　　 若是是all結尾，返回的就是一個列表，first結尾也是一個列表：

result = session.query(Teachers).all()
print(result)

# [<models.Teachers object at 0x00000178EB0B5550>, <models.Teachers object at 0x00000178EB0B5518>, <models.Teachers object at 0x00000178EB0B5048>]

執行SQL語句

　　 執行原生SQL：

from sqlalchemy.orm import scoped_session
from sqlalchemy.orm import sessionmaker

# 導入引擎，模型表等
from models import *

# 經過Session綁定引擎和數據庫創建關係
Session = sessionmaker(bind=engine)
# 建立連接池，使用session便可爲當前線程拿出一個連接對象。內部採用threading.local進行隔離
session = scoped_session(Session)

cursor = session.execute(r"select * from students where id <= (:num)",params={"num":2})
print(cursor.fetchall())

# 提交
session.commit()

# 關閉連接
session.close()
相關文章
相關標籤/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公眾號
   歡迎關注本站公眾號,獲取更多信息
聯繫我們 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程