flask中orm增刪改查操作

一、建立表

# model.py
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

Base = declarative_base()


# make_declarative_base

class Users(Base):
    __tablename__ = 'users'  # 資料庫表名稱
    id = Column(Integer, primary_key=True)  # id 主鍵
    name = Column(String(32), index=True, nullable=False)  # name列，索引，不可為空
    age = Column(Integer)
    email = Column(String(32), unique=True)
    # datetime.datetime.now不能加括號，加了括號，以後永遠是當前時間
    ctime = Column(DateTime, default=datetime.datetime.now)
    extra = Column(Text, nullable=True)

    def __repr__(self):
        return self.name


    __table_args__ = (
        UniqueConstraint('id', 'name', name='uix_id_name'), #聯合唯一
        Index('ix_id_name', 'name', 'email'), #索引
    )



class Boy(Base):
    __tablename__ = 'boy'
    id = Column(Integer, primary_key=True, autoincrement=True)
    name = Column(String(64), unique=True, nullable=False)

    # 與生成表結構無關，僅用於查詢方便,放在哪個單表中都可以
    girl = relationship('Girl', secondary='boy2girl', backref='boys')


def init_db():
    """
    根據類建立資料庫表
    :return:
    """
    engine = create_engine(
        "mysql+pymysql://root:[email protected]:3306/db_flask?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:[email protected]:3306/db_flask?charset=utf8",
        max_overflow=0,  # 超過連線池大小外最多建立的連線
        pool_size=5,  # 連線池大小
        pool_timeout=30,  # 池中沒有執行緒最多等待的時間，否則報錯
        pool_recycle=-1  # 多久之後對執行緒池中的執行緒進行一次連線的回收（重置）
    )

    Base.metadata.drop_all(engine)


if __name__ == '__main__':
    # 建立表
    init_db()

    # 刪除表
    # drop_db()
 

二、增加資料

from sqlalchemy.orm import sessionmaker
from sqlalchemy import create_engine
from model import Users

engine = create_engine("mysql+pymysql://root:[email protected]:3306/db_flask", max_overflow=0, pool_size=5)
Session = sessionmaker(bind=engine)

session = Session()
# ################ 新增 ################

# 1. 新增單條記錄
obj1 = Users(name="randy", age=10)
session.add(obj1) # 返回及結果為none

# 2. 新增多條記錄
user_obj1 = Users(name='laowang', age=14)
user_obj2 = Users(name='laozhang', age=12)
session.add_all([user_obj1, user_obj2])

# 提交
session.commit()
session.close()
 

三、刪除記錄

from sqlalchemy.orm import sessionmaker
from sqlalchemy import create_engine
from model import Users

engine = create_engine("mysql+pymysql://root:[email protected]:3306/db_flask", max_overflow=0, pool_size=5)
Session = sessionmaker(bind=engine)

session = Session()
################ 刪除 ################

delete_num = session.query(Users).filter(Users.id == 2).delete()
# 刪除的條數
print(delete_num)
session.close()
 

四、修改資料

from sqlalchemy.orm import sessionmaker
from sqlalchemy import create_engine
from model import Users

engine = create_engine("mysql+pymysql://root:[email protected]:3306/db_flask", max_overflow=0, pool_size=5)
Session = sessionmaker(bind=engine)

session = Session()
################ 修改 ################
# 方式一，以字典的方式修改
res = session.query(Users).filter(Users.id==1).update({'name': 'randysun', 'age': 18})
# 修改記錄條數
print(res)

################ 修改 ################


# 方式一，以字典的方式修改
# res = session.query(Users).filter(Users.id==1).update({'name': 'randysun', 'age': 18})
# print(res)

# 方式二， 類似於django的F查詢

# 拼接字串需要將synchronize_session設定為 False
res = session.query(Users).filter(Users.id == 3).update({Users.name: Users.name + 'sun'}, synchronize_session=False)
# 修改記錄條數
print(res)

# 數字增加需要將synchronize_session設定為 'evaluate'
res1 = session.query(Users).filter(Users.id == 3).update({Users.age: Users.age + 6}, synchronize_session='evaluate')
# 修改記錄條數
print(res1)

session.commit()

session.close()

五、查詢

from sqlalchemy.orm import sessionmaker
from sqlalchemy import create_engine
from model import Users

engine = create_engine("mysql+pymysql://root:[email protected]:3306/db_flask", max_overflow=0, pool_size=5)
Session = sessionmaker(bind=engine)

session = Session()

################ 查詢 ################
# 方式一
res = session.query(Users).all()
res1 = session.query(Users)

# 物件列表
print(res)
print(res[0].name, res[0].age)
# 返回sql查詢語句, 如果沒有使用 .all或first就是sql語句
print(res1)
print(res1.all()[0].name)

# 方式二，取別名，必須用別名獲取資料

res = session.query(Users.name.label('user_name'), Users.age.label('user_age')).all()
res1 = session.query(Users.name.label('user_name'), Users.age.label('user_age'))

print(res)
print(res[0].user_name, res[0].user_age)

# print(res[0].user, res[0].age) 報錯

# SELECT users.name AS user_name, users.age AS user_age FROM users
print(res1)
print(res1.all()[0].user_name, res1.all()[0].user_age)


# 方式三

# filter查詢傳入的是表示式
res = session.query(Users).filter(Users.name == 'randy').first()
res1 = session.query(Users).filter(Users.name == 'randy')
print(res)
# sql語句 兩者查詢sql語句一樣
print(res1)
print(res.name)

# filter_by 傳入的是引數
res_filter_by = session.query(Users).filter_by(name='randysun').first()
res_filter_by1 = session.query(Users).filter_by(name='randysun')
print(res_filter_by)
# sql語句 兩者查詢sql語句一樣
print(res_filter_by1)

print(res_filter_by.name)
session.close()

六、總結

• query: 引數為類名則查詢出來的資料是一個一個物件，如果引數為具體表欄位，則查詢出來的結果為具體的資料，query引數可以為其查詢設定對應查詢欄位的別名，
• add: 增加單條記錄，引數為單條物件
• add_all:增加多條資料，引數為物件列表
• delete: 與查詢條件filter一起連用刪除具體的一條記錄
• update: 修改資料可以採用字典的形式，key為修改的欄位，value要修改的值,類似django 中的F查詢,但是字串相加需要將synchronize_session 設定為True,數字相加需要將 synchronize_session設定為 ’evaluate‘
• all:查詢所有資料
• filter：查詢單條資料，引數為具體的條件表示式
• filter_by: 查詢單條資料，傳入引數為具體為引數

新增，刪除，修改資料都需要commit提交才能對資料操作成功