SQLAlchemy學習

一SQLAlchemy簡介

SQLAlchemy是一個基於Python實現的ORM框架。該框架創建在 DB API之上，使用關係對象映射進行數據庫操做，簡言之即是：將類和對象轉換成SQL，而後使用數據API執行SQL並獲取執行結果。

1.與Django中models的區別

不少小夥伴說SQLAlchemy不如Django的models好用，這裏咱們須要知道。

Models其實只是配置和使用比較簡單，畢竟是Django自帶的ORM框架，可是兼容性遠不如SQLAchemy，真正算得上全面的ORM框架必然是SQLAlchemy。

不管使用什麼ORM框架，其實都是爲了方便不熟練數據庫使用的同窗，最推薦的仍是使用原生的SQL語句，也建議你們攻克SQL難關。

2.SQLAlchemy組成

組成部分：

• Engine，框架的引擎

• Connection Pooling ，數據庫鏈接池

• Dialect，選擇鏈接數據庫的DB API種類

• Schema/Types，架構和類型

• SQL Exprression Language，SQL表達式語言

SQLAlchemy自己沒法操做數據庫，其必須以來pymsql等第三方插件，Dialect用於和數據API進行交流，根據配置文件的不一樣調用不一樣的數據庫API，從而實現對數據庫的操做，如：

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

2、SQLAlchemy使用

1.執行原生sql語句

經過SQLAlchemy執行源生的sql語句

方式一：

from sqlalchemy import create_engine
​
engine = create_engine(
    "mysql+pymysql://root:123@127.0.0.1:3306/sqlalchemy01?charset=utf8",
    max_overflow=0,  # 超過鏈接池大小外最多建立的鏈接
    pool_size=5,  # 鏈接池大小
    pool_timeout=30,  # 池中沒有線程最多等待的時間，不然報錯
    pool_recycle=-1  # 多久以後對線程池中的線程進行一次鏈接的回收（重置）
)
​
​
def task():
    conn = engine.raw_connection()
    cursor = conn.cursor()
    cursor.execute(
        "select * from t1"
    )
    result = cursor.fetchall()
    print(">>>",result)
    cursor.close()
    conn.close()
​
task()

方式二：

from sqlalchemy import create_engine
​
engine = create_engine("mysql+pymysql://root:123@127.0.0.1:3306/sqlalchemy01", max_overflow=0, pool_size=5)
​
​
def task():
    conn = engine.connect()
    with conn:
        cur = conn.execute(
            "select * from t1"
        )
        result = cur.fetchall()
        print(result)
​
task()

方式三

from sqlalchemy import create_engine
​
engine = create_engine("mysql+pymysql://root:123@127.0.0.1:3306/sqlalchemy01", max_overflow=0, pool_size=5)
​
​
def task():
    cur = engine.execute("select * from t1")
    result = cur.fetchall()
    cur.close()
    print(result)
​
task()

2.數據表的操做

經過sqlalchemy來建立表和刪除表

import datetime
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, Text, ForeignKey, DateTime, UniqueConstraint, Index
​
# 創建基礎類 R關係 M映射 類
Base = declarative_base()
​
​
class Users(Base):
    __tablename__ = 'users'  # 指定建立的表名
​
    # 寫字段
    id = Column(Integer, primary_key=True)
    name = Column(String(32), index=True, nullable=False)
    email = Column(String(32), unique=True)
    # ctime = Column(DateTime, default=datetime.datetime.now)
    # extra = Column(Text, nullable=True)
​
    __table_args__ = (
        UniqueConstraint('id', 'name', name='uix_id_name'),  # 設置位移約束
        Index('ix_id_name', 'name', 'email'),  # 設置索引
    )
​
​
# 建立數據庫的引擎
engine = create_engine(
    "mysql+pymysql://root:123@127.0.0.1:3306/sqlalchemy01?charset=utf8",
    max_overflow=0,  # 超過鏈接池大小外最多建立的鏈接
    pool_size=5,  # 鏈接池大小
    pool_timeout=30,  # 池中沒有線程最多等待的時間，不然報錯
    pool_recycle=-1  # 多久以後對線程池中的線程進行一次鏈接的回收（重置）
)
​
# 檢索全部繼承Base的Object並在 engine 指向的數據庫中建立全部的表
Base.metadata.create_all(engine)
​
# 刪除全部的數據庫表
Base.metadata.drop_all(engine)

1.單表建立示例

import datetime
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, Text, ForeignKey, DateTime, UniqueConstraint, Index
​
# 創建基礎類 R關係 M映射 類
Base = declarative_base()
​
​
class Users(Base):
    __tablename__ = 'users'  # 指定建立的表名
​
    # 寫字段
    id = Column(Integer, primary_key=True)
    name = Column(String(32), index=True, nullable=False)
    email = Column(String(32), unique=True)
    # ctime = Column(DateTime, default=datetime.datetime.now)
    # extra = Column(Text, nullable=True)
​
    __table_args__ = (
        UniqueConstraint('id', 'name', name='uix_id_name'),  # 設置位移約束
        Index('ix_id_name', 'name', 'email'),  # 設置索引
    )
​
​
# 建立數據庫的引擎
engine = create_engine(
    "mysql+pymysql://root:123@127.0.0.1:3306/sqlalchemy01?charset=utf8",
    max_overflow=0,  # 超過鏈接池大小外最多建立的鏈接
    pool_size=5,  # 鏈接池大小
    pool_timeout=30,  # 池中沒有線程最多等待的時間，不然報錯
    pool_recycle=-1  # 多久以後對線程池中的線程進行一次鏈接的回收（重置）
)
​
# 檢索全部繼承Base的Object並在 engine 指向的數據庫中建立全部的表
Base.metadata.create_all(engine)
​
# 刪除全部的數據庫表
Base.metadata.drop_all(engine)

2.一對多示例

# ########## 一對多示例 ##########
class School(Base):
    __tablename__ = "school"
    id = Column(Integer,primary_key=True)
    name = Column(String(32),nullable=False)
​
​
class Student(Base):
    __tablename__ = "student"
    id = Column(Integer,primary_key=True)
    name = Column(String(32),nullable=False)
    school_id = Column(Integer,ForeignKey("school.id")) # 多對一關係存儲列
​
    # 與生成表結構無關，僅用於查詢方便
    school = relationship("School", backref='student')
    
engine = create_engine("mysql+pymysql://root:root@127.0.0.1:3306/sqlalchemy01?charset=utf8")
​
# 檢索全部繼承 Model 的Object 並在 engine 指向的數據庫中建立 全部的表
Model.metadata.create_all(engine)

3.多對多表結構建立

from sqlalchemy import Column,Integer,String,ForeignKey
from sqlalchemy.orm import relationship
​
class Girls(Model):
    __tablename__ = "girl"
    id = Column(Integer,primary_key=True)
    name = Column(String(32),nullable=False)
    # relationship
    g2b = relationship("Boys",backref="b2g",secondary="hotel")
​
​
class Boys(Model):
    __tablename__ = "boy"
    id = Column(Integer,primary_key=True)
    name = Column(String(32),nullable=False)
​
​
class Hotels(Model):
    __tablename__ = "hotel"
    id = Column(Integer,primary_key=True)
    boy_id = Column(Integer,ForeignKey("boy.id"))
    girl_id = Column(Integer,ForeignKey("girl.id"))
    
engine = create_engine("mysql+pymysql://root:root@127.0.0.1:3306/sqlalchemy01?charset=utf8")
​
# 檢索全部繼承 Model 的Object 並在 engine 指向的數據庫中建立 全部的表
Model.metadata.create_all(engine)

4.定義函數來建立和刪除表

def init_db():
    """
    根據類建立數據庫表
    :return: 
    """
    engine = create_engine(
        "mysql+pymysql://root:123@127.0.0.1:3306/s6?charset=utf8",
        max_overflow=0,  # 超過鏈接池大小外最多建立的鏈接
        pool_size=5,  # 鏈接池大小
        pool_timeout=30,  # 池中沒有線程最多等待的時間，不然報錯
        pool_recycle=-1  # 多久以後對線程池中的線程進行一次鏈接的回收（重置）
    )
​
    Base.metadata.create_all(engine)
​
​
def drop_db():
    """
    根據類刪除數據庫表
    :return: 
    """
    engine = create_engine(
        "mysql+pymysql://root:123@127.0.0.1:3306/s6?charset=utf8",
        max_overflow=0,  # 超過鏈接池大小外最多建立的鏈接
        pool_size=5,  # 鏈接池大小
        pool_timeout=30,  # 池中沒有線程最多等待的時間，不然報錯
        pool_recycle=-1  # 多久以後對線程池中的線程進行一次鏈接的回收（重置）
    )
​
    Base.metadata.drop_all(engine)
​
​
if __name__ == '__main__':
    drop_db()
    init_db()

3.記錄的增刪改查

數據庫記錄操做的兩種方式

from sqlalchemy.orm import sessionmaker
from sqlalchemy import create_engine
from models import Users
 
engine = create_engine("mysql+pymysql://root:123@127.0.0.1:3306/s6", max_overflow=0, pool_size=5)
############方式一#############
Session = sessionmaker(bind=engine)
 
# 每次執行數據庫操做時，都須要建立一個session
session = Session()
 
# ############# 執行ORM操做 #############
obj1 = Users(name="alex1")
session.add(obj1)
 
# 提交事務
session.commit()
# 關閉session
session.close()
​
###########方式二###########
# 方式二：支持線程安全，爲每一個線程建立一個session
            #               - threading.Local
            #               - 惟一標識
            # ScopedSession對象
            #       self.registry(), 加括號 建立session
            #       self.registry(), 加括號 建立session
            #       self.registry(), 加括號 建立session
            from greenlet import getcurrent as get_ident
Session = sessionmaker(bind=engine)
            session = scoped_session(Session,get_ident)
            # session.add
            # 操做
            session.remove()

1.單表的增刪改查

from day101_sqlAlchemy.SQLAlchemy02_create_table_single import engine,Users
from sqlalchemy.orm import sessionmaker
​
Session = sessionmaker(engine)  # 新建數據庫的查詢窗口
db_session = Session()  # 打開查詢窗口
​
# 增長單條數據
# u = Users(name="ryxiong")  # 新建insert語句 insert into
# db_session.add(u)  # 將insert語句移動到 db_session 查詢窗口
# db_session.commit()  # 執行查詢窗口中的全部語句
# db_session.close()  # 關閉查詢窗口
​
# 增長多條數據
# u_list = [Users(name="egon"),Users(name="alex")]
# db_session.add_all(u_list)  # 添加多條數據
# db_session.commit()
# db_session.close()
​
​
# 查詢數據
# res = db_session.query(Users).all()  # 查詢全部數據
# for user in res:
#     print(user.id,user.name)
​
# res = db_session.query(Users).first()  # 查詢符合條件的第一條數據
# print(res.id,res.name)  # 3 alex
​
# 並列條件查詢
# res = db_session.query(Users).filter(Users.id<3,Users.name=="ryxiong").all()
# for user in res:
#     print(user.id,user.name)  # 1 ryxiong
​
# res = db_session.query(Users).filter(Users.id<3,Users.name=="ryxiong").first()
# print(res.id,res.name)  # 1 ryxiong
​
​
# 修改數據
# db_session.query(Users).filter(Users.id==2).update({"name":"Egon"})
# db_session.commit()
​
# 刪除數據
db_session.query(Users).filter(Users.id==3).delete()
db_session.commit()

2.一對多的增刪改查

from day101_sqlAlchemy.SQLAlchemy03_create_table_foreignKey import engine,Student,School
from sqlalchemy.orm import sessionmaker
​
Session = sessionmaker(engine)  # 新建數據庫的查詢窗口
db_session = Session()  # 打開查詢窗口
​
# 增長一條數據
# school = School(name="新東方")
# db_session.add(school)
# db_session.commit()
​
# 在添加學生
# school_fir = db_session.query(School).filter(School.name=="新東方").first()
#
# student = Student(name="ryxiong",school_id=school_fir.id)
# db_session.add(student)
# db_session.commit()
​
# 1.添加數據 relationship 正向添加數據
# stu = Student(name="alex",school=School(name="藍翔"))
# db_session.add(stu)
# db_session.commit()
​
# 2.添加數據relationship 反向添加數據
# sch = School(name="藍翔")
# sch.student = [
#     Student(name="egon"),
#     Student(name="wusir")
# ]
# db_session.add(sch)
# db_session.commit()
​
​
# 查詢
# 1.relationship正向查詢
res = db_session.query(Student).all()
for stu in res:
    print(stu.id,stu.name,stu.school.name)
​
# 2.relationship反向查詢
res = db_session.query(School).all()
for sch in res:
    for stu in sch.student:
        print(sch.name,stu.id,stu.name)

3.多對多查詢

from day101_sqlAlchemy.SQLAlchemy04_create_table_M2M import engine,Boys,Girls
from sqlalchemy.orm import sessionmaker
​
Session = sessionmaker(engine)  # 新建數據庫的查詢窗口
db_session = Session()  # 打開查詢窗口
​
# 添加數據
# 1.relationship正向添加
# girl = Girls(name="Nancy",boy=[Boys(name="ryxiong"),Boys(name="alex")])
# db_session.add(girl)
# db_session.commit()
​
# 2.relationship反向添加
# boy = Boys(name="egon")
# boy.girl = [
#     Girls(name="羅玉鳳"),
#     Girls(name="朱利安"),
# ]
#
# db_session.add(boy)
# db_session.commit()
​
# 查詢數據
​
# 1.relationship 正向查詢
​
res = db_session.query(Girls).all()
for girl in res:
    for boy in girl.boy:
        print(girl.name,boy.name)
​
# 2.relationship 反向查詢
res = db_session.query(Boys).all()
for boy in res:
    for girl in boy.girl:
        print(boy.name,girl.name)

 4.記錄的高級查詢

from day101_sqlAlchemy.SQLAlchemy02_create_table_single import engine,Users
from sqlalchemy.sql import text
from sqlalchemy.orm import sessionmaker
from sqlalchemy import and_,or_
​
Session = sessionmaker(engine)  # 新建數據庫的查詢窗口
db_session = Session()  # 打開查詢窗口
​
# 邏輯條件查詢 and/or
# ret1 = db_session.query(Users).filter(and_(Users.id<3,Users.name=="ryxiong")).all()
# print(ret1)
# ret2 = db_session.query(Users).filter(or_(Users.id<2,Users.name=="egon")).all()
# print(ret2)
#
# ret3 = db_session.query(Users).filter(
#     or_(
#         and_(Users.id==1,Users.name=="ryxiong"),
#         and_(Users.id==2,Users.name=="egon")
#     )
# ).all()
# print(ret3)
​
​
# 查詢全部數據排序
# ret = db_session.query(Users).order_by(Users.id.asc()).all()  # 按照id升序排列
#
# print(ret)
​
​
# 查詢數據，指定查詢數據列，加入別名
# ret = db_session.query(Users.name.label("username"),Users.id).first()
#
# print(ret)  # ('alex', 3)
# print(ret.id,ret.username)  # 3 alex
​
# 表達式篩選條件
# user_list = db_session.query(Users).filter(Users.name=="ryxiong").all()
# user_list1 = db_session.query(Users).filter_by(name="ryxiong").all()
# for user in user_list:
#     print(user.name)
​
​
# 複雜查詢
# user_list2 = db_session.query(Users).filter(text("id<:value and name=:name")).params(value=3,name="ryxiong")
# print(user_list2)
​
​
# 查詢語句
# user_list3 = db_session.query(Users).filter(text("select * from user id<:value and name=:name")).params(value=3,name="ryxiong")
# print(user_list3)
​
​
# 其餘查詢條件
# ret = db_session.query(Users).filter(Users.id.between(1,3)).all()  # 查詢id值在1-3之間，不包含3的
# print(ret)
#
# ret1 = db_session.query(Users).filter(Users.id.in_([1,2])).all()  # 查詢id在列表[1,2]中的用戶
# print(ret1)
#
# ret2 = db_session.query(Users).filter(~Users.id.in_([1,2])).all()  # 查詢用戶id不在列表[1,2]中的。
# print(ret2)
​
# 子查詢
# ret3 = db_session.query(Users).filter(Users.id.in_(db_session.query(Users.id).filter_by(name="ryxiong"))).all()
# print(ret3)
​
# 通配符
# ret4 = db_session.query(Users).filter(Users.name.like("%ong")).all()
# print(ret4)
# ret5 = db_session.query(Users).filter(~Users.name.like("%ong")).all()
# print(ret5)
​
​
# 切片
# ret6 = db_session.query(Users)[1:2]
# print(ret6)
​
​
# 分組 group_by
from sqlalchemy.sql import func
# ret7 = db_session.query(Users).group_by(Users.name).all()
# print(ret7)
​
​
# 聚合函數
ret8 = db_session.query(
    func.max(Users.id),
    func.sum(Users.id),
    func.min(Users.id),
).group_by(Users.name).all()
print(ret8)  # [(3, Decimal('3'), 3), (2, Decimal('2'), 2), (1, Decimal('1'), 1)]
​
​
ret9 = db_session.query(
    func.max(Users.id),
    func.sum(Users.id),
    func.min(Users.id),
).group_by(Users.name).having(func.min(Users.id)>2).all()
print(ret9)  # [(3, Decimal('3'), 3)]