SQLAlchemy

安裝介紹

- SQLAlchemy 是Python的一款Orm框架

建立

 

 

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

#建立基類,至關於Django中的 models.Model,被各個數據表類所繼承
Base = declarative_base()


# ##################### 單表示例 #########################
# 建立一張數據表
class Users(Base):
    __tablename__ = 'users'

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

    __table_args__ = (
        # UniqueConstraint('id', 'name', name='uix_id_name'),
        # Index('ix_id_name', 'name', 'extra'),
    )

# 建立另外一張數據表
class School(Base):
    __tablename__ = "school"

    id = Column(Integer,primary_key=True,autoincrement=True)
    name = Column(String,unique=True)


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


# 經過Base.metadata找到全部繼承 Base 的數據表class
Base.metadata.create_all(engine)

# SQLAlchemy數據表進行修改後,沒法直接進行更新,只能刪除表後進行操做,從新進行操做

 

 

增刪改查相關操做

 

 

from CreateDB import Users, School

# 1. 建立一個用戶添加到數據庫
# 建立鏈接
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
engine = create_engine("mysql+pymysql://root:DragonFire@127.0.0.1:3306/dragon?charset=utf8",
                       )

# 建立數據表操做對象 sessionmaker
DB_session = sessionmaker(engine)
db_session = DB_session()

# 1.增長 操做數據表
# 單行操做
# obj1 = Users(name="123") # 經過 Users數據表類建立一條數據
# db_session.add(obj1) # 將建立好的數據添加到 數據表操做對象的內存中,此時數據庫還並不知道要添加數據
# db_session.commit() # 將數據表操做對象內存中的全部數據及操做提交到數據庫中
# 多行操做
# db_session.add_all([
#     Users(name="zhangsan"),
#     Users(name="lisi"),
# ])
# db_session.commit()

# 2.查詢 數據表操做
# user_list = db_session.query(Users).all() # 查詢全部數據
# user_list = db_session.query(Users).filter(Users.id >=2 ) # 查詢帶有條件的數據 表達式 返回sql語句,循環依然能夠獲取到數據
# user_list = db_session.query(Users).filter(Users.id >=2 ).all() # 查詢帶有條件的數據 表達式 返回數據列表
# print(user_list)
# for row in user_list:
#     print(row.id,row.name)

# 3.刪除數據 數據表操做
# db_session.query(Users).filter(Users.id == 1).delete() # 刪除帶有條件的數據
# db_session.commit()

# 4.修改
# db_session.query(Users).filter(Users.id == 3).update({"name":"alexDSB"}) # 更新id=3的數據
# db_session.commit()

# 關閉鏈接
db_session.close()

 

 

高級版查詢操做

 

 

from CreateDB import Users, School

# 1. 建立一個用戶添加到數據庫
# 建立鏈接
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
engine = create_engine("mysql+pymysql://root:DragonFire@127.0.0.1:3306/dragon?charset=utf8",
                       )

# 建立數據表操做對象 sessionmaker
DB_session = sessionmaker(engine)
db_session = DB_session()

# 查詢數據表操做
"""
r1 = session.query(Users).all()
r2 = session.query(Users.name.label('xx'), Users.age).all()
r3 = session.query(Users).filter(Users.name == "alex").all()
r4 = session.query(Users).filter_by(name='alex').all()
r5 = session.query(Users).filter_by(name='alex').first()
r6 = session.query(Users).filter(text("id<:value and name=:name")).params(value=224, name='fred').order_by(Users.id).all()
r7 = session.query(Users).from_statement(text("SELECT * FROM users where name=:name")).params(name='ed').all()
"""
# 篩選查詢列
# user_list = db_session.query(Users.name).all()
# print(user_list) # [('alexDSB',), ('zhangsan',)] 雖然看到的是元祖,可是依然能夠經過對象打點兒調用屬性
# for row in user_list:
#     print(row.name)

# 別名映射  name as nick
# user_list = db_session.query(Users.name.label("nick")).all()
# print(user_list) # [('alexDSB',), ('zhangsan',)] 雖然看到的是元祖,可是依然能夠經過對象打點兒調用屬性
# for row in user_list:
#     print(row.nick) # 這裏要寫別名了

# 篩選條件格式
# user_list = db_session.query(Users).filter(Users.name == "alexDSB").all()
# user_list = db_session.query(Users).filter(Users.name == "alexDSB").first()
# user_list = db_session.query(Users).filter_by(name="alexDSB").first()
# for row in user_list:
#     print(row.nick)

# 複雜查詢
# from sqlalchemy.sql import text
# user_list = db_session.query(Users).filter(text("id<:value and name=:name")).params(value=3,name="alexDSB")

# 查詢語句
# from sqlalchemy.sql import text
# user_list = db_session.query(Users).filter(text("select * from users id<:value and name=:name")).params(value=3,name="alexDSB")

# 排序 :
# user_list = db_session.query(Users).order_by(Users.id).all()
# user_list = db_session.query(Users).order_by(Users.id.desc()).all()
# for row in user_list:
#     print(row.name,row.id)

#其餘查詢條件
"""
ret = session.query(Users).filter_by(name='alex').all()
ret = session.query(Users).filter(Users.id > 1, Users.name == 'eric').all()
ret = session.query(Users).filter(Users.id.between(1, 3), Users.name == 'eric').all() # between 大於1小於3的
ret = session.query(Users).filter(Users.id.in_([1,3,4])).all() # in_([1,3,4]) 只查詢id等於1,3,4的
ret = session.query(Users).filter(~Users.id.in_([1,3,4])).all() # ~xxxx.in_([1,3,4]) 查詢不等於1,3,4的
ret = session.query(Users).filter(Users.id.in_(session.query(Users.id).filter_by(name='eric'))).all() 子查詢
from sqlalchemy import and_, or_
ret = session.query(Users).filter(and_(Users.id > 3, Users.name == 'eric')).all()
ret = session.query(Users).filter(or_(Users.id < 2, Users.name == 'eric')).all()
ret = session.query(Users).filter(
    or_(
        Users.id < 2,
        and_(Users.name == 'eric', Users.id > 3),
        Users.extra != ""
    )).all()
# select * from users where id<2 or (name="eric" and id>3) or extra != "" 
"""

# 關閉鏈接
db_session.close()

 

 

高級版更新操做

 

 

"""
db_session.query(Users).filter(Users.id > 0).update({"name" : "099"})
db_session.query(Users).filter(Users.id > 0).update({Users.name: Users.name + "099"}, synchronize_session=False)
db_session.query(Users).filter(Users.id > 0).update({"age": Users.age + 1}, synchronize_session="evaluate")
db_session.commit()
"""

 

 

擴展內容

- 一對多：

- 建立表：

 

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

#建立基類,至關於Django中的 models.Model,被各個數據表類所繼承
Base = declarative_base()


# ##################### 多表示例 #########################
# 建立一張數據表
class Author(Base):
    __tablename__ = 'author'

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


#  建立另外一張數據表
class Book(Base):
    __tablename__ = 'book'

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

    author_id = Column(Integer,ForeignKey("author.id"))

    # relationshi 不會單獨生成字段 ,只用於增長查詢操做
    user = relationship("Author",backref="author2book") # backref 反向查找的名字




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

# 經過Base.metadata找到全部繼承 Base 的數據表class
Base.metadata.create_all(engine)

 

 

 

- 增刪改查：

 

from SQLAlchemy_ready.ss2_ForeignKey_relationship import Author, Book

# 1. 建立一個用戶添加到數據庫
# 建立鏈接
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
engine = create_engine("mysql+pymysql://root:DragonFire@127.0.0.1:3306/dragon?charset=utf8",
                       )

# 建立數據表操做對象 sessionmaker
DB_session = sessionmaker(engine)
db_session = DB_session()

# 1.增長 操做數據表
# 添加兩個做者:
# db_session.add_all([
#     Author(name="alex"),
#     Author(name="yinwangba")
# ])
# db_session.commit()
# 添加一本書 jinpingmei 做者是 yinwangba
# author = db_session.query(Author).filter(Author.name == "yinwangba").first()
# db_session.add(Book(name="jinpingmei",author_id=author.id))
# db_session.commit()

# 2.查詢全部數據,並顯示做者名稱,連表查詢
# book_list = db_session.query(Book).all()
# for row in book_list:
#     print(row.name,row.author_id)

# book_list = db_session.query(Book.name.label("bname"),Author.name.label ("aname")).join(Author,Book.author_id == Author.id,isouter=True).all()
# print(book_list)
# for row in book_list:
#     print(row.aname,row.bname)

# 查詢之relationship 快速連表
# 建立表的時候加入 relationship
#普通版添加
# obj = Author(name="yinwangba")
# db_session.add(obj)
# db_session.commit()
# print(obj.id,obj.name)
#
# obj_book = Book(name="jinpingmei",author_id=obj.id)
# db_session.add(obj_book)
# db_session.commit()
# obj = Author(name="yinwangba")

# relationship版 添加
# bobj = Book(name="jinpingmei",user=Author(name="yinwangba"))
# db_session.add(bobj)
# db_session.commit()

# 查詢之relationship 快速連表
# book_list = db_session.query(Book).all()
# for row in book_list:
#     print(row.name,row.user.name)

# 查詢做者的全部書籍
# obj = db_session.query(Author).filter(Author.name=="yinwangba").first()
# print(obj.author2book)

# 反向字段添加
# author_obj = Author(name="alex")
# author_obj.author2book = [Book(name="儒林外史之銀林大戰"),Book(name="鄰家小妹妹")]
# db_session.add(author_obj)
# db_session.commit()

# 關閉鏈接
db_session.close()

 

 

 

- 多對多 ManyToMany：

　　- 建立表：

 

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
from sqlalchemy.orm import relationship

#建立基類,至關於Django中的 models.Model,被各個數據表類所繼承
Base = declarative_base()

# 建立第三張表 Boys and Girls
class Hotel(Base):
    __tablename__ = "hotel"

    id = Column(Integer,primary_key=True)
    boy_id = Column(Integer,ForeignKey("boys.id"))
    girl_id = Column(Integer,ForeignKey("girls.id"))

# 建立一張數據表
class Boys(Base):
    __tablename__ = 'boys'

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

    # 建立於酒店的關係
    girls = relationship("Girls",secondary="hotel",backref="boys")

# 建立另外一張數據表
class Girls(Base):
    __tablename__ = "girls"

    id = Column(Integer,primary_key=True,autoincrement=True)
    name = Column(String(32),index  =True)

    # 建立於酒店的關係
    # boy = relationship("Boys",secondary="hotel",backref="toHotel")









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

# 經過Base.metadata找到全部繼承 Base 的數據表class
Base.metadata.create_all(engine)


# SQLAlchemy數據表進行修改後,沒法直接進行更新,只能刪除表後進行操做,從新進行操做,
# 但Flask-SQLAlchemy + Flask-migrate + Flask-script 就能夠實現 Django 的數據遷移 MakeMigration migrate

#

多對多關係創建

 

 

- 操做：

 

from SQLAlchemy_ready.ss4_M2M import Girls, Boys,Hotel

# 1. 建立一個用戶添加到數據庫
# 建立鏈接
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
engine = create_engine("mysql+pymysql://root:DragonFire@127.0.0.1:3306/dragon?charset=utf8")


# 建立數據表操做對象 sessionmaker
DB_session = sessionmaker(engine)
db_session = DB_session()

# 1.增長 操做數據表
# boy = Boys(name="jinwangba")
# boy.girls = [Girls(name="fengjie"),Girls(name="juaner")]
# db_session.add(boy)
# db_session.commit()

# 2.查詢
# 使用relationship正向查詢
# boy = db_session.query(Boys).first()
# print(boy.name,boy.girls[0].name)

# 使用relationship反向查詢
# girls = db_session.query(Girls).first()
# print(girls.boys[0].name , girls.name)


# 關閉鏈接
db_session.close()