Flask+Vue 构建个人博客01 ---用户模块

• 分享一个前后端分离(Python3 + Flask +Vuew)的个人博客，一个使用Python Flask 框架编写的个人博客。项目很轻。却很实用。
• GitHub地址：哆啦瞎梦
• 技术栈：Flask、Vue、Docker
• 初始化，项目结构我就不再写了，之前写过了，没有的可以看我之前的文章。或者直接私信我，我特地准备了一个干净的模板

---

无论什么项目用户这一块肯定是先要完成的

原本打算用户这一块我用第三方登录但是考虑到这是一个很简单的博客，就没必要写那么多。又没人会来看你博客，只是写着玩玩啊

用户model

• 分为系统用户和注册用户。偷懒了下没有归类。

系统用户(也就是管理员)

# biz->orm->user.py
from datetime import datetime, timedelta
from uuid import uuid4
from sqlalchemy import *
from werkzeug.security import generate_password_hash, check_password_hash
from common.error import UnauthorizedError
from flask_app.orm.base import BaseModel

class AdminUser(BaseModel):
    __tablename__ = 'admin_user'

    id = Column(u'id', VARCHAR(length=50), primary_key=True, nullable=False)
    name = Column(u'name', VARCHAR(length=50), nullable=False)
    status = Column(u'status', Integer, default=0)
    create_at = Column(u'create_at', DateTime, default=datetime.now)
    modify_at = Column(u'modify_at', DATETIME, default=datetime.now)
    password = Column(u'password', VARCHAR(length=50), nullable=False)

    def _before_insert(self):
        self.id = uuid4().hex

• 基础的model编写，就不说了。这里出现一个BaseModel这个这里解释下，原本的是db.Model。这个BaseModel是对它的封装，这里封装了一些好用的方法。可以先这样写着。再用到的时候我在跳回来解释下。现在让我解释。我不清楚从哪里入手
• error 这个是错误处理机制。项目中的一些错误都是这里处理的。同样用到我就直接跳过来解释

---

注册用户

class User(BaseModel):
    __tablename__ = 'user'

    id = Column(u'id', VARCHAR(length=50), primary_key=True, nullable=False)
    email = Column(u'email', VARCHAR(length=50), nullable=False)
    _password = Column(u'password', VARCHAR(length=256), nullable=False)
    status = Column(u'status', Integer, default=0)
    create_at = Column(u'create_at', DateTime, default=datetime.now)
    modify_at = Column(u'modify_at', DATETIME, default=datetime.now)

    @property
    def password(self):
        return self._password

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

    def check_password(self, raw):
        if not self._password:
            return False
        return check_password_hash(self._password, raw)

    @staticmethod
    def verify(email, password):
        user = User.query.filter_by(email=email).first_or_404()
        if not user.check_password(password):
            raise UnauthorizedError(msg='授权失败，密码核对失败！')
        scope = ''
        # login_user(user)
        return {'uid': user.id, 'scope': scope}

    def _before_insert(self):
        self.id = uuid4().hex

• property 在这里是为了让在调用的时候，直接把password当做一个方法。在model传值存入的时候，遍历到password 他会执行password这个函数。这个是另一篇文章，用来解释python 中的property
  到此。model编写完成了。如果现在运行肯定会报错的。Basemodel没有扩展。现在。我把BaseModel中部分代码贴上，

---

• BaseModel

# 在flask_app->orm->base.py
from datetime import datetime
from decimal import Decimal

import six
from aniso8601 import date
from flask_sqlalchemy import SQLAlchemy as _SQLAlchemy, BaseQuery
from sqlalchemy import inspect, Column, Integer, SmallInteger, orm
from contextlib import contextmanager

from common.error import NotFoundError
from flask_app.orm import transfer

db = SQLAlchemy(session_options={'autocommit': True}, query_class=Query)
class BaseModel(db.Model):
    __abstract__ = True

    def insert(self):
        self._before_insert()
        db.session.add(self)
        db.session.flush()
        self._after_insert()
        return self

    def update(self):
        self._before_update()
        db.session.merge(self)
        db.session.flush()
        self._after_update()
        return self

    def delete(self):
        self._before_delete()
        db.session.delete(self)
        db.session.flush()
        self._after_delete()

    def _before_insert(self):
        pass

    def _after_insert(self):
        pass

    def _before_update(self):
        pass

    def _after_update(self):
        pass

    def _before_delete(self):
        pass

    def _after_delete(self):
        pass

    def save(self):
        if self.id:
            self.update()
        else:
            self.insert()

这些封装的在引用model的时候你要把更新啊什么的直接xxx.insert等就可以了

view 部分

• 用户这块主要为注册和登录
• 介绍几个封装的方法(大家需要过去看看，要不下面代码可能会看不懂，我不喜欢把所有的东西写到一篇文章中，每一个功能都有一个地方待着)
  1.BaseModelHandler :BaseModelHandler代码
  2.common_response :common_response 代码

用户注册

# 在v1中创建user.py
from flask import session
from itsdangerous import TimedJSONWebSignatureSerializer as Serializer

from apps.v1.validators.user import UserForm, UserLoginForm
from biz.orm.user import User
from common.error import NotFoundError
from common.globalconfig import g_config
from flask_app.restful_api_manage.base import BaseModelHandler
from flask_app.restful_api_manage.hepler import common_response, error_response_default


class UserHandle(BaseModelHandler):
    def post(self):
        form = UserForm().validate_for_api(User)
        user_obj = User.from_dict(form.data)
        user_obj.insert()
        user_dict = user_obj.to_dict()
        return common_response(self, 201, user_dict)
# 这样一个简单的注册就做好了

通过postman去测试API是否正确

image.png image.png

• 一个简单的注册api就写出来了

现在把登录获取token写上

class UserLoginHandle(BaseModelHandler):
    def post(self):
        form = UserLoginForm().validate_for_api(User)
        config = g_config
        # 判断用户是否存在
        user = User.query.filter_by(email=form.email.data).first()
        if not user:
            raise NotFoundError("用户不存在")

        user_ref = user.verify(form.email.data, form.password.data)
        expiration = config['TOKEN_KEY'].get('token_expiration')
        token = generate_auth_token(user_ref['uid'], expiration)
        # session['user'] = user.id
        t = {
            'token': token.decode('ascii')
        }
        resp = common_response(self, 200, t)
        return resp


def generate_auth_token(uid, expiration=None):
    """生成令牌"""

    config = g_config
    s = Serializer('{}'.format(config['TOKEN_KEY'].get('token_key')), expires_in=expiration)
    return s.dumps({
        'uid': uid
    })