python 全栈开发，Day141(flask之应用上下文,SQLAlchemy)

悠悠 2022-02-02 08:15 219阅读 0赞

# [python 全栈开发，Day141(flask之应用上下文,SQLAlchemy)][python _Day141_flask_SQLAlchemy] #

# 一、flask之应用上下文 #

由于时间关系，详细过程略...

草稿图

![1341090-20181012203051132-754698641.png][]

参考链接：

[http://www.cnblogs.com/zhaopanpan/p/9457343.html][http_www.cnblogs.com_zhaopanpan_p_9457343.html]

总结：

![ContractedBlock.gif][] ![ExpandedBlockStart.gif][]

上下文管理（应用上下文）
    关键点
    请求上线文:
    ctx = RequestContext(request/session)
    应用上下文:
    app_ctx = AppContext(app/g)
    
    问：flask中g的作用？一次请求周期中保存一个公用值。
    
    g应用示例：
        from flask import Flask,request,session,g,current_app
    
        app = Flask(__name__)
    
        @app.before_request
        def b1():
            g.x1 = 123
    
    
        @app.route('/index')
        def index():
            print(g.x1)
    
        if __name__ == '__main__':
            app.run()

# 二、SQLAlchemy #

## 介绍 ##

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

pip3 install sqlalchemy

## 流程图 ##

![1341090-20181011200135472-1911203306.png][]

## 组成部分 ##

*  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

## 使用 ##

### 1. 执行原生SQL语句 ###

举例1

![ContractedBlock.gif][] ![ExpandedBlockStart.gif][]

import time
    import threading
    import sqlalchemy
    from sqlalchemy import create_engine
    from sqlalchemy.engine.base import Engine
     
    engine = create_engine(
        "mysql+pymysql://root:123@127.0.0.1:3306/t1?charset=utf8",
        max_overflow=0,  # 超过连接池大小外最多创建的连接
        pool_size=5,  # 连接池大小
        pool_timeout=30,  # 池中没有线程最多等待的时间，否则报错
        pool_recycle=-1  # 多久之后对线程池中的线程进行一次连接的回收（重置）
    )
     
     
    def task(arg):
        conn = engine.raw_connection()
        cursor = conn.cursor()
        cursor.execute(
            "select * from t1"
        )
        result = cursor.fetchall()
        cursor.close()
        conn.close()
     
     
    for i in range(20):
        t = threading.Thread(target=task, args=(i,))
        t.start()

举例2

![ContractedBlock.gif][] ![ExpandedBlockStart.gif][]

#!/usr/bin/env python
    # -*- coding:utf-8 -*-
    import time
    import threading
    import sqlalchemy
    from sqlalchemy import create_engine
    from sqlalchemy.engine.base import Engine
    
    engine = create_engine("mysql+pymysql://root:123@127.0.0.1:3306/t1", max_overflow=0, pool_size=5)
    
    
    def task(arg):
        conn = engine.contextual_connect()
        with conn:
            cur = conn.execute(
                "select * from t1"
            )
            result = cur.fetchall()
            print(result)
    
    
    for i in range(20):
        t = threading.Thread(target=task, args=(i,))
        t.start()

举例3

![ContractedBlock.gif][] ![ExpandedBlockStart.gif][]

#!/usr/bin/env python
    # -*- coding:utf-8 -*-
    import time
    import threading
    import sqlalchemy
    from sqlalchemy import create_engine
    from sqlalchemy.engine.base import Engine
    from sqlalchemy.engine.result import ResultProxy
    engine = create_engine("mysql+pymysql://root:123@127.0.0.1:3306/t1", max_overflow=0, pool_size=5)
    
    
    def task(arg):
        cur = engine.execute("select * from t1")
        result = cur.fetchall()
        cur.close()
        print(result)
    
    
    for i in range(20):
        t = threading.Thread(target=task, args=(i,))
        t.start()

注意: 查看连接 show status like 'Threads%';

### 2. ORM ###

### a. 创建数据库表 ###

#### 创建单表 ####

![ContractedBlock.gif][] ![ExpandedBlockStart.gif][]

#### 创建多个表并包含Fk、M2M关系 ####

![ContractedBlock.gif][] ![ExpandedBlockStart.gif][]

#!/usr/bin/env python
    # -*- coding:utf-8 -*-
    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()
    
    
    # ##################### 单表示例 #########################
    class Users(Base):
        __tablename__ = 'users'
    
        id = Column(Integer, primary_key=True)
        name = Column(String(32), index=True)
        age = Column(Integer, default=18)
        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', 'extra'),
        )
    
    
    class Hosts(Base):
        __tablename__ = 'hosts'
    
        id = Column(Integer, primary_key=True)
        name = Column(String(32), index=True)
        ctime = Column(DateTime, default=datetime.datetime.now)
    
    
    # ##################### 一对多示例 #########################
    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_id = Column(Integer, ForeignKey("hobby.id"))
    
        # 与生成表结构无关，仅用于查询方便
        hobby = relationship("Hobby", backref='pers')
    
    
    # ##################### 多对多示例 #########################
    
    class Server2Group(Base):
        __tablename__ = 'server2group'
        id = Column(Integer, primary_key=True, autoincrement=True)
        server_id = Column(Integer, ForeignKey('server.id'))
        group_id = Column(Integer, ForeignKey('group.id'))
    
    
    class Group(Base):
        __tablename__ = 'group'
        id = Column(Integer, primary_key=True)
        name = Column(String(64), unique=True, nullable=False)
    
        # 与生成表结构无关，仅用于查询方便
        servers = relationship('Server', secondary='server2group', backref='groups')
    
    
    class Server(Base):
        __tablename__ = 'server'
    
        id = Column(Integer, primary_key=True, autoincrement=True)
        hostname = Column(String(64), unique=True, nullable=False)
    
    
    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()
    
    创建多个表并包含Fk、M2M关系

*指定关联列：hobby = relationship("Hobby", backref='pers',foreign\_keys="Person.hobby\_id")*

### b. 操作数据库表 ###

![ContractedBlock.gif][] ![ExpandedBlockStart.gif][]

#!/usr/bin/env python
    # -*- coding:utf-8 -*-
    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()

#### 基于scoped\_session实现线程安全 ####

![ContractedBlock.gif][] ![ExpandedBlockStart.gif][]

#!/usr/bin/env python
    # -*- coding:utf-8 -*-
    from sqlalchemy.orm import sessionmaker
    from sqlalchemy import create_engine
    from sqlalchemy.orm import scoped_session
    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
    # 特殊的：scoped_session中有原来方法的Session中的一下方法：
    
    public_methods = (
        '__contains__', '__iter__', 'add', 'add_all', 'begin', 'begin_nested',
        'close', 'commit', 'connection', 'delete', 'execute', 'expire',
        'expire_all', 'expunge', 'expunge_all', 'flush', 'get_bind',
        'is_modified', 'bulk_save_objects', 'bulk_insert_mappings',
        'bulk_update_mappings',
        'merge', 'query', 'refresh', 'rollback',
        'scalar'
    )
    """
    session = scoped_session(Session)
    
    
    # ############# 执行ORM操作 #############
    obj1 = Users(name="alex1")
    session.add(obj1)
    
    
    
    # 提交事务
    session.commit()
    # 关闭session
    session.close()
    
    基于scoped_session实现线程安全

#### 多线程执行示例 ####

![ContractedBlock.gif][] ![ExpandedBlockStart.gif][]

#### 基本增删改查示例 ####

![ContractedBlock.gif][] ![ExpandedBlockStart.gif][]

#!/usr/bin/env python
    # -*- coding:utf-8 -*-
    import time
    import threading
    
    from sqlalchemy.ext.declarative import declarative_base
    from sqlalchemy import Column, Integer, String, ForeignKey, UniqueConstraint, Index
    from sqlalchemy.orm import sessionmaker, relationship
    from sqlalchemy import create_engine
    from sqlalchemy.sql import text
    
    from db import Users, Hosts
    
    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()
    
    # ################ 添加 ################
    """
    obj1 = Users(name="wupeiqi")
    session.add(obj1)
    
    session.add_all([
        Users(name="wupeiqi"),
        Users(name="alex"),
        Hosts(name="c1.com"),
    ])
    session.commit()
    """
    
    # ################ 删除 ################
    """
    session.query(Users).filter(Users.id > 2).delete()
    session.commit()
    """
    # ################ 修改 ################
    """
    session.query(Users).filter(Users.id > 0).update({"name" : "099"})
    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()
    """
    # ################ 查询 ################
    """
    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()
    """
    
    
    session.close()
    
    基本增删改查示例

#### 常用操作 ####

![ContractedBlock.gif][] ![ExpandedBlockStart.gif][]

#　条件
    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()
    ret = session.query(Users).filter(Users.id.in_([1,3,4])).all()
    ret = session.query(Users).filter(~Users.id.in_([1,3,4])).all()
    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()
    
    
    # 通配符
    ret = session.query(Users).filter(Users.name.like('e%')).all()
    ret = session.query(Users).filter(~Users.name.like('e%')).all()
    
    # 限制
    ret = session.query(Users)[1:2]
    
    # 排序
    ret = session.query(Users).order_by(Users.name.desc()).all()
    ret = session.query(Users).order_by(Users.name.desc(), Users.id.asc()).all()
    
    # 分组
    from sqlalchemy.sql import func
    
    ret = session.query(Users).group_by(Users.extra).all()
    ret = session.query(
        func.max(Users.id),
        func.sum(Users.id),
        func.min(Users.id)).group_by(Users.name).all()
    
    ret = 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()
    
    # 连表
    
    ret = session.query(Users, Favor).filter(Users.id == Favor.nid).all()
    
    ret = session.query(Person).join(Favor).all()
    
    ret = session.query(Person).join(Favor, isouter=True).all()
    
    
    # 组合
    q1 = session.query(Users.name).filter(Users.id > 2)
    q2 = session.query(Favor.caption).filter(Favor.nid < 2)
    ret = q1.union(q2).all()
    
    q1 = session.query(Users.name).filter(Users.id > 2)
    q2 = session.query(Favor.caption).filter(Favor.nid < 2)
    ret = q1.union_all(q2).all()
    
    常用操作

#### 原生SQL语句 ####

![ContractedBlock.gif][] ![ExpandedBlockStart.gif][]

#### 基于relationship操作ForeignKey ####

![ContractedBlock.gif][] ![ExpandedBlockStart.gif][]

#!/usr/bin/env python
    # -*- coding:utf-8 -*-
    import time
    import threading
    
    from sqlalchemy.ext.declarative import declarative_base
    from sqlalchemy import Column, Integer, String, ForeignKey, UniqueConstraint, Index
    from sqlalchemy.orm import sessionmaker, relationship
    from sqlalchemy import create_engine
    from sqlalchemy.sql import text
    from sqlalchemy.engine.result import ResultProxy
    from db import Users, Hosts, Hobby, Person
    
    engine = create_engine("mysql+pymysql://root:123@127.0.0.1:3306/s6?charset=utf8", max_overflow=0, pool_size=5)
    Session = sessionmaker(bind=engine)
    session = Session()
    # 添加
    """
    session.add_all([
        Hobby(caption='乒乓球'),
        Hobby(caption='羽毛球'),
        Person(name='张三', hobby_id=3),
        Person(name='李四', hobby_id=4),
    ])
    
    person = Person(name='张九', hobby=Hobby(caption='姑娘'))
    session.add(person)
    
    hb = Hobby(caption='人妖')
    hb.pers = [Person(name='文飞'), Person(name='博雅')]
    session.add(hb)
    
    session.commit()
    """
    
    # 使用relationship正向查询
    """
    v = session.query(Person).first()
    print(v.name)
    print(v.hobby.caption)
    """
    
    # 使用relationship反向查询
    """
    v = session.query(Hobby).first()
    print(v.caption)
    print(v.pers)
    """
    
    session.close()
    
    基于relationship操作ForeignKey

#### 基于relationship操作m2m ####

![ContractedBlock.gif][] ![ExpandedBlockStart.gif][]

#!/usr/bin/env python
    # -*- coding:utf-8 -*-
    import time
    import threading
    
    from sqlalchemy.ext.declarative import declarative_base
    from sqlalchemy import Column, Integer, String, ForeignKey, UniqueConstraint, Index
    from sqlalchemy.orm import sessionmaker, relationship
    from sqlalchemy import create_engine
    from sqlalchemy.sql import text
    from sqlalchemy.engine.result import ResultProxy
    from db import Users, Hosts, Hobby, Person, Group, Server, Server2Group
    
    engine = create_engine("mysql+pymysql://root:123@127.0.0.1:3306/s6?charset=utf8", max_overflow=0, pool_size=5)
    Session = sessionmaker(bind=engine)
    session = Session()
    # 添加
    """
    session.add_all([
        Server(hostname='c1.com'),
        Server(hostname='c2.com'),
        Group(name='A组'),
        Group(name='B组'),
    ])
    session.commit()
    
    s2g = Server2Group(server_id=1, group_id=1)
    session.add(s2g)
    session.commit()
    
    
    gp = Group(name='C组')
    gp.servers = [Server(hostname='c3.com'),Server(hostname='c4.com')]
    session.add(gp)
    session.commit()
    
    
    ser = Server(hostname='c6.com')
    ser.groups = [Group(name='F组'),Group(name='G组')]
    session.add(ser)
    session.commit()
    """
    
    
    # 使用relationship正向查询
    """
    v = session.query(Group).first()
    print(v.name)
    print(v.servers)
    """
    
    # 使用relationship反向查询
    """
    v = session.query(Server).first()
    print(v.hostname)
    print(v.groups)
    """
    
    
    session.close()
    
    基于relationship操作m2m

#### 其他 ####

![ContractedBlock.gif][] ![ExpandedBlockStart.gif][]

由于时间关系，详细过程略...

本文参考链接：

[https://www.cnblogs.com/wupeiqi/articles/8259356.html][https_www.cnblogs.com_wupeiqi_articles_8259356.html]

**未完待续...**

posted @ 2018-10-11 20:19 [肖祥][Link 1] 阅读( ...) 评论( ...) [编辑][Link 2] 收藏

[python _Day141_flask_SQLAlchemy]: https://www.cnblogs.com/xiao987334176/p/9774802.html
[1341090-20181012203051132-754698641.png]: /images/20220202/f09c25934c374a0f96a24f3732a0ff2f.png
[http_www.cnblogs.com_zhaopanpan_p_9457343.html]: http://www.cnblogs.com/zhaopanpan/p/9457343.html
[ContractedBlock.gif]: https://images.cnblogs.com/OutliningIndicators/ContractedBlock.gif
[ExpandedBlockStart.gif]: /images/20220202/928b86bf31514a91948ca69de8e83437.png
[1341090-20181011200135472-1911203306.png]: /images/20220202/7a07ca559a3b4bf584f6e2c3779c5879.png
[https_www.cnblogs.com_wupeiqi_articles_8259356.html]: https://www.cnblogs.com/wupeiqi/articles/8259356.html
[Link 1]: https://www.cnblogs.com/xiao987334176/
[Link 2]: https://i.cnblogs.com/EditPosts.aspx?postid=9774802