flask-sqlalchemy组件

一、简介

　　flask本身没有内置orm框架，需要依赖第三方模块，这里介绍flask-sqlalchemy,而flask-sqlalchemy是一个flask的扩展，本质上是对sqlalchemy的进一步封装，当然也需要sqlalchemy的支持，使用起来和本来的sqlalchemy的orm是一样的。本文主要介绍sqlalchemy的使用以及如何在flask中是用orm。 

二、sqlalchemy

　　sqlalchemy是python编程语言下的一款orm框架，该框架建立在数据库api之上，使用关系对象映射进行数据库操作，简言之便是：将对象转换成sql，然后使用数据api执行sql并获取执行结果。图示：

基本使用

安装：

pip3 install sqlalchemy

注：sqlalchemy无法修改表结构，如果需要可以使用sqlalchemy开发者开源的另外一个软件alembic来完成，官网doc：http://docs.sqlalchemy.org/en/latest/core/expression_api.html

原生sql

使用 engine/connectionpooling/dialect 进行数据库操作，engine使用connectionpooling连接数据库，然后再通过dialect执行sql语句

#!/usr/bin/env python3
#_*_ coding:utf-8 _*_
#author:wd
from sqlalchemy import create_engine
engine=create_engine("mysql+pymysql://stu:1234qwer@10.0.0.241:3307/student", max_overflow=5)#创建连接，允许溢出5个连接
result = engine.execute('select * from student')#使用excute执行原生sql
print(result.fetchall())#获取所有结果，与pymyql类似

事务

#!/usr/bin/env python3
#_*_ coding:utf-8 _*_
#author:wd
from sqlalchemy import create_engine
engine=create_engine("mysql+pymysql://stu:1234qwer@10.0.0.241:3307/student", max_overflow=5)#创建连接，允许溢出5个连接
result = engine.execute('select * from student')#使用excute执行原生sql
with engine.begin() as conn: #事务操作
    conn.execute("insert into student (name, age, res_date) values ('weikang', 33, '1992-11-11')")
    
print(result.fetchall())#获取所有结果，与pymyql类似

建表

　　定义数据表，才能进行sql表达式的操作，毕竟sql表达式的表的确定，是sqlalchemy制定的，如果数据库已经存在了数据表还需要定义么？当然，这里其实是一个映射关系，如果不指定，查询表达式就不知道是附加在那个表的操作，当然定义的时候，注意表名和字段名，代码和数据的必须保持一致。定义好之后,就能创建数据表，一旦创建了，再次运行创建的代码，数据库是不会创建的。

　　sqlalchemy内部组件调用顺序为：使用 schema type/sql expression language/engine/connectionpooling/dialect 进行数据库操作。engine使用schema type创建一个特定的结构对象，之后通过sql expression language将该对象转换成sql语句，然后通过 connectionpooling 连接数据库，再然后通过 dialect 执行sql，并获取结果。

tips：使用类的方式和使用metadata方式创建表时候区别在于metadata可以不指定主键，而是用class方式必须要求有主键。

demo1:

from sqlalchemy import create_engine,table,column,integer,string,foreignkey,metadata

engine=create_engine("mysql+pymysql://stu:1234qwer@10.0.0.241:3307/student?charset=gbk",
                     encoding="utf-8",
                     echo=true,
                      max_overflow=5
                     )
#?charset是字符集编码，echo=true打印输出信息和执行的sql语句默认flase，max_overflow=5允许溢出连接池连接数量
meta=metadata()#生成源类
#定义表结构
user=table('user',meta,
           column('id',integer,nullable=table,autoincrement=true,primary_key=true),
           column('name',string(20),nullable=true),
           column('age',integer,nullable=true)
           )

host=table('host',meta,
            column('ip',string(20),nullable=true),
            column('hostname',string(20),nullable=true),

)
meta.create_all(engine)#创建表，如果存在则忽视

demo2:

使用orm基类创建

#!/usr/bin/env python3
#_*_ coding:utf-8 _*_
#author:wd
from sqlalchemy import create_engine,table,column,integer,string,foreignkey,metadata,date
from sqlalchemy.ext.declarative import declarative_base

engine=create_engine("mysql+pymysql://stu:1234qwer@10.0.0.241:3307/student?charset=gbk",
                     encoding="utf-8",
                     echo=true,
                      max_overflow=5
                     )
#?charset是字符集编码，echo=true打印输出信息和执行的sql语句默认flase，max_overflow=5允许溢出连接池连接数量
base=declarative_base()#生成orm基类
#定义表结构
class user(base):
    __tablename__='book' #表明
    id = column(integer, primary_key=true)
    name=column(string(32))
    date=column(date)

base.metadata.create_all(engine)#创建表，如果存在则忽视

orm增删改查

增

#!/usr/bin/env python3
#_*_ coding:utf-8 _*_
#author:wd
from sqlalchemy import create_engine,table,column,integer,string,foreignkey
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.declarative import declarative_base
engine=create_engine("mysql+pymysql://stu:1234qwer@10.0.0.241:3307/student?charset=gbk",
                     max_overflow=5,
                     echo=true)
#数据库连接信息为，连接类型://用户名:密码@数据库地址:端口／数据库名字？编码
#max_overflow创建连接，允许溢出5个连接,echo=true，输出相应的sql信息到控制台，方便调试。

base=declarative_base()#生成orm基类

class user(base):  #映射表
    __tablename__='user'
    id=column(integer,autoincrement=true,primary_key=true)
    name=column(string(20))
    age=column(integer)

sessoion_class=sessionmaker(bind=engine)#创建与数据库的会话类，这里的sessoion_class是类
session=sessoion_class()#生成会话实例
user1=user(name='wd',age=22)#生成user对象
session.add(user1)   #添加user1，可以使用add_all,参数为列表或者tuple
session.commit()     #提交
#session.rollback()  #回滚
session.close()      #关闭会话

删

data=session.query(user).filter(user.age==33).delete()
session.commit()        #提交
session.close()      #关闭会话

改

#data=session.query(user).filter(user.age>20).update({"name":'jarry'})#update语法
data=session.query(user).filter(user.age==22).first()#面向对象语法
data.name='coco'#如果data中数据多条需要使用for循环设置
session.commit()        #提交
session.close()      #关闭会话

查

#!/usr/bin/env python3
#_*_ coding:utf-8 _*_
#author:wd
from sqlalchemy import create_engine,table,column,integer,string,foreignkey
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.declarative import declarative_base
engine=create_engine("mysql+pymysql://stu:1234qwer@10.0.0.241:3307/student?charset=gbk",
                     max_overflow=5,
                     echo=true)
#数据库连接信息为，连接类型://用户名:密码@数据库地址:端口／数据库名字？编码
#max_overflow创建连接，允许溢出5个连接,echo=true，输出相应的sql信息到控制台，方便调试。

base=declarative_base()#生成orm基类

class user(base):  #映射表
    __tablename__='user'
    id=column(integer,autoincrement=true,primary_key=true)
    name=column(string(20))
    age=column(integer)

    def __repr__(self):   #定义
        return "(%s,%s,%s)" % (self.id,self.name,self.age)

sessoion_class=sessionmaker(bind=engine)#创建与数据库的会话类，这里的sessoion_class是类
session=sessoion_class()#生成会话实例



#data=session.query(user).get(2)  #get语法获取primrykey中的关键字，在这里主键为id，获取id为2的数据
#data=session.query(user).filter(user.age>22,user.name=='mack').first()
#filter语法两个等于号，filter_by语法一个等于号，可以有多个filter,如果多个数据返回列表，first代表获取第一个，为all()获取所有
data=session.query(user).filter(user.age>20,user.name.in_(['mack','wd'])).all()#in语法
print(data[0]) #打印第一个结果
session.commit()        #提交，如果回滚的话，数据将不存在了
session.close()      #关闭会话

common filter operators

here’s a rundown of some of the most common operators used in filter():

equals:

     query.filter(user.name == 'ed')
not equals:

     query.filter(user.name != 'ed')
like:

query.filter(user.name.like('%ed%'))

in:

not in:
query.filter(~user.name.in_(['ed', 'wendy', 'jack']))

is null:

is not null:

and:
2.1. objectrelationaltutorial 17

query.filter(user.name.in_(['ed', 'wendy', 'jack']))
# works with query objects too:
query.filter(user.name.in_( session.query(user.name).filter(user.name.like('%ed%'))

))

query.filter(user.name == none)
# alternatively, if pep8/linters are a concern
query.filter(user.name.is_(none))
query.filter(user.name != none)
# alternatively, if pep8/linters are a concern
query.filter(user.name.isnot(none))
sqlalchemy documentation, release 1.1.0b1

# use and_()

from sqlalchemy import and_
query.filter(and_(user.name == 'ed', user.fullname == 'ed jones'))

# or send multiple expressions to .filter()
query.filter(user.name == 'ed', user.fullname == 'ed jones')
# or chain multiple filter()/filter_by() calls
query.filter(user.name == 'ed').filter(user.fullname == 'ed jones')
note: makesureyouuseand_()andnotthepythonandoperator! • or:

note: makesureyouuseor_()andnotthepythonoroperator! • match:

query.filter(user.name.match('wendy'))
note: match() uses a database-specific match or contains f 

常用查询语法

常用查询

其他操作

##获取所有数据
data=session.query(user).all()#获取user表所有数据
for i in data:
    print(i)


##统计
#count=session.query(user).count()#获取所有的条数
count=session.query(user).filter(user.name.like("ja%")).count()#获取某些条数
print(count)


##分组
from sqlalchemy import func#需要导入func函数
res=session.query(func.count(user.name),user.name).group_by(user.name).all()
print(res)

外间关联

tips：设置外检的另一种方式 foreignkeyconstraint(['other_id'], ['othertable.other_id'])

#!/usr/bin/env python3
#_*_ coding:utf-8 _*_
#author:wd
from sqlalchemy import create_engine,table,column,integer,string,foreignkey
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import relationship

engine=create_engine("mysql+pymysql://stu:1234qwer@10.0.0.241:3307/student?charset=gbk",
                     encoding="utf-8",
                     echo=true,
                      max_overflow=5
                     )
#?charset是连接数据库的字符集编码（和数据库的编码一样），echo=true打印输出信息和执行的sql语句默认flase，max_overflow=5允许溢出连接池连接数量


base=declarative_base()

class user(base):
    __tablename__='user'
    id=column(integer,primary_key=true,autoincrement=true)
    name=column(string(20))
    age=column(integer)
    def __repr__(self):
        return "<id:%s,name:%s,age:%s>"%(self.id,self.name,self.age)

class host(base):
    __tablename__='host'
    user_id=column(integer,foreignkey('user.id'))#user_id关联user表中的id
    hostname=column(string(20))
    ip=column(string(20),primary_key=true)
    host_user=relationship('user',backref='user_host')
    #通过host_user查询host表中关联的user信息,通过user_host，在user表查询关联的host，与生成的表结构无关，只是为了方便查询
    def __repr__(self):
        return "<user_id:%s,hostname:%s,ip:%s>"%(self.user_id,self.hostname,self.ip)

base.metadata.create_all(engine)
session_class=sessionmaker(bind=engine)
session=session_class()
host1=session.query(host).first()
print(host1.host_user)
print(host1)
user1=session.query(user).first()
print(user1.user_host)

多外键关联一个表中的一个字段

应用场景：当我们购物时候，你会发现有一个收发票地址，和一个收货地址。关系如下：默认情况下，发票地址和收获地址是一致的，但是也有可能我想买东西送给别人，而发票要自己留着，那收货的地址和寄送发票的地址可以不同。即：同一个人的两个收获地址可以不同，多个收获地址关联同一个人。

#!/usr/bin/env python3
#_*_ coding:utf-8 _*_
#author:wd
from sqlalchemy import integer, foreignkey, string, column
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import relationship

base = declarative_base()


class customer(base):
    __tablename__ = 'customer'
    id = column(integer, primary_key=true)
    name = column(string)

    billing_address_id = column(integer, foreignkey("address.id"))
    shipping_address_id = column(integer, foreignkey("address.id"))

    billing_address = relationship("address", foreign_keys=[billing_address_id])
    shipping_address = relationship("address", foreign_keys=[shipping_address_id])
    #同时关联同一个字段，使用relationship需要指定foreign_keys，为了让sqlalchemy清楚关联的外键

class address(base):
    __tablename__ = 'address'
    id = column(integer, primary_key=true)
    street = column(string)
    city = column(string)
    state = column(string)

多对多外键关联

很多时候，我们会使用多对多外键关联，例如：书和作者，学生和课程，即：书可以有多个作者，而每个作者可以写多本书，orm提供了更简单方式操作多对多关系，在进行删除操作的时候，orm会自动删除相关联的数据。

表结构创建

#!/usr/bin/env python3
#_*_ coding:utf-8 _*_
#author:wd
from sqlalchemy import column,table,string,integer,foreignkey
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import create_engine
from sqlalchemy.orm import relationship

engine=create_engine("mysql+pymysql://stu:1234qwer@10.0.0.241:3307/student?charset=gbk",
                     encoding="utf-8",
                     echo=true,
                      max_overflow=5
                     )

base=declarative_base()
stu_cour=table('stu_cour',base.metadata,
               column('stu_id',integer,foreignkey('student.id')),
               column('cour_id',integer,foreignkey('course.id'))
               )

class student(base):
    __tablename__='student'
    id=column(integer,autoincrement=true,primary_key=true)
    stu_name=column(string(32))
    stu_age=column(string(32))
    courses=relationship('course',secondary=stu_cour,backref='students')
    #course是关联的第一张表，stu_cour是关联的第二张表，当然，也可以在第三张关联表中使用两个relationship关联student表和course表
    def __repr__(self):
        return '<%s>'%self.stu_name


class course(base):
    __tablename__='course'
    id=column(integer,autoincrement=true,primary_key=true)
    cour_name=column(string(32))
    def __repr__(self):
        return '<%s>'%self.cour_name
base.metadata.create_all(engine)

创建表结构

建表

插入数据

#!/usr/bin/env python3
#_*_ coding:utf-8 _*_
#author:wd
from sqlalchemy import column,table,string,integer,foreignkey
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import create_engine
from sqlalchemy.orm import relationship
from sqlalchemy.orm import sessionmaker

engine=create_engine("mysql+pymysql://stu:1234qwer@10.0.0.241:3307/student?charset=gbk",
                     encoding="utf-8",
                     echo=true,
                      max_overflow=5
                     )

base=declarative_base()
stu_cour=table('stu_cour',base.metadata,
               column('stu_id',integer,foreignkey('student.id')),
               column('cour_id',integer,foreignkey('course.id'))
               )

class student(base):
    __tablename__='student'
    id=column(integer,autoincrement=true,primary_key=true)
    stu_name=column(string(32))
    stu_age=column(string(32))
    courses=relationship('course',secondary=stu_cour,backref='students')
    #course是关联的第一张表，stu_cour是关联的第二张表，当然，也可以在第三张关联表中使用两个relationship关联student表和course表
    def __repr__(self):
        return '<%s>'%self.stu_name


class course(base):
    __tablename__='course'
    id=column(integer,autoincrement=true,primary_key=true)
    cour_name=column(string(32))
    def __repr__(self):
        return '<%s>'%self.cour_name

stu1=student(stu_name='wd',stu_age='22')
stu2=student(stu_name='jack',stu_age=33)
stu3=student(stu_name='rose',stu_age=18)
c1=course(cour_name='linux')
c2=course(cour_name='python')
c3=course(cour_name='go')
stu1.courses=[c1,c2]  #添加学生课程关联
stu2.courses=[c1]
stu3.courses=[c1,c2,c3]
session_class=sessionmaker(bind=engine)
session=session_class()
session.add_all([stu1,stu2,stu3,c1,c2,c3])
session.commit()

数据插入

插入数据

查询

session_class=sessionmaker(bind=engine)
session=session_class()
stu_obj=session.query(student).filter(student.stu_name=='wd').first()
print(stu_obj.courses)#查询wd学生所报名的课程
cour_obj=session.query(course).filter(course.cour_name=='python').first()
print(cour_obj.students)#查询报名python课程所对应的课程
session.commit()

查询

删除

session_class=sessionmaker(bind=engine)
session=session_class()
cour_obj=session.query(course).filter(course.cour_name=='python').first()
session.delete(cour_obj)#删除python课程
session.commit()

删除

三、flask-sqlalchemy

flask中使用sqlalchemy时候，个人比较习惯安装django的项目来进行构建，以下示例以登陆验证进行说明,项目结构：

taskmanager
├── app01
│   ├── __init__.py       #初始化文件
│   ├── models.py        #数据模型
│   └── views
│       └── account.py   # 视图函数
├── create_table.py    # 建表
├── run.py             # 启动服务器
├── settings.py        #配置文件
├── static             #静态资源
└── templates          #模版
    └── login.html

安装：

pip3 install flask-sqlalchemy

数据库配置

sqlalchemy_database_uri    #用于连接的数据库 uri 。例如:sqlite:////tmp/test.dbmysql://username:password@server/db
sqlalchemy_binds     #一个映射 binds 到连接 uri 的字典。更多 binds 的信息见用 binds 操作多个数据库。
sqlalchemy_echo    #如果设置为ture， sqlalchemy 会记录所有 发给 stderr 的语句，这对调试有用。(打印sql语句)
sqlalchemy_record_queries    #可以用于显式地禁用或启用查询记录。查询记录 在调试或测试模式自动启用。更多信息见get_debug_queries()。
sqlalchemy_native_unicode        #可以用于显式禁用原生 unicode 支持。当使用 不合适的指定无编码的数据库默认值时，这对于 一些数据库适配器是必须的（比如 ubuntu 上 某些版本的 postgresql ）。
sqlalchemy_pool_size    #数据库连接池的大小。默认是引擎默认值（通常 是 5 ）
sqlalchemy_pool_timeout    #设定连接池的连接超时时间。默认是 10 。
sqlalchemy_pool_recycle    #多少秒后自动回收连接。这对 mysql 是必要的， 它默认移除闲置多于 8 小时的连接。注意如果 使用了 mysql ， flask-sqlalchemy 自动设定 这个值为 2 小时。

参考

新建立settings.py

#!/usr/bin/env python3
# -*- coding:utf-8 -*-
# author:wd
import redis
class baseconfig(object):
    #session配置
    session_type = 'redis'  # session类型为redis
    session_key_prefix = 'session:'  # 保存到session中的值的前缀
    session_permanent = true  # 如果设置为false，则关闭浏览器session就失效。
    session_use_signer = false  # 是否对发送到浏览器上 session:cookie值进行加密
    session_redis= redis.redis(host='10.1.210.33', port='6379')
    #数据库配置
    sqlalchemy_database_uri = "mysql+pymysql://root:1234qwer@10.1.210.33:3306/devops?charset=utf8"
    sqlalchemy_pool_size = 10    #数据库连接池的大小。默认值 5
    sqlalchemy_pool_timeout = 30  # 指定数据库连接池的超时时间。默认是 10
    sqlalchemy_pool_recycle = -1
    sqlalchemy_max_overflow = 3  # 控制在连接池达到最大值后可以创建的连接数。当这些额外的连接回收到连接池后将会被断开和抛弃
    sqlalchemy_track_modifications = false  # 追踪对象的修改并且发送信号


class productionconfig(baseconfig):
    """生产配置文件"""
    pass


class developmentconfig(baseconfig):
    """开发配置文件"""
    pass


class testingconfig(baseconfig):
    """
    测试配置文件
    """
    pass

在app目录下建立models.py

from . import db


class userprofile(db.model):
    """
    用户
    """
    __tablename__ = 'userprofile'
    id = db.column(db.integer, primary_key=true)
    username = db.column(db.string(64), unique=true, nullable=false)
    password = db.column(db.string(64), unique=true, nullable=false)
    email = db.column(db.string(128), unique=true, nullable=false)

    def __repr__(self):
        return '<user %s>' % self.username

创建视图函数初始化app，app目录新建__init__.py 

from flask import flask
from flask_sqlalchemy import sqlalchemy
from .models import *
from .views import account


db = sqlalchemy()  #实例化

def init_app():
    app = flask(__name__, template_folder='../templates', static_folder='../static', static_url_path='/static')
    app.config.from_object('settings.developmentconfig')

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

    # 注册蓝图
    app.register_blueprint(account.account)

    return app

在app目录创建目录views单个py文件也可以

#!/usr/bin/env python3
# -*- coding:utf-8 -*-
# author:wd

from flask import blueprint
from flask import request
from flask import render_template
from .. import db
from .. import models

account = blueprint('account', __name__)


@account.route('/login',methods=['get','post'])
def login():
    if request.method=="get":
        return render_template("login.html")
    else:
        user_obj=db.session.query(models.userprofile).filter(models.userprofile.username==request.form["username"],
                                                             models.userprofile.password==request.form["password"]).first()
        db.session.close()
        if user_obj:
            return '登陆成功'
        else:
            return render_template("login.html",errors="用户名或密码错误！")

templates/login.html

<!doctype html>
<html lang="en">
<head>
    <meta charset="utf-8">
    <title>title</title>
</head>
<body>
<div><h1>用户登陆</h1></div>
<div>
    <form method="post">
        <input type="text" name="username" placeholder="用户名">
        <input type="password" name="password" placeholder="密码">
        <input type="submit">
        {{ errors }}
    </form>
</div>
</body>
</html>

login.html

创建启动脚本run.py 

#!/usr/bin/env python3
# -*- coding:utf-8 -*-
# author:wd

from app01 import init_app

app = init_app()  # 创建app

if __name__ == '__main__':
    app.run()

run.py

离线创建数据库表脚本create_table.py

#!/usr/bin/env python3
# -*- coding:utf-8 -*-
# author:wd

from app01 import init_app
from app01 import db

app = init_app()  #创建app

with app.app_context():  # 执行脚本创建数据库表
    db.create_all()

create_table.py