一.数据库查询
pip install flask-sqlalchemy
pip install mysqlclient
1. 主要文件结构
extention.py
- 实例化sqlalchemy连接数据库模块
- 和迁移模块
from flask_sqlalchemy import SQLAlchemy
from flask_migrate import Migrate
db: SQLAlchemy = SQLAlchemy()
migrate = Migrate()
models.py
- 多对多关系梳理:
- 中间表名db.Table
- 字段id, 表1外键, 表2外键
用户表UserModel <<--多对多-->> 关联权限表RoleModel roles = db.relationship( - RoleModel, # 填写正向查询的用的表名 - secondary=user_role, # secondary 中间表名 - backref=db.backref('users'), # 提供对方反向查询(填写自己表名) - lazy='dynamic') lazy = 'dynamic'积极查询, 当用户查询执行user.roles就会执行查询语句 正常情况下不配置lazy默认 需要执行user.roles.all()等查询方法才会进行数据库查询
from extention import db
# 用户 & 角色
user_role = db.Table(
"user_to_role", # 表名
db.Column('id', db.Integer, primary_key=True, comment='标识id'),
db.Column('user_id', db.Integer, db.ForeignKey('tb_user.id'), comment='用户编号'),
db.Column('role_id', db.Integer, db.ForeignKey('tb_role.id'), comment='角色编号')
)
# 角色 & 权限
role_power = db.Table(
"role_to_power",
db.Column('id', db.Integer, primary_key=True, autoincrement=True, comment='标识'),
db.Column('power_id', db.Integer, db.ForeignKey('tb_power.id'), comment='用户编号'),
db.Column('role_id', db.Integer, db.ForeignKey('tb_role.id'), comment='角色编号')
)
# 用户
'''
用户 ---> 权限
'''
class UserModel(db.Model):
__tablename__ = 'tb_user'
id = db.Column(db.Integer, primary_key=True, autoincrement=True, comment = '用户id')
username = db.Column(db.String(64), unique=True, nullable=False)
password = db.Column(db.String(64))
roles = db.relationship('RoleModel',
secondary=user_role, # secondary 中间表名
backref=db.backref('users'), # 反向查询
lazy='dynamic') # 正向懒惰查询 直接点方法不会执行查询, 需要.all()
# 角色
class RoleModel(db.Model):
__tablename__ = 'tb_role'
id = db.Column(db.Integer, primary_key=True)
name = db.Column(db.String(64), unique=True, comment='角色唯一名称')
code = db.Column(db.String(64), unique=True, comment='角色唯一标识')
powers = db.relationship('PowerModel',
secondary=role_power,
backref=db.backref('roles'))
# 权限
class PowerModel(db.Model):
__tablename__ = 'tb_power'
id = db.Column(db.Integer, primary_key=True)
url = db.Column(db.String(64), unique=True, comment='权限路径')
code = db.Column(db.String(64), comment='权限标识')
# 自关联
parent_id = db.Column(db.Integer, db.ForeignKey('tb_power.id'), comment='父类编号')
parent = db.relationship('PowerModel', remote_side=[id]) # 自关联
utils.py 其他工具
保存权限字典信息
permission = {
'read': '/auth/read',
'commit': '/auth/comment',
'write': '/auth/write',
'admin': '/admin'
}
app.py主入口文件
from flask import Flask, request, redirect, url_for
from flask import render_template, session
# from auth_decorator import permission_required, login_required, rolu_required, admin_required
from utils import permission
from extention import migrate
from extention import db
from models import UserModel, PowerModel, RoleModel
# from models import UserModel, RoleModel
class Config:
SQLALCHEMY_DATABASE_URI = 'mysql://chf:123456@192.168.50.213:3306/flask'
SQLALCHEMY_TRACK_MODIFICATIONS = False
app = Flask(__name__, template_folder='templates', static_folder='static')
app.secret_key = '1234576'
app.config.from_object(Config)
# 初始化模型
db.init_app(app)
# 生成迁移文件
migrate.init_app(app, db)
# 首页
@app.route('/')
def index():
return render_template('index.html')
@app.cli.command()
def create():
# db.drop_all()
# db.create_all()
# user1 = UserModel(username='chf', password='456')
# user2 = UserModel(username='ozj', password='123')
# db.session.add(user1)
# db.session.add(user2)
# db.session.commit()
# roles = [
# {'name': '普通用户', 'code': 'level1'},
# {'name': '会员用户', 'code': 'level2'},
# {'name': '管理员用户', 'code': 'level3'},
# ]
# role1= RoleModel(name=roles[0]['name'], code= roles[0]['code'])
# role2= RoleModel(name=roles[1]['name'], code= roles[1]['code'])
# role3= RoleModel(name=roles[2]['name'], code= roles[2]['code'])
# db.session.add_all([role1, role2, role3])
# db.session.commit()
# _permission = {
# 'read': '/auth/read',
# 'commit': '/auth/commit',
# 'write': '/auth/write',
# 'admin': '/admin'
# }
# power1 = PowerModel(url=_permission['read'], code='read')
# power2 = PowerModel(url=_permission['commit'], code='commit')
# power3 = PowerModel(url=_permission['write'], code='write')
# power3 = PowerModel(url=_permission['admin'], code='admin')
# db.session.add_all([power1, power2, power3])
# db.session.commit()
users = UserModel.query.all()
roles = RoleModel.query.all()
powers = PowerModel.query.all()
# print(users)
# print(roles)
# print(powers)
# # 给<1角色>添加<权限>
# roles[0].powers.append(powers[0]) # 读
# roles[1].powers.append(powers[1]) # 提交
# # 给<2角色>添加<权限>
# roles[2].powers.append(powers[0]) # 读
# roles[2].powers.append(powers[1]) # 提交
# roles[2].powers.append(powers[2]) # 写
# # 给<用户>添加<角色>
# users[0].roles.append(roles[0]) # chf -> 普通用户 -> 读, 提交
# users[1].roles.append(roles[1]) # chf -> 管理员 -> 读, 提交, 写
# 用户 --> 角色 --> 角色 --> 权限密码
# db.session.commit()
role = users[0].roles.all()
print(role)
powers = role[0].powers
print(powers)
# # 注册
# @app.route('/register', methods=['GET', 'POST'])
# def register():
# if request.method == 'POST':
# username = request.json.get('username')
# password = request.json.get('passowrd')
# print(username,password)
# user = UserModel()
# user.username = username
# user.password = password
# db.session.add(user)
# user.role.append(RoleModel.query.get(1))
# db.session.commit()
# return {'status': 'success', 'message': '成功注册, 接下来可以去登录', 'next': '/login'}
# return render_template('register.html')
# # 登录
# @app.route('/login', methods=['GET', 'POST'])
# def login():
# if request.method == 'POST':
# username = request.json.get('username')
# password = request.json.get('password')
# user = UserModel.query.filter(UserModel.username=username).first()
# session['_user_id'] = user.id
# session['permission'] = ':'.join([power.code for role in user.role for power in role.power])
# return {'statuc': 'success', 'next': './home', 'message': '登录成功'}
# return render_template('login.html')
# # 退出登录
# @app.route('/logout')
# def logout():
# session['_user_id'] = ''
# session['permission'] = ''
# return redirect('/')
# # home
# @app.route('/home')
# @login_required
# def home():
# return render_template('home.html')
if __name__ == '__main__':
app.run()
启动命令
# 1. 进入到flask的项目文件下面, app.py文件同级, 终端运行项目
flask run
# 2. 终端执行迁移, 和运行指定函数初始化
flask create #相当于执行app.py文件里的create函数
