Flask-SQLAlchemy是一个Flask扩展，简化了在Flask程序中使用SQLAlchemy的操作。SQLAlchemy是一个很强大的关系型数据库框架，支持多种数据库后台。SQLAlchemy提供了高层ORM，也提供了使用数据库原生SQL的低层功能。

1、安装

1).安装flask-sqlalchemy

pip install flask-sqlalchemy

2). 安装mysqldb

pip install flask-mysqldb

2、配置

1). 程序使用的数据库URL必须保存到Flask配置对象的SQLALCHEMY_DATABASE_URI键中。

app.config['SQLALCHEMY_DATABASE_URI'] = 'mysql://root:password@127.0.0.1:3306/test'

2). 其他

配置对象中还有一个很有用的选项，即SQLALCHEMY_COMMIT_ON_TEARDOWN键，将其设为True时，每次请求结束后都会自动提交数据库中的变动。

配置的列表如下：

3). 连接数据库

不同的数据库采用不同的引擎连接语句：

(1)MySQL： mysql://username:password@hostname/database

(2)SQLite(Unix): sqlite:absolute/path/to/database

(3)SQLite(Windows): sqlite:///c:/absolute/path/to/database

(4)Oracle: oracle://username:password@127.0.0.1:3306/database

参考代码如下：

from flask import Flask

from flask_sqlalchemy import SQLAlchemy

app = Flask(__name__)

# 设置连接数据库的URL

app.config['SQLALCHEMY_DATABASE_URI'] = 'mysql://root:password@127.0.0.1:3306/test'

# 设置每次请求结束后会自动提交数据库的改动

app.config['SQLALCHEMY_COMMIT_ON_TEARDOWN'] = True

app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = True

# 查询时显示原始SQL语句

app.config['SQLALCHEMY_ECHO'] = True

db = SQLAlchemy(app)

3、定义模型

在ORM中，模型一般是一个Python类，类中的属性对应数据库表中的列。

Flask-SQLAlchemy创建的数据库实例为模型提供了一个基类以及一系列辅助类和辅助函数，可用于定义模型的结构。

常用的sqlalchemy字段类型：

常用的sqlalchemy列表项

常用的sqlalchemy关系选项

参考代码如下：

class Role(db.Model):

__tablename__ = 'roles'

id = db.Column(db.Integer, primary_key=True)

name = db.Column(db.String(64))

user = db.relationship('User', backref='role')

def __repr__(self):

return '' % self.name

class User(db.Model):

__tablename__ = 'users'

id = db.Column(db.Integer, primary_key=True)

name = db.Column(db.String(64), unique=True, index=True)

email = db.Column(db.String(64), unique=True)

pswd = db.Column(db.String(64))

role_id = db.Column(db.Integer, db.ForeignKey('roles.id'))

def __repr__(self):

return 'User:%s' % self.name

4、操作数据库

创建表：db.create_all()

删除表: db.drop_all()

插入行:

db.session.add()

db.session.addAll([])

db.session.commit()

查询全部数据: User.query.all()

过滤查询: User.query.filter_by(id=id).first()

join多表查询：User.query.filter_by(env_id=env_id,id=id).join(Environments,Variable.env_id == Environments.id).first_or_404()

count返回数量: User.query.filter_by(id=id).count()

修改数据

#根据条件查询一行数据

admin_role = Role.query.filter_by(role_name = 'Amdmin').first()

#修改数据-

admin_role.role_name = 'Admin'

db.session.add(admin_role)

db.session.commit()

删除数据：

db.session.delete(User)

db.session.commit()

删除多条数据

variablelists= Variable.query.filter_by(env_id=env_id).all()

for var in variablelists:

db.session.delete(var)

db.session.commit()

直接执行sql语句

sql = u""" update net_internet_ip set removed = current_timestamp() where uuid not in :uuid and removed is null

"""

results = db.session.execute(sql, {"uuid": uuid})

全部代码

全部代码如下，经测试，已跑通。

#!/usr/bin/python

# -*- coding: utf-8 -*-

from flask import Flask

from flask_sqlalchemy import SQLAlchemy

app = Flask(__name__)

# 设置连接数据库的URL

app.config['SQLALCHEMY_DATABASE_URI'] = 'mysql://root:password@127.0.0.1:3306/test'

# 设置每次请求结束后会自动提交数据库的改动

app.config['SQLALCHEMY_COMMIT_ON_TEARDOWN'] = True

app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = True

# 查询时显示原始SQL语句

app.config['SQLALCHEMY_ECHO'] = True

db = SQLAlchemy(app)

class Role(db.Model):

__tablename__ = 'roles'

id = db.Column(db.Integer, primary_key=True)

name = db.Column(db.String(64))

user = db.relationship('User', backref='role')

def __repr__(self):

return '' % self.name

class User(db.Model):

__tablename__ = 'users'

id = db.Column(db.Integer, primary_key=True)

name = db.Column(db.String(64), unique=True, index=True)

email = db.Column(db.String(64), unique=True)

pswd = db.Column(db.String(64))

role_id = db.Column(db.Integer, db.ForeignKey('roles.id'))

def __repr__(self):

return 'User:%s' % self.name

if __name__ == '__main__':

db.drop_all()

db.create_all()

ro1 = Role(name='admin')

ro2 = Role(name='user')

db.session.add_all([ro1, ro2])

db.session.commit()

us1 = User(name='zhangsan', email='zhangsan@qq.com',pswd='12345a',role_id=ro1.id)

us2 = User(name='lisi', email='lisi@qq.com', pswd='12345a', role_id=ro2.id)

db.session.add_all([us1, us2])

db.session.commit()

app.run(debug=True)