类似库联合查询系统

 源码地址：https://github.com/FFlask/search_db

1、需求分析

N个数据库，依次模糊查询后返回结果，汇总后展示到界面上

数据库是保存在本地数据库中，并在后台界面动态的添加到查询系统中

为了保证安全，加个登陆，但不开放注册，统一在后台界面添加新用户

用户分为管理员和普通用户，管理员可以添加普通用户

单一IP暴力破解登陆的问题使用Nginx+验证码(暂未实现)来解决

2、数据库设计

共需要三张表，用户表和权限表，数据库管理表(用于保存需要查询的数据库)

3、框架选择

由于最近在学习Flask框架，本项目采用Flask框架和其配套模块

前端使用Bootstrap框架

由于功能简单(代码500行内)，就所有代码就写在一个文件上了

4、具体功能实现

4.1 实现说明

已实现功能：登陆，汇总查询，数据表的增删改查，权限控制，普通用户的增删改查

未实现功能：验证码

系统的难点：数据表和本地数据库表的动态关联，使用了元类，通过字符串找类，动态创建变量名

初始化：初始化创建表的时候应将元类创建模型的部分注释掉

4.2 代码实现

# -*- coding: UTF-8 -*-
from flask import Flask,render_template,url_for,redirect,request,flash,session,abort
from flask_sqlalchemy import SQLAlchemy
from flask_wtf import FlaskForm
from wtforms import StringField,PasswordField,SubmitField,SelectField,ValidationError
from wtforms.validators import Email,EqualTo,Length,DataRequired
from werkzeug.security import generate_password_hash,check_password_hash
from flask_login import LoginManager,UserMixin,login_required,login_user,logout_user,current_user
from functools import wraps


app = Flask(__name__)
login_manager = LoginManager()
#设置验证强度，以提供不同的安全等级防止用户会话遭篡改
#设为'strong' 时，Flask-Login 会记录客户端IP地址和浏览器的用户代理信息，如果发现异动就登出用户
login_manager.session_protection = 'strong'
#设置登陆失败后的跳转
login_manager.login_view = 'index'


class Config:
    SECRET_KEY = '1a1@z2'
    SQLALCHEMY_DATABASE_URI = 'mysql+pymysql://root:root@127.0.0.1/world?charset=utf8'

    #SQLALCHEMY_TRACK_MODIFICATIONS = True
    @staticmethod
    def init_app(app):
        pass

app.config.from_object(Config)
db=SQLAlchemy(app)

#数据库模型

@login_manager.user_loader
def load_user(user_id):
    return User.query.get(int(user_id))
login_manager.init_app(app)
#用户表
class User(UserMixin,db.Model):
    __tablename__ = 'user'
    id = db.Column(db.Integer,primary_key=True,index=True)
    username = db.Column(db.String(64),unique=True,index=True)
    password_hash = db.Column(db.String(128))
    role_id = db.Column(db.Integer, db.ForeignKey('role.id'))

    @property
    def password(self):
        raise AttributeError('password unread')

    @password.setter
    def password(self,password):
        self.password_hash = generate_password_hash(password)

    def verify_password(self,password):
        return check_password_hash(self.password_hash,password)

    #新增用户默认是普通用户权限
    def __init__(self,*args,**kwargs):
        super(User,self).__init__(*args,**kwargs)
        if self.role is None:
            if self.username == 'admin':
                self.role = Role.query.filter_by(role_name='Admin').first()
            else:
                self.role = Role.query.filter_by(role_name='User').first()

    def can(self,permissions):
        return self.role is not None and (self.role.permissions & permissions) == permissions

    def is_admin(self):
        return self.can(Permissions.Admin)

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


#权限类
class Permissions:
    Visit = 0x01
    Read = 0x02
    Write = 0x04
    Manager = 0x08
    Admin = 0xFF

#角色权限表
class Role(db.Model):
    __tablename__ = 'role'
    id = db.Column(db.Integer,primary_key=True,index=True)
    role_name = db.Column(db.String(64),unique=True,index=True)
    permissions = db.Column(db.Integer)
    user = db.relationship('User',backref='role')

    @staticmethod
    def init_role():
        roles = {
            'Visitor':Permissions.Visit,
            'User':Permissions.Visit|Permissions.Read|Permissions.Write,
            'Manager':Permissions.Visit|Permissions.Read|Permissions.Write|Permissions.Manager,
            'Admin':Permissions.Admin
        }
        for role_name,role_permissions in roles.items():
            role = Role(role_name=role_name,permissions=role_permissions)
            db.session.add(role)
            db.session.commit()

    def __repr__(self):
        return '<Role %r>'%(self.role_name)



#数据库表
class DBTable(db.Model):
    __tablename__ = 'db_table'
    id = db.Column(db.Integer, primary_key=True, index=True)
    table_name = db.Column(db.String(64), unique=True)
    table_src = db.Column(db.String(64))
    def __repr__(self):
        return 'db_table <%r>'%(self.table_name)



#动态创建数据库模型
table1s = DBTable.query.all()
#把字符串变成变量
createVar = locals()
for table1 in table1s:
    #使用元类动态创建表中存储的表名
    createVar[str(table1.table_name)] = type(str(table1.table_name),(db.Model,),{
        '__tablename__':table1.table_name,
        'id':db.Column(db.Integer, primary_key=True, index=True),
        'username':db.Column(db.String(64), unique=True,index=True),
        'password':db.Column(db.String(64)),
        'email':db.Column(db.String(64), unique=True,index=True),
        })


#表单类
#搜索框
class SearchForm(FlaskForm):
    search = StringField(validators=[DataRequired()],
                         render_kw={'class':'form-control',
                                    'size':'50',
                                    'placeholder':'input search content'})
    kind = SelectField(choices=[
        ('username','username'),('email','email')
    ],render_kw={'class':'form-control'})
    submmit = SubmitField('Search',render_kw={'class':'btn btn-info'})

#登陆框
class LoginForm(FlaskForm):
    username = StringField('Username', validators=[DataRequired(), Length(1, 64)],
                           render_kw={'class':'form-control','placeholder':'enter your username'})
    password = PasswordField('Password', validators=[DataRequired()],
                             render_kw={'class':'form-control','placeholder':'enter your password'})
    verify_code = StringField('CheckCode',
                              render_kw={'class': 'form-control', 'placeholder': 'zzzzzzzzz'})
    submmit = SubmitField('Login',render_kw={'class':'btn btn-info'})

#新增数据库
class DBAddForm(FlaskForm):
    table_name = StringField('table_name', validators=[DataRequired()],
                           render_kw={'class':'form-control','placeholder':'table_name'})
    table_src = StringField('table_src', validators=[DataRequired()],
                              render_kw={'class': 'form-control', 'placeholder': 'table_src'})
    submmit = SubmitField('Add',render_kw={'class':'btn btn-info'})

#编辑数据库
class DBEditForm(FlaskForm):
    table_name = StringField('table_name', validators=[DataRequired()],
                           render_kw={'class':'form-control','placeholder':'table_name'})
    table_src = StringField('table_src', validators=[DataRequired()],
                              render_kw={'class': 'form-control', 'placeholder': 'table_src'})
    submmit = SubmitField('Edit',render_kw={'class':'btn btn-info'})

#新增用户
class UserAddForm(FlaskForm):
    username = StringField('username', validators=[DataRequired()],
                           render_kw={'class':'form-control','placeholder':'username'})
    password = StringField('password', validators=[DataRequired()],
                              render_kw={'class': 'form-control', 'placeholder': 'password'})
    submmit = SubmitField('Add',render_kw={'class':'btn btn-info'})

    def validate_username(self,field):
        if User.query.filter_by(username = field.data).first():
            raise ValidationError('username is repeat')

#修改用户信息
class UserEditForm(FlaskForm):
    username = StringField('username', validators=[DataRequired()],
                           render_kw={'class':'form-control','readonly':'readonly'})
    new_password = StringField('new_password',
                              render_kw={'class': 'form-control', 'placeholder': 'new password'})
    role = SelectField(coerce=int,render_kw={'class':'form-control'})
    submmit = SubmitField('Edit',render_kw={'class':'btn btn-info'})
    def __init__(self,*args,**kwargs):
        super(UserEditForm,self).__init__(*args,**kwargs)
        self.role.choices = [(role.id,role.role_name) for role in Role.query.order_by(Role.role_name).all()
                             if role.role_name=='Admin' or role.role_name == 'User']

#权限控制装饰器
def permission_required(permission):
    def decorator(func):
        @wraps(func)
        def decorated_function(*args, **kwargs):
            if not current_user.can(permission):
                abort(403)
            return func(*args, **kwargs)
        return decorated_function
    return decorator
#具有admin权限
def is_admin(func):
    return permission_required(Permissions.Admin)(func)



#业务逻辑
#初次运行时运行一下，创建admin账户
@app.route('/create_user')
def create_user():
    user1 = User.query.filter_by(username = 'admin').first()
    if not user1:
        user2 = User(username='admin',password='admin')
        db.session.add(user2)
        db.session.commit()
    return 'ok'

#测试
@app.route('/test')
def test():
    return 'test'

#登陆页面
@app.route('/',methods=['GET','POST'])
def index():
    login_form = LoginForm()
    if current_user.is_authenticated:
        print(current_user.username)
    if login_form.validate_on_submit():
        user1 = User.query.filter_by(username = login_form.username.data).first()
        if user1 and user1.verify_password(login_form.password.data):
            login_user(user1)
            return redirect(request.args.get('next') or url_for('search_db'))
        else:
            flash('username or password error')
            return redirect(url_for('index'))
    return render_template('index.html',login_form=login_form)

@app.route('/logout')
@login_required
def logout():
    logout_user()
    return redirect(url_for('index'))

#搜索页面
@app.route('/search_db',methods=['GET','POST'])
@login_required
def search_db():
    search_form = SearchForm()
    ip = request.remote_addr
    results = {}
    if search_form.validate_on_submit():
        search_item = search_form.kind.data
        for index,table1 in enumerate(DBTable.query.all()):
            try:
                #动态创建变量名
                createVar[str(table1.table_name)+str(index)] =
                    globals()[table1.table_name].query.filter(getattr(globals()[table1.table_name],search_item)
                                                              .like('%'+'%s'%search_form.search.data+'%')).all()
                if createVar[str(table1.table_name)+str(index)] is not None:
                    results.update({table1.table_src: createVar[str(table1.table_name)+str(index)]})
            except:
                flash('Your db is not have table <%s>,Please update it!'%(str(table1.table_name)))
    return render_template('search_db.html',search_form=search_form,results=results,ip=ip)

#数据库管理/查
@app.route('/admin_db/query')
@login_required
@is_admin
def admin_db_query():
    return render_template('admin_db_query.html', table1s = DBTable.query.all())

#数据库管理/增
@app.route('/admin_db/add',methods=['POST','GET'])
@login_required
@is_admin
def admin_db_add():
    db_add_form = DBAddForm()
    if db_add_form.validate_on_submit():
        new_table = DBTable(
            table_name=db_add_form.table_name.data,
            table_src=db_add_form.table_src.data,
        )
        db.session.add(new_table)
        db.session.commit()
        return redirect(url_for('admin_db_query'))
    return render_template('admin_db_add.html',db_add_form=db_add_form)

#数据库管理/改
@app.route('/admin_db/edit',methods=['POST','GET'])
@login_required
@is_admin
def admin_db_edit():
    edit_id = request.args.get('edit_id')
    db_table1 = DBTable.query.filter_by(id=edit_id).first()
    db_edit_form = DBEditForm()
    if db_edit_form.validate_on_submit():
        db_table1.table_name = db_edit_form.table_name.data
        db_table1.table_src = db_edit_form.table_src.data
        db.session.add(db_table1)
        db.session.commit()
        return redirect(url_for('admin_db_query'))
    db_edit_form.table_name.data = db_table1.table_name
    db_edit_form.table_src.data = db_table1.table_src
    return render_template('admin_db_edit.html',db_edit_form=db_edit_form,db_table1=db_table1)

#数据库管理/删
@app.route('/admin_db/delete')
@login_required
@is_admin
def admin_db_delete():
    delete_id = request.args.get('delete_id')
    db_table1 = DBTable.query.filter_by(id=delete_id).first()
    db.session.delete(db_table1)
    db.session.commit()
    return redirect(url_for('admin_db_query'))


#用户管理/查
@app.route('/admin_user/query',methods=['POST','GET'])
@login_required
@is_admin
def admin_user_query():
    return render_template('admin_user_query.html',table1s=User.query.all())

#用户管理/增
@app.route('/admin_user/add',methods=['POST','GET'])
@login_required
@is_admin
def admin_user_add():
    user_add_form = UserAddForm()
    if user_add_form.validate_on_submit():
        new_user = User(
            username=user_add_form.username.data,
            password=user_add_form.password.data
        )
        db.session.add(new_user)
        db.session.commit()
        return redirect(url_for('admin_user_query'))
    return render_template('admin_user_add.html',user_add_form=user_add_form)

#用户管理/改
@app.route('/admin_user/edit',methods=['POST','GET'])
@login_required
@is_admin
def admin_user_edit():
    edit_id = request.args.get('edit_id')
    user1 = User.query.filter_by(id=edit_id).first()
    user_edit_form = UserEditForm()
    if user_edit_form.validate_on_submit():
        user1.username = user_edit_form.username.data
        user1.role_id = user_edit_form.role.data
        if user_edit_form.new_password.data:
            user1.password = user_edit_form.new_password.data
        db.session.add(user1)
        db.session.commit()
        return redirect(url_for('admin_user_query'))
    user_edit_form.username.data = user1.username
    user_edit_form.role.data = user1.role_id
    return render_template('admin_user_edit.html',user_edit_form=user_edit_form,user1=user1)

#用户管理/删
@app.route('/admin_user/delete',methods=['POST','GET'])
@login_required
@is_admin
def admin_user_delete():
    delete_id = request.args.get('delete_id')
    user1 = User.query.filter_by(id=delete_id).first()
    db.session.delete(user1)
    db.session.commit()
    return redirect(url_for('admin_user_query'))


if __name__ == '__main__':
    app.run(host="0.0.0.0", port=5000)

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <title>首页</title>
        <link rel="stylesheet" href="https://cdn.staticfile.org/twitter-bootstrap/3.3.7/css/bootstrap.min.css">
    <script src="https://cdn.staticfile.org/jquery/2.1.1/jquery.min.js"></script>
    <script src="https://cdn.staticfile.org/twitter-bootstrap/3.3.7/js/bootstrap.min.js"></script>
</head>
<body>
<div class="col-sm-6 col-sm-offset-3">
    <h2>登陆</h2><br>
    <form method="POST" action="{{ url_for('index') }}" >
        {{ login_form.hidden_tag() }}
        {{ login_form.username }}<br>
        {{ login_form.password }}<br>
        {{ login_form.submmit }}
    </form>
    <div>
    {% for msg in get_flashed_messages() %}
        {{ msg }}
    {% endfor %}
    </div>
</div>

</body>
</html>

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <title>搜索</title>
    <link rel="stylesheet" href="https://cdn.staticfile.org/twitter-bootstrap/3.3.7/css/bootstrap.min.css">
    <script src="https://cdn.staticfile.org/jquery/2.1.1/jquery.min.js"></script>
    <script src="https://cdn.staticfile.org/twitter-bootstrap/3.3.7/js/bootstrap.min.js"></script>
</head>
<body>
<div class="col-sm-6 col-sm-offset-3">
<h2>查询，查询，查询！</h2>
<form method="POST" action="{{ url_for('search_db') }}" class="form-inline" >
    {{ search_form.hidden_tag() }}
    {{ search_form.search }}
    {{ search_form.kind }}
    {{ search_form.submmit }}
</form>

<br>
{% for msg in get_flashed_messages() %}
    <strong style="color: red">{{ msg }}</strong>
{% endfor %}
<hr>
<table border="1" class="table table-bordered">
    <thead>
        <tr class="info">
            <td><strong>来源</strong></td>
            <td><strong>用户</strong></td>
            <td><strong>密码</strong></td>
            <td><strong>邮箱</strong></td>
        </tr>
    </thead>
    {% for result_key,result_value in results.items() %}
        {% for result in result_value %}
        <tr>
            <td>{{ result_key }}</td>
            <td>{{ result.username }}</td>
            <td>{{ result.password }}</td>
            <td>{{ result.email }}</td>
        </tr>
        {% endfor %}
    {% endfor %}
</table>
</div>
<div class="col-sm-2">
    <a href="{{ url_for('index') }}">logout</a>&nbsp;&nbsp;
    {% if current_user.username == 'admin' %}
    <a href="{{ url_for('admin_db_query') }}">后台</a>
    {% endif %}
    <br>欢迎{{ current_user.username }}<br>
    你的IP是:<{{ ip }}><br>
    不要乱搞呦！
</div>
</body>
</html>

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <title>后台管理</title>
    <link rel="stylesheet" href="https://cdn.staticfile.org/twitter-bootstrap/3.3.7/css/bootstrap.min.css">
    <script src="https://cdn.staticfile.org/jquery/2.1.1/jquery.min.js"></script>
    <script src="https://cdn.staticfile.org/twitter-bootstrap/3.3.7/js/bootstrap.min.js"></script>
</head>
<body>
<div class="col-sm-6 col-sm-offset-3">
<ul class="nav nav-tabs nav-justified">
 <li role="presentation" class="active"><a href="#">数据库管理</a></li>
  <li role="presentation"><a href="{{ url_for('admin_user_query') }}">用户管理</a></li>
  <li role="presentation"><a href="{{ url_for('search_db') }}">返回查询</a></li>
</ul>
<ul class="nav nav-pills nav-justified">
  <li role="presentation" ><a href="{{ url_for('admin_db_query') }}">查</a></li>
  <li role="presentation" class="active"><a href="{{ url_for('admin_db_add') }}">增</a></li>
  <li role="presentation"><a href="#">改</a></li>
    <li role="presentation"><a href="#">删</a></li>
</ul>
<hr>
    <div class="jumbotron">
            <h4>新增数据库说明</h4>
            <ul>
                <li>本地数据库名必须为search_db</li>
                <li>新增表名必须与数据库中表名完全一致</li>
                <li>列名至少要包含username,password,email</li>
            </ul>
        </div>
    <div class="col-sm-10 col-sm-offset-1">
        <form method="POST" action="{{ url_for('admin_db_add') }}" >
        {{ db_add_form.hidden_tag() }}
        {{ db_add_form.table_name }}<br>
        {{ db_add_form.table_src }}<br>
        {{ db_add_form.submmit }}
    </form>
    </div>
</div>
</body>
</html>

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <title>后台管理</title>
    <link rel="stylesheet" href="https://cdn.staticfile.org/twitter-bootstrap/3.3.7/css/bootstrap.min.css">
    <script src="https://cdn.staticfile.org/jquery/2.1.1/jquery.min.js"></script>
    <script src="https://cdn.staticfile.org/twitter-bootstrap/3.3.7/js/bootstrap.min.js"></script>
</head>
<body>
<div class="col-sm-6 col-sm-offset-3">
<ul class="nav nav-tabs nav-justified">
 <li role="presentation" class="active"><a href="#">数据库管理</a></li>
  <li role="presentation"><a href="#">用户管理</a></li>
  <li role="presentation"><a href="{{ url_for('search_db') }}">返回查询</a></li>
</ul>
<ul class="nav nav-pills nav-justified">
  <li role="presentation" ><a href="{{ url_for('admin_db_query') }}">查</a></li>
  <li role="presentation" ><a href="{{ url_for('admin_db_add') }}">增</a></li>
  <li role="presentation" class="active"><a href="#">改</a></li>
    <li role="presentation"><a href="#">删</a></li>
</ul>
<hr>
    <div class="jumbotron">
            <h4>修改数据库说明</h4>
            <ul>
                <li>本地数据库名必须为search_db</li>
                <li>新增表名必须与数据库中表名完全一致</li>
                <li>列名至少要包含username,password,email</li>
            </ul>
        </div>
    <div class="col-sm-10 col-sm-offset-1">
        <form method="POST" action="{{ url_for('admin_db_edit',edit_id = db_table1.id) }}" >
        {{ db_edit_form.hidden_tag() }}
        {{ db_edit_form.table_name }}<br>
        {{ db_edit_form.table_src }}<br>
        {{ db_edit_form.submmit }}
    </form>
    </div>
</div>
</body>
</html>

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <title>后台管理</title>
    <link rel="stylesheet" href="https://cdn.staticfile.org/twitter-bootstrap/3.3.7/css/bootstrap.min.css">
    <script src="https://cdn.staticfile.org/jquery/2.1.1/jquery.min.js"></script>
    <script src="https://cdn.staticfile.org/twitter-bootstrap/3.3.7/js/bootstrap.min.js"></script>
</head>
<body>
<div class="col-sm-6 col-sm-offset-3">
<ul class="nav nav-tabs nav-justified">
 <li role="presentation" class="active"><a href="#">数据库管理</a></li>
  <li role="presentation"><a href="{{ url_for('admin_user_query') }}">用户管理</a></li>
  <li role="presentation"><a href="{{ url_for('search_db') }}">返回查询</a></li>
</ul>
<ul class="nav nav-pills nav-justified">
  <li role="presentation" class="active"><a href="{{ url_for('admin_db_query') }}">查</a></li>
  <li role="presentation"><a href="{{ url_for('admin_db_add') }}">增</a></li>
  <li role="presentation"><a href="#">改</a></li>
    <li role="presentation"><a href="#">删</a></li>
</ul>
<hr>
    <div class="col-sm-10 col-sm-offset-1">
        <table border="1" class="table table-bordered">
            <thead>
                <tr class="info">
                    <td>id</td>
                    <td>表名</td>
                    <td>来源</td>
                    <td>操作</td>
                </tr>
            </thead>
            {% for table in table1s %}
                <tr>
                    <td>{{ table.id }}</td>
                    <td>{{ table.table_name }}</td>
                    <td>{{ table.table_src }}</td>
                    <td><a href="{{ url_for('admin_db_edit',edit_id = table.id) }}">修改</a>|<a href="{{ url_for('admin_db_delete',delete_id = table.id) }}">删除</a></td>
                </tr>
            {% endfor %}
        </table>
    </div>
</div>
</body>
</html>

4.3 界面展示

5、经验总结

所有后端代码都写一个文件还是有点乱，下次还是分开

用户登陆，权限控制应当制作成一个模块，以后每次复制粘贴就行了