Flask項目初始化

數據庫實現命令初始化

1.實現命令主腳本

# coding=utf-8
from functools import wraps
from getpass import getpass
import sys
import os

#執行到這里的時候commands是為空對象
commands = {}


# 工具函數
def check_input_password(password):
    if len(password.strip()) < 6:
        return check_input_password(getpass('密碼長度必須大於6位，請重新輸入管理員賬戶密碼：'))
    password2 = getpass('再次輸入：')
    if password == password2.strip():
        return password
    else:
        return check_input_password(getpass('兩次輸入密碼不一致，請重新輸入管理員賬戶密碼：'))


def check_input_username(username):
    if username.strip():
        return username
    else:
        return check_input_username(input('請輸入管理員賬戶登錄名：'))


# 注冊命令
def registry_command(cmd_str):
    def decorate(func):
        commands[cmd_str] = func

        @wraps(func)
        def wrapper(*args, **kwargs):
            return func(*args, **kwargs)
        return wrapper
    return decorate


@registry_command('init_db')
#到這里的時候commands里面添加一個元素
#commands["init_db"]=init_db
def init_db():
    from public import db
    from config import BASE_DIR
    import apps.account.models
    import apps.configuration.models
    import apps.deploy.models
    import apps.assets.models
    import apps.schedule.models
    import apps.setting.models

    user_input = input('是否要初始化數據庫，該操作會清空所有數據[y|n]？')
    if user_input.strip() == 'y':
        db.drop_all()
        db.create_all()
        with open(os.path.join(BASE_DIR, 'libs', 'sql', 'permissions.sql'), encoding='utf-8') as f:
            line = f.readline()
            while line:
                if line.startswith('INSERT INTO'):
                    db.engine.execute(line.strip())
                line = f.readline()
        print('數據庫已初始化成功！')
        user_input = input('是否需要創建管理員賬戶[y|n]？')
        if user_input.strip() == 'y':
            create_admin()


@registry_command('create_admin')
def create_admin():
    from apps.account.models import User

    admin = User.query.filter_by(is_supper=True).first()
    if admin:
        user_input = input('已存在管理員賬戶 <%s>，需要重置密碼[y|n]？' % admin.username)
        if user_input.strip() == 'y':
            password = check_input_password(getpass('請輸入新的管理員賬戶密碼：'))
            admin.password = password
            admin.token_expired = 0
            admin.save()
            print('重置管理員密碼成功！')
    else:
        username = check_input_username(input('請輸入管理員賬戶登錄名：'))
        password = check_input_password(getpass('請輸入管理員賬戶密碼：'))
        User(username=username, password=password, nickname='管理員', is_supper=True).save()
        print('創建管理員賬戶成功！')


@registry_command('enable_admin')
def enable_admin():
    from apps.account.models import User

    admin = User.query.filter_by(is_supper=True).first()
    admin.update(is_active=True)
    print('管理員賬戶狀態已修改為啟用！')


def print_usage():
    print('''
usage: %s <command>

command:
    init_db         初始化數據庫
    create_admin    創建管理員賬戶
    enable_admin    啟用管理員賬戶，用於登錄失敗次數過多賬戶被禁用時使用
    ''' % sys.argv[0])


if __name__ == '__main__':
    #裝飾器的作用是在把此模塊加載到內存中的時候先
    #把registry_command函數執行了也就是初始化了commands
    if len(sys.argv) == 1:
        print_usage()
        sys.exit(1)
    cmd = sys.argv.pop(0)  #manage.py
    arg1 = sys.argv.pop(0) #init_db
    r_func = commands.get(arg1)
    if callable(r_func):
        r_func(*sys.argv)
    else:
        print('遇到了不可能會出現的錯誤！')

INSERT INTO account_permissions (id, name, `desc`) VALUES (100, 'home_view', 'Dashboard');

-- 用戶管理 -> 用戶列表
INSERT INTO account_permissions (id, name, `desc`) VALUES (101, 'account_user_view', '獲取用戶列表');
INSERT INTO account_permissions (id, name, `desc`) VALUES (102, 'account_user_add', '添加用戶');
INSERT INTO account_permissions (id, name, `desc`) VALUES (103, 'account_user_edit', '編輯用戶');
INSERT INTO account_permissions (id, name, `desc`) VALUES (104, 'account_user_del', '刪除用戶');
INSERT INTO account_permissions (id, name, `desc`) VALUES (105, 'account_user_disable', '禁用用戶');

2.統一配置信息模塊

from pytz import timezone
import os

DEBUG = True
TIME_ZONE = timezone('Asia/Shanghai')
BASE_DIR = os.path.dirname(os.path.abspath(__file__))
#SQLALCHEMY_DATABASE_URI = 'mysql+pymysql://root:root@127.0.0.1/testdb'
SQLALCHEMY_DATABASE_URI = 'sqlite:///' + os.path.join(BASE_DIR, 'test.db')
SQLALCHEMY_TRACK_MODIFICATIONS = False
SQLALCHEMY_ECHO = False

DOCKER_REGISTRY_SERVER = 'localhost:5000'
DOCKER_REGISTRY_AUTH = {'username': 'user', 'password': 'password'}

3.創建數據庫連接實例模塊

from flask import Flask
from flask_sqlalchemy import SQLAlchemy
import config


app = Flask(__name__)

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

4.實現效果

 

Flask實現api接口調用鑒權

   1.定義一個總體裝飾器

def require_permission(str_code):
    def decorate(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            if not g.user.is_supper:
                or_list = [x.strip() for x in str_code.split('|')]
                # print(or_list)     
                #["account_user_view","account_role_view"]
                #接口要求的權限集合
                for or_item in or_list:
                    and_set = {x.strip() for x in or_item.split('&')}
                    #把列表中單項元素放到一個集合中去
                    #or_item.split('&') 這里的作用是防止把單詞給打散了
                    #split('&') 這里面只需要指定一個單詞中不存在的符號就可以了
                    #{"account_user_view"}
                    and_set = {x.strip() for x in or_item}
                    #{’a','c','c','o','u','n','t'...,'w'}
                    # print(and_set)  #{"account_user_view"}
                    if and_set.issubset(g.user.permissions):
                        break
                else:
                    return json_response(message='Permission denied'), 403
                #當迭代的對象迭代完並為空時,位於else的子句將執行
                #而如果在for循環中含有break時則直接終止循環,並不會執行else子句
            return func(*args, **kwargs)

        return wrapper

    return decorate

  2.定義處理對應url的函數

@blueprint.route('/<int:u_id>', methods=['PUT'])
@require_permission('account_user_edit | account_user_disable')
def put(u_id):
    form, error = JsonParser('nickname', 'is_active',
                             Argument('role_id', type=int, required=False, help='請選擇角色'),
                             Argument('email', nullable=True),
                             Argument('password', nullable=False, required=False),
                             Argument('mobile', nullable=True)).parse()

    if error is None:
        u_info = User.query.get_or_404(u_id)
        if form.password:
            u_info.password = form.password
        if not u_info.update(**form) and form.password:
            u_info.save()
        return json_response(u_info)
    return json_response(message=error)

  3.定義無需鑒權的url

def init_app(app):
    excel.init_excel(app)
    app.before_request(cross_domain_access_before)
    app.before_request(auth_middleware)




def auth_middleware():
    if request.path == '/api/account/users/login/' or request.path.startswith('/api/apis/configs/') \
            or request.path.startswith('/api/apis/files/') or "static" in request.path or "favicon.ico" in request.path or "/" == request.path or "/login" == request.path:
        return None
    token = request.headers.get('X-TOKEN')
    if token and len(token) == 32:
        g.user = User.query.filter_by(access_token=token).first()
        if g.user and g.user.is_active and g.user.token_expired >= time.time():
            g.user.token_expired = time.time() + 8 * 60 * 60
            g.user.save()
            return None
    return json_response(message='Auth fail, please login'), 401

 

Flask藍圖url配置

   1.app的url配置 無需經過藍圖  直接在app對象上進行設置

from flask import Flask
from flask_sqlalchemy import SQLAlchemy
import config


app = Flask(__name__)

@app.route("/")
def index():
    return app.send_static_file('index.html')


@app.route("/login", methods=['GET'])
def index_login():
    return app.send_static_file('index.html')

   2.通過藍圖配置接口的url地址

       1.在每個藍圖app目錄下的__init__.py中配置一個url訪問前綴

from apps.account import user
from apps.account import role


def register_blueprint(app):
    app.register_blueprint(user.blueprint, url_prefix='/api/account/users')
    app.register_blueprint(role.blueprint, url_prefix='/api/account/roles')

       2.在view處理函數中指定最后的url地址   所以藍圖下的每個處理函數對應的url地址由這兩個部分拼接組成

@blueprint.route('/setting/password', methods=['POST'])
def setting_password():
    form, error = JsonParser(
        Argument('password', help='請輸入原密碼'),
        Argument('newpassword', help='請輸入新密碼')
    ).parse()
    if error is None:
        if g.user.verify_password(form.password):
            g.user.password = form.newpassword
            g.user.save()
        else:
            return json_response(message='原密碼錯誤')
    return json_response(message=error)

      3.拼接后的url地址為http://127.0.0.1:3000/api/account/users/setting/info

         1./api/account/users  為url_prefix='/api/account/users'

         2./setting/info             為@blueprint.route('/setting/info')