【Flask】SQLAlchemy

文章目录

• • SQLAlchemy是什么
  • • 为什么使用orm
    • 定义
    • 安装
    • 组成部分
    • SQLAlchemy本身无法操作数据库，其必须以来pymsql等第三方插件
    • SQLAlchemy的使用
    • • 原生sql
      • 使用orm映射数据表
      • 外键关系
      • • 一对多(ForeignKey)
        • 多对多
        • 使用orm操作记录
        • • 简单表操作
          • 基于scoped_session实现线程安全
          • CRUD
          • • 基础操作
            • 进阶操作
            • Flask集成sqlalchemy
            • • 构建模型类
              • 常用的字段类型
              • 常用的字段选项
              • 增
              • 查
              • 改
              • • 先查询, 再更新
                • 基于过滤条件的更新 (推荐方案)
                • 删
                • • 先查询, 再删除
                  • 基于过滤条件的删除 (推荐方案)

                    SQLAlchemy是什么

                    为什么使用orm

                    优点

                    • 有语法提示, 省去自己拼写SQL，保证SQL语法的正确性
                    • orm提供方言功能(dialect, 可以转换为多种数据库的语法), 减少学习成本
                    • 防止sql注入攻击
                    • 搭配数据迁移, 更新数据库方便
                    • 面向对象, 可读性强, 开发效率高

                      缺点

                      • 需要语法转换, 效率比原生sql低
                      • 复杂的查询往往语法比较复杂 (可以使用原生sql替换)

                        定义

                        SQLAlchemy是一个基于Python实现的ORM框架。该框架建立在 DB API之上，使用关系对象映射进行数据库操作，简言之便是：将类和对象转换成SQL，然后使用数据API执行SQL并获取执行结果。

                        安装

                        pip3 install sqlalchemy

                        组成部分

                        • Engine，框架的引擎
                        • Connection Pooling ，数据库连接池
                        • Dialect，选择连接数据库的DB API种类
                        • Schema/Types，架构和类型
                        • SQL Exprression Language，SQL表达式语言

                          SQLAlchemy本身无法操作数据库，其必须以来pymsql等第三方插件

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

                          SQLAlchemy的使用

                          原生sql

                          from sqlalchemy import create_enginefrom sqlalchemy.engine.base import Enginefrom urllib import parseimport threadinguser = "root"password = "xxx@000"pwd = parse.quote_plus(password)   # 解决密码中含@符导致报错host = "127.0.0.1:"# 第一步: 创建engineengine = create_engine(    f"mysql+pymysql://{user}:{pwd}@{host}3306/test1?charset=utf8",    max_overflow=0,  # 超过连接池大小外最多创建的连接    pool_size=5,  # 连接池大小    pool_timeout=30,  # 池中没有线程最多等待的时间，否则报错    pool_recycle=-1  # 多久之后对线程池中的线程进行一次连接的回收（重置）)# 第二步：使用def task():    conn = engine.raw_connection()  # 从连接池中取一个连接    cursor = conn.cursor()    sql = "select * from signer"    cursor.execute(sql)    print(cursor.fetchall())if __name__ == '__main__':    for i in range(20):        t = threading.Thread(target=task)        t.start()

                          使用orm映射数据表

                          import datetimefrom sqlalchemy.ext.declarative import declarative_basefrom model import engine  # 用的简单使用里面的enginefrom sqlalchemy import Column, Integer, String, Text, ForeignKey, DateTime, UniqueConstraint, IndexBase = declarative_base()   # 基类# 表模型class Users(Base):    __tablename__ = 'users'  # 数据库表名称， 必须写        id = Column(Integer, primary_key=True)  # id 主键    name = Column(String(32), index=True, nullable=False)  # name列，索引，不可为空    email = Column(String(32), unique=True)    # datetime.datetime.now不能加括号，加了括号，以后永远是当前时间    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'),  # 索引    )def init_db():    """    根据类创建继承base类的表    :return:    """    Base.metadata.create_all(engine)def drop_db():    """    根据类删除继承base类的表    :return:    """    Base.metadata.drop_all(engine)if __name__ == '__main__':# sqlalchemy 只支持 创建和删除表，不支持修改表(django orm支持)。sqlalchemy 需要借助第三方实现    init_db()  # 创建表    # drop_db()   # 删除表

                          外键关系

                          一对多(ForeignKey)

                          class Hobby(Base):    __tablename__ = 'hobby'    id = Column(Integer, primary_key=True)    caption = Column(String(50), default='篮球')class Person(Base):    __tablename__ = 'person'    nid = Column(Integer, primary_key=True)    name = Column(String(32), index=True, nullable=True)    # hobby指的是tablename而不是类名    hobby_id = Column(Integer, ForeignKey("hobby.id"))        # 跟数据库无关，不会新增字段，只用于快速链表操作    # 类名，backref用于反向查询  参数 uselist=False , 设置就变成了一对一，其他和一对多一样    hobby=relationship('Hobby',backref='pers')    # hobby=relationship('Hobby',backref='pers', uselist=False)  # 一对一关系

                          多对多

                          class Boy2Girl(Base):    __tablename__ = 'boy2girl'    id = Column(Integer, primary_key=True, autoincrement=True)    girl_id = Column(Integer, ForeignKey('girl.id'))    boy_id = Column(Integer, ForeignKey('boy.id'))class Girl(Base):    __tablename__ = 'girl'    id = Column(Integer, primary_key=True)    name = Column(String(64), unique=True, nullable=False)class Boy(Base):    __tablename__ = 'boy'    id = Column(Integer, primary_key=True, autoincrement=True)    hostname = Column(String(64), unique=True, nullable=False)        # 与生成表结构无关，仅用于查询方便,放在哪个单表中都可以    servers = relationship('Girl', secondary='boy2girl', backref='boys')    '''girl_id = Column(Integer, ForeignKey("hobby.id", ondelete='SET NULL'))  # 一般用SET NULL外键约束    1. RESTRICT：若子表中有父表对应的关联数据，删除父表对应数据，会阻止删除。默认项    2. NO ACTION：在MySQL中，同RESTRICT。    3. CASCADE：级联删除。    4. SET NULL：父表对应数据被删除，子表对应数据项会设置为NULL。'''

                          扩充：在 django 中，外键管理有个参数 db_contraint=False 用来在逻辑上关联表，但实体不建立约束。同样在SQLAlchemy 中也可以通过配值 relationship 参数来实现同样的效果

                          class Boy2Girl(Base):    __tablename__ = 'boy2girl'    id = Column(Integer, primary_key=True, autoincrement=True)    girl_id = Column(Integer)  # 不用ForeignKey    boy_id = Column(Integer)        gitl = db.relationship(        "Girl",        # uselist=False,  # 一对一设置        backref=backref("to_course", uselist=False),  # backref用于反向查询 uselist 作用同上        lazy="subquery",    # 懒加载 用来指定sqlalchemy 什么时候加载数据        primaryjoin="Girl.id==Boy2Girl.girl_id",  # 指定对应关系        foreign_keys="Boy2Girl.girl_id"  # 指定表的外键字段    )'''lazy 可选值    select：就是访问到属性的时候，就会全部加载该属性的数据  默认值    joined：对关联的两个表使用联接    subquery：与joined类似，但使用子子查询    dynamic：不加载记录，但提供加载记录的查询，也就是生成query对象'''

                          使用orm操作记录

                          简单表操作

                          from sqlalchemy.orm import sessionmakerfrom model import enginefrom db_model import Users# 定义一个 session， 以后操作数据都用 session 来执行Session = sessionmaker(bind=engine)session = Session()# 创建User对象usr = Users(name="yxh", email="152@11.com", extra="xxx")# 通过 user对象 添加到session中 session.add(usr)# 提交，才会刷新到数据库中，不提交不会执行session.commit()

                          基于scoped_session实现线程安全

                          session 如果是一个全局对象。那么在多线程的情况下，并发使用同一个变量 session 是不安全的，解决方案如下：

                          • 将session定义在局部，每一个view函数都定义一个session。 代码冗余，不推荐

                          • 基于scoped_session 实现线程安全。原理同 request对象，g对象一致。也是基于local，给每一个线程创造一个session

                            from sqlalchemy.orm import sessionmaker, scoped_sessionfrom model import enginefrom db_model import Users# 定义一个 sessionSession = sessionmaker(bind=engine)# session = Session()session = scoped_session(Session)  # 后续使用这个session就是线程安全的# 创建User对象usr = Users(name="yxh", email="152@11.com", extra="xxx")# 通过 user对象 添加到session中session.add(usr)# 提交，才会刷新到数据库中，不提交不会执行session.commit()

                            CRUD

                            创建（Create）、读取（Read）、更新（Update）和删除（Delete）

                            基础操作

                            from sqlalchemy.orm import sessionmaker, scoped_sessionfrom sqlalchemy.sql import textfrom model import enginefrom db_model import Users# 定义一个 sessionSession = sessionmaker(bind=engine)# session = Session()session = scoped_session(Session)  # 后续使用这个session就是线程安全的# 1  增加操作obj1 = Users(name="yxh003")session.add(obj1)# 增加多个，不同对象session.add_all([    Users(name="yxh009"),    Users(name="yxh008"),])session.commit()# 2 删除操作---》查出来再删---》session.query(Users).filter(Users.id > 2).delete()session.commit()# 3 修改操作--》查出来改  传字典session.query(Users).filter(Users.id > 0).update({"name": "yxh"})# 类似于django的F查询# 字符串加session.query(Users).filter(Users.id > 0).update({Users.name: Users.name + "099"}, synchronize_session=False)# 数字加session.query(Users).filter(Users.id > 0).update({"age": Users.age + 1}, synchronize_session="evaluate")session.commit()# 4 查询操作----》r1 = session.query(Users).all()  # 查询所有# 只取age列，把name重命名为xx# select name as xx,age from user;r2 = session.query(Users.name.label('xx'), Users.email).all()# filter传的是表达式，filter_by传的是参数r3 = session.query(Users).filter(Users.name == "yxh").all()r3 = session.query(Users).filter(Users.id >= 1).all()r4 = session.query(Users).filter_by(name='yxh').all()r5 = session.query(Users).filter_by(name='yxh').first()# :value 和:name 相当于占位符，用params传参数r6 = session.query(Users).filter(text("id