Flask实践python进阶Python

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

2018-08-10  本文已影响18人  我的昵称很霸气

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

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

用户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

注册用户
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

# 在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 部分

用户注册
# 在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

现在把登录获取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
    })

上一篇下一篇

猜你喜欢

热点阅读