Flask操作数据库:
需要提前安装:flask_sqlalchemy,(pip install flask_sqlalchemy)如果连接的是mysql数据库,需要安装mysqldb驱动;(pip install flask_mysqldb)
一对多:
代码如下:
from flask import Flask
from flask_sqlalchemy import SQLAlchemy
class Config(object):
"""项目配置类"""
# 连接数据库
# mysql://数据库的账号:密码/数据库ip地址:端口号/数据库名称
SQLALCHEMY_DATABASE_URI = "mysql://root:123456@127.0.0.1:3306/py18"
# 跟踪数据库修改操作
SQLALCHEMY_TRACK_MODIFICATIONS = True
# 1:创建app对象
app = Flask(__name__)
# 加载项目配置
app.config.from_object(Config)
# 创建数据库对象
db = SQLAlchemy(app)
# 定义数据库表模型类
class User(db.Model):
# 设置表名称,不设置默认为类名的小写字母(user)
__tablename__ = "users"
# 定义字段
id = db.Column(db.Integer, primary_key=True)
name = db.Column(db.String(128), unique=True)
# role = Role()
# role.users:该角色下有哪些用户
# user=User()
# user.roles:该用户属于那种角色
# 定义关系字段relationship,这个字段在数据库是不存在的,只是为了方便查询
roles = db.relationship("Role", backref="users")
def __repr__(self):
"""自定义格式化输出"""
return "Role: %s %s" % (self.id, self.name)
# 自定义数据库表的模型类:db.Model
class Role(db.Model):
__tablename__ = "role"
id = db.Column(db.Integer, primary_key=True)
name = db.Column(db.String(128), unique=True)
# 定义外键
role_id = db.Column(db.Integer, db.ForeignKey(User.id))
def __repr__(self):
return "User:%s %s" % (self.id, self.name)
@app.route("/")
def hello_world():
return "hello world!"
if __name__ == '__main__':
# 删除所有表
db.drop_all()
# 创建所有表
db.create_all()
app.run(debug=True)
多对多:
代码如下:
from flask import Flask
from flask_sqlalchemy import SQLAlchemy
class Config(object):
"""项目配置类"""
# 连接数据库
# mysql://数据库的账号:密码/数据库ip地址:端口号/数据库名称
SQLALCHEMY_DATABASE_URI = "mysql://root:123456@127.0.0.1:3306/py18"
# 跟踪数据库修改操作
SQLALCHEMY_TRACK_MODIFICATIONS = True
# 1:创建app对象
app = Flask(__name__)
# 加载项目配置
app.config.from_object(Config)
# 创建数据库对象
db = SQLAlchemy(app)
# 定义第三张表
tb_user_role = db.Table("tb_user_role",
db.Column("user_id", db.Integer, db.ForeignKey("user.id")),
db.Column("role_id", db.Integer, db.ForeignKey("role.id"))
)
# 定义数据库表模型类
class User(db.Model):
# 设置表名称,不设置默认为类名的小写字母(user)
__tablename__ = "user"
# 定义字段
id = db.Column(db.Integer, primary_key=True)
name = db.Column(db.String(128), unique=True)
# role = Role()
# role.users:该角色下有哪些用户
# user=User()
# user.roles:该用户属于那种角色
# 定义关系字段relationship,这个字段在数据库是不存在的,只是为了方便查询
# TODO 添加第三张表
# secondary 指明第三张表, lazy提高性能,值返回结果对象,需要获取数据时需要".all()"才会展示出来
roles = db.relationship("Role",
backref="user",
secondary=tb_user_role,
lazy="dynamic")
def __repr__(self):
"""自定义格式化输出"""
return "Role: %s %s" % (self.id, self.name)
# 自定义数据库表的模型类:db.Model
class Role(db.Model):
__tablename__ = "role"
id = db.Column(db.Integer, primary_key=True)
name = db.Column(db.String(128), unique=True)
# 定义外键
role_id = db.Column(db.Integer, db.ForeignKey(User.id))
def __repr__(self):
return "User:%s %s" % (self.id, self.name)
@app.route("/")
def hello_world():
return "hello world!"
if __name__ == '__main__':
# 删除所有表
db.drop_all()
# 创建所有表
db.create_all()
app.run(debug=True)
//////////////////////////////////////////////////////////
数据常用查询操作
from flask_sqlalchemy import SQLAlchemy
from flask import Flask
from sqlalchemy import not_, and_, or_
class Config(object):
"""项目配置类"""
SQLAlCHEMY_DATABASE_URI = "mysql://root:123456@127.0.0,1:3306/test18"
SQLALCHEMY_TRACK_MODIFICATIONS = True
app = Flask(__name__)
app.config.from_object(Config)
db = SQLAlchemy(app)
class Role(db.Model):
"""角色表"""
id = db.Column(db.Integer, primary_key=True)
name = db.Column(db.String(128), unique=True)
users = db.relationship("User", backref="role")
def __repr__(self):
return "Role:%s %s" %(self.id, self.name)
class User(db.Model):
id= db.Column(db.Integer, primary_key=True)
name = db.Column(db.String(128), unique=True)
email = db.Column(db.String(64))
password = db.Column(db.String(64))
# 定义外键
role_id = db.Column(db.Integer, db.ForeignKey(Role.id))
def __repr__(self):
return "User:%s %s %s %s" %(self.id, self.name, self.email, self.role_id)
@app.route("/")
def hello_world():
return "hello world"
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='wang', email='wang@163.com', password='123456', role_id=ro1.id)
us2 = User(name='zhang', email='zhang@189.com', password='201512', role_id=ro2.id)
us3 = User(name='chen', email='chen@126.com', password='987654', role_id=ro2.id)
us4 = User(name='zhou', email='zhou@163.com', password='456789', role_id=ro1.id)
us5 = User(name='tang', email='tang@itheima.com', password='158104', role_id=ro2.id)
us6 = User(name='wu', email='wu@gmail.com', password='5623514', role_id=ro2.id)
us7 = User(name='qian', email='qian@gmail.com', password='1543567', role_id=ro1.id)
us8 = User(name='liu', email='liu@itheima.com', password='867322', role_id=ro1.id)
us9 = User(name='li', email='li@163.com', password='4526342', role_id=ro2.id)
us10 = User(name='sun', email='sun@163.com', password='235523', role_id=ro2.id)
db.session.add_all([us1, us2, us3, us4, us5, us6, us7, us8, us9, us10]) db.session.commit() # 查询所有用户数据 User.query.all() # 查询有多少个用户 User.query.count() # 查询id为4的用户[3种方式] User.query.get(4) # filter_by精确查询 User.query.filter_by(id=4).first() User.query.filter(User.id == 4).first() # 查询名字结尾字符为g的所有数据[开始startswith/包含contains] User.query.filter(User.name.endswith("g")).all() User.query.filter(User.name.contains("g")).all() # 查询名字不等于wang的所有数据[2种方式] User.query.filter(User.name != "wang").all() User.query.filter(not_(User.name == "wang")).all() # 查询名字和邮箱都以"li"开头的所有数据[2种方式] User.query.filter(User.name.startswith('li'), User.email.startswith('li')) User.query.filter(and_(User.name.startswith("li"), User.email.startswith("li"))) # 查询password是"123456"或者"email"以"itheima.com"结尾的邮箱 User.query.filter(or_(User.password == "123456", User.email.endswith("itheima.com"))) # 查询id为[1,3,5,7,9]的用户列表 # 使用in_()包含函数 User.query.filter(User.id in([1,3,5,7,9])).all() # 查询name为liu的数据 User.query.filter(User.name =="liu").first() # 查询所有用户数据,并以邮箱降序排列 User.query.filter(User.email.desc()).all() # 将所有数据分页,每页3个,查询第二页的数据 # 参数1:第几页的数据, 参数2:每页多少条数据 paginate = User.query.paginate(2, 3) # 获取当前页面所有数据 paginate.items # 获取当前页码 paginate.page # 获取总页数 paginate.pages app.run(debug=True)