sqlalchmy和Flask-SQLAlchemy ，flask-flask-migrate 项目依赖包

sqlalchmy

链接：https://www.cnblogs.com/xiaoyuanqujing/articles/11715497.html#_label0

一.介绍

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

二.多种使用方式(能创建表，删除表，不能修改表)

前提安装

pip3 install sqlalchemy

1.第一种使用方式：执行原生sql（不常用，类似于pymysql）

import time
import threading
import sqlalchemy
from sqlalchemy import create_engine
from sqlalchemy.engine.base import Engine

#引擎数据库连接池
engine = create_engine(
    # 数据库+连接方式：//用户名：密码@ip:端口/数据库?编码
    "mysql+pymysql://root:123@127.0.0.1:3306/tset?charset=utf8",
    max_overflow=0,  # 超过连接池大小外最多创建的连接
    pool_size=5,  # 连接池大小
    pool_timeout=30,  # 池中没有线程最多等待的时间，否则报错
    pool_recycle=-1  # 多久之后对线程池中的线程进行一次连接的回收（重置）
)

# 任务函数
def task(arg):
    # 获取连接池
    conn = engine.raw_connection()
    # 产生游标
    cursor = conn.cursor()
    # 执行原生sql语句
    cursor.execute(
        "select * from user"
    )
    # 获取结果
    result = cursor.fetchall()
    print(result) # (1, '张三', 18), (2, '李四', 20))
    # 关闭游标和连接
    cursor.close()
    conn.close()

for i in range(20):
    t = threading.Thread(target=task, args=(i,))
    t.start()

2.第二种orm使用方式

#1.models.py创建表和删除表操作

#1.导入模块
import datetime
#引擎
from sqlalchemy import create_engine
#基类
from sqlalchemy.ext.declarative import declarative_base

from sqlalchemy.orm import relationship

# 字段类型
from sqlalchemy import Column, Integer, String, Text, ForeignKey, DateTime, UniqueConstraint, Index

#2.创建基类
Base = declarative_base()

#3.定义表格
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'), #索引
    )

# 4.创建表
def init_db():
    # 创建引擎建立数据库连接
    engine = create_engine(
        "mysql+pymysql://root:123@127.0.0.1:3306/tset?charset=utf8",
        max_overflow=0,  # 超过连接池大小外最多创建的连接
        pool_size=5,  # 连接池大小
        pool_timeout=30,  # 池中没有线程最多等待的时间，否则报错
        pool_recycle=-1  # 多久之后对线程池中的线程进行一次连接的回收（重置）
    )
    #会将当前执行文件中所有继承自Base类的类, 生成表
    Base.metadata.create_all(engine)


#4.删除表格
def drop_db():
    # 创建引擎建立数据库连接
    engine = create_engine(
        "mysql+pymysql://root:123@127.0.0.1:3306/tset?charset=utf8",
        max_overflow=0,  # 超过连接池大小外最多创建的连接
        pool_size=5,  # 连接池大小
        pool_timeout=30,  # 池中没有线程最多等待的时间，否则报错
        pool_recycle=-1  # 多久之后对线程池中的线程进行一次连接的回收（重置）
    )
    # 会将当前执行文件中所有继承自Base类的类, 删除表
    Base.metadata.drop_all(engine)

# 执行
if __name__ == '__main__':
    #删除表格
    drop_db()
    #创建表格
    # init_db()

#2.app.py增删改查

#1.导入模块
from sqlalchemy.orm import sessionmaker
from sqlalchemy import create_engine
# 导入models.py文件中类
from models.py import Users

#2.建立连接
engine = create_engine(
        "mysql+pymysql://root:123@127.0.0.1:3306/tset?charset=utf8",
        max_overflow=0,  # 超过连接池大小外最多创建的连接
        pool_size=5,  # 连接池大小
        pool_timeout=30,  # 池中没有线程最多等待的时间，否则报错
        pool_recycle=-1  # 多久之后对线程池中的线程进行一次连接的回收（重置）
    )
# 3.
Session = sessionmaker(bind=engine)
# 每次执行数据库操作时，都需要创建一个Connection
session = Session()

# 4.增删改查############# 执行ORM操作 #############
obj1 = Users(name="owen")
session.add(obj1)

# 5.提交事务
session.commit()

# 关闭session，其实是将连接放回连接池
session.close()

#3.表格类型补充和对应的增删改查对应上述2个文件位置即可

1.一对多关系

#表格
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而不是类名，uselist=False
    hobby_id = Column(Integer, ForeignKey("hobby.id"))
    
    # 跟数据库无关，不会新增字段，只用于快速链表操作
    # 类名，backref用于反向查询
    hobby=relationship('Hobby',backref='pers')

#增删改查
#增
#1多增
session.add_all([
    Hobby(caption='乒乓球'),
    Hobby(caption='羽毛球'),
    Person(name='张三', hobby_id=1),
    Person(name='李四', hobby_id=2),
])
#2.单增和正向增（hobby为hobby=relationship('Hobby',backref='pers')中的，同时会为2张表增加数据并绑定关系）
person = Person(name='张九', hobby=Hobby(caption='姑娘'))
session.add(person)
#3.反向增
hb = Hobby(caption='人妖')
# pers为hobby=relationship('Hobby',backref='pers')中的，同时会为2张表增加数据并绑定关系）
hb.pers = [Person(name='文飞'), Person(name='博雅')]
session.add(hb)

#查
# 1.使用relationship正向查询
"""
v = session.query(Person).first()
print(v.name)
print(v.hobby.caption)
"""

# 2.使用relationship反向查询
"""
v = session.query(Hobby).first()
print(v.caption)
print(v.pers)

2.多对多

#表
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)
   boyname = Column(String(64), unique=True, nullable=False)
    
    # 与生成表结构无关，仅用于查询方便,放在哪个单表中都可以
    girls = relationship('Girl', secondary='boy2girl', backref='boys')

session.add_all([
    Boy(boyname='c1.com'),
    Boy(boyname='c2.com'),
    Girl(girlname='A组'),
    Girl(girlname='B组'),
])


bg = Boy2Girl(boy_id=1, girl_id=1)
session.add(bg)


boy = Boy(boyname='lqz')
boy.girls=[Girl(girlname="lyf"),Girl(girlname='dlrb')]

3.常用基本语法

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="lqz"),
    Users(name="egon"),
    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" : "lqz"})
#类似于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()
"""
# ################ 查询 ################
"""
r1 = session.query(Users).all()
#只取age列，把name重命名为xx
r2 = session.query(Users.name.label('xx'), Users.age).all()
#filter传的是表达式，filter_by传的是参数
r3 = session.query(Users).filter(Users.name == "lqz").all()
r4 = session.query(Users).filter_by(name='lqz').all()
r5 = session.query(Users).filter_by(name='lqz').first()
#:value 和:name 相当于占位符，用params传参数
r6 = session.query(Users).filter(text("id<:value and name=:name")).params(value=224, name='fred').order_by(Users.id).all()
#自定义查询sql
r7 = session.query(Users).from_statement(text("SELECT * FROM users where name=:name")).params(name='ed').all()
"""

#增，删，改都要commit()
session.close()

#　条件
ret = session.query(Users).filter_by(name='lqz').all()
#表达式，and条件连接
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_
#or_包裹的都是or条件，and_包裹的都是and条件
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()


# 通配符，以e开头，不以e开头
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]

# 排序，根据name降序排列（从大到小）
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()
#分组之后取最大id，id之和，最小id
ret = session.query(
    func.max(Users.id),
    func.sum(Users.id),
    func.min(Users.id)).group_by(Users.name).all()
#haviing筛选
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()

# 连表（默认用forinkey关联）

ret = session.query(Users, Favor).filter(Users.id == Favor.nid).all()
#join表，默认是inner join
ret = session.query(Person).join(Favor).all()
#isouter=True 外连，表示Person left join Favor，没有右连接，反过来即可
ret = session.query(Person).join(Favor, isouter=True).all()
#打印原生sql
aa=session.query(Person).join(Favor, isouter=True)
print(aa)
# 自己指定on条件（连表条件）,第二个参数，支持on多个条件，用and_,同上
ret = session.query(Person).join(Favor,Person.id==Favor.id, isouter=True).all()
# 组合（了解）UNION 操作符用于合并两个或多个 SELECT 语句的结果集
#union和union 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()

Flask-SQLAlchemy和Flsk-migraste

1.安装模块：

pip3 install flask-sqlalchemy

pip3 install flask-migrate

使用

-flask-sqlachemy
        -安装
        -from flask_sqlalchemy import SQLAlchemy
        -db=SQLAlchemy()
        - # 将db注册到app中
        -db.init_app(app)
        -创建表模型，都继承db.Model
        -增删查改：db.session

    -flask-migrate
        -安装（flask-script）
        -from flask_migrate import Migrate, MigrateCommand
        -Migrate(app, db)
        -manager.add_command('db', MigrateCommand)
        
        -python3 manage.py db init 初始化：只执行一次(刚开始用执行)
        -python3 manage.py db migrate 等同于 makemigartions
        -python3 manage.py db upgrade 等同于migrate

2.相关配置

2.1 settings.py：sqlachmy数据库连接池配置

class BaseConfig(object):
    # SESSION_TYPE = 'redis'  # session类型为redis
    # SESSION_KEY_PREFIX = 'session:'  # 保存到session中的值的前缀
    # SESSION_PERMANENT = True  # 如果设置为False，则关闭浏览器session就失效。
    # SESSION_USE_SIGNER = False  # 是否对发送到浏览器上 session:cookie值进行加密

    SQLALCHEMY_DATABASE_URI = "mysql+pymysql://root:123@127.0.0.1:3306/flask01?charset=utf8"
    SQLALCHEMY_POOL_SIZE = 5
    SQLALCHEMY_POOL_TIMEOUT = 30
    SQLALCHEMY_POOL_RECYCLE = -1

    # 追踪对象的修改并且发送信号
    SQLALCHEMY_TRACK_MODIFICATIONS = False


class ProductionConfig(BaseConfig):
    pass


class DevelopmentConfig(BaseConfig):
    pass


class TestingConfig(BaseConfig):
    pass

2.2 __init__.py：

from flask import Flask

from flask_migrate import Migrate, MigrateCommand

#1.导入模块
from flask_sqlalchemy import SQLAlchemy

#2.产生db对象
db = SQLAlchemy()
#导入表和视图
from .models import *
from .views import account

from flask_script import Manager


def create_app():
    app = Flask(__name__)
    #3.引入配置
    app.config.from_object('settings.DevelopmentConfig')

    # 4.将db注册到app中
    db.init_app(app)


    # 注册蓝图
    app.register_blueprint(account.account)
    manager=Manager(app)

    migrate = Migrate(app, db)
    manager.add_command('db', MigrateCommand)

    return manager

models.py

#导入init.py中的db对象
from . import db

class Users(db.Model):
    """
    用户表
    """
    __tablename__ = 'users'
    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(80), unique=True, nullable=False)
    email = db.Column(db.String(120), unique=True, nullable=False)
    email2 = db.Column(db.String(120), unique=True, nullable=False)

    def __repr__(self):
        return '<User %r>' % self.username

视图使用

from flask import Blueprint
from .. import db
from .. import models

account = Blueprint('account', __name__)


@account.route('/login')
def login():
    db.session.add(models.Users(username='lqz', email='123'))
    db.session.commit()
    # 添加示例
    """
    db.session.add(models.Users(username='lqz', pwd='123', gender=1))
    db.session.commit()

    obj = db.session.query(models.Users).filter(models.Users.id == 1).first()
    print(obj)

项目依赖包

1.安装模块

pip3 install pipreqs

2.生成包

方式1：
pip3 freeze > 具体文件路径
例：
pip3 freeze > d:/requirements.txt


方式2：pipreq模块
pipreqs ./ --encoding=utf8
        

3.安装包

pip3 install -r 包的具体文件路径