zoukankan      html  css  js  c++  java
  • 鱼书项目模块化总结

    在线源码: GITHub : https://github.com/whyjust/fisher

    鱼书项目模块化总结

    项目总体思路

    模型类

    视图蓝本

    表单验证数据API

    flask上下文/ajax

    1 异步发送邮件模块

    邮件发送:

    需要进行注册邮件发送或者功能模块需要发送邮件时可以采用:

    from threading import Thread
    
    from app import mail
    from flask import current_app,render_template
    from flask_mail import Message
    
    '''开启异步线程'''
    def send_async_mail(app,msg):
        '''mail.send发送需要获取上下文,因此添加with'''
        with app.app_context():
            try:
                mail.send(msg)
            except Exception as e:
                raise e
    
    def send_mail(to, subject, template,**kwargs):
        '''
        发送邮件
        :param to: 收件人
        :param subject: 标题
        :param template: 渲染模板
        :param kwargs: 关键字参数
        :return:
        '''
        msg = Message('[鱼书]'+ '' +subject,
                      sender=current_app.config['MAIL_USERNAME'],
                      recipients=[to])
        msg.html = render_template(template,**kwargs)
        '''
        current_app是代理对象,开启新的线程时,我们直接获取真实的app核心对象_get_current_object()
        '''
        app = current_app._get_current_object()
        thr = Thread(target=send_async_mail,args=[app,msg])
        thr.start()
    

    2 枚举类标识状态

    状态标识

    状态的变更需要进行标识时(等待,成功,拒绝,撤销)

    from enum import Enum
    
    class PendingStatus(Enum):
        '''
        交易状态: 枚举方法实现
        '''
        Waiting = 1
        Success = 2
        Reject = 3
        Redraw = 4
    
        @classmethod
        def pending_str(cls,status,key):
            key_map = {
                cls.Waiting:{
                    'requester':'等待对方邮寄',
                    'gifter':'等待你邮寄'
                },
                cls.Success: {
                    'requester': '对方已邮寄',
                    'gifter': '你已邮寄,交易完成'
                },
                cls.Reject: {
                    'requester': '对方已拒绝',
                    'gifter': '你已经拒绝'
                },
                cls.Redraw: {
                    'requester': '你已撤销',
                    'gifter': '对方已撤销'
                }
            }
            return key_map[status][key]
    

    3 User模型的方法

    模型定义的方法:

    1 password以hash方式存储

    2 用户信息生成token

    3 password的私有化

    from math import floor
    
    from werkzeug.security import generate_password_hash, check_password_hash
    from flask_login import UserMixin
    from itsdangerous import  TimedJSONWebSignatureSerializer as Serializer
    
    from app.libs.enums import PendingStatus
    from app.models.base import Base, db
    from sqlalchemy import Column, Integer, String, Boolean, Float
    from app import login_manager
    from app.models.drift import Drift
    from app.models.gift import Gift
    from app.models.wish import Wish
    from app.spider.yushu_book import YuShuBook
    from app.libs.helper import is_isbn_or_key
    from flask import current_app
    
    class User(Base, UserMixin):
        '''
        模型属性设置 , UserMixin 记录用户账号的状态
        '''
        id = Column(Integer, primary_key=True)
        nickname = Column(String(24), nullable=False)
        _password = Column('password', String(128), nullable=True)
        phone_number = Column(String(18), unique=True)
        email = Column(String(50), unique=True, nullable=False)
        confirmed = Column(Boolean, default=False)
        beans = Column(Float, default=0)
        send_counter = Column(Integer, default=0)
        receive_counter = Column(Integer, default=0)
        wx_open_id = Column(String(50))
        wx_name = Column(String(32))
    
        '''
        将password方法hash加密只读并将其变为属性访问
        '''
        @property
        def password(self):
            return self._password
    
        @password.setter
        def password(self, raw):
            self._password = generate_password_hash(raw)
    
        def check_password(self, raw):
            return check_password_hash(self._password, raw)
    
        def generate_token(self, expiration=600):
            s = Serializer(current_app.config['SECRET_KEY'],expiration)
            return s.dumps({'id':self.id}).decode('utf-8')
    
        @staticmethod
        def reset_password(token,new_password):
            s = Serializer(current_app.config['SECRET_KEY'])
            try:
                data = s.loads(token.encode('utf-8'))
            except:
                return False
            uid = data.get('id')
            with db.auto_commit():
                user = User.query.get(uid)
                if user:
                    user.password = new_password
            return True
    
        @property
        def summary(self):
            return dict(
                nickname = self.nickname,
                beans = self.beans,
                email = self.email,
                send_receive = str(self.send_counter)+ '/' + str(self.receive_counter)
            )
    
    @login_manager.user_loader
    def get_user(uid):
        '''
        继承UserMixin,进行用户的回调
        :param uid:
        :return:
        '''
        return User.query.get(int(uid))
    
    

    方案二

    from app.extensions import db,login_manager
    from werkzeug.security import generate_password_hash,check_password_hash
    #生成token的模块
    from itsdangerous import TimedJSONWebSignatureSerializer as Seralize
    from flask_login import UserMixin
    from flask import current_app
    from .posts import Posts
    
    class User(UserMixin,db.Model):
        __tablename__ = 'user'
        id = db.Column('id',db.Integer,primary_key=True)
        username = db.Column(db.String(12),index=True)
        password_hash = db.Column(db.String(128))
        sex = db.Column(db.Boolean,default=True)
        age = db.Column(db.Integer)
        email = db.Column(db.String(40))
        icon = db.Column(db.String(70),default='default.jpg')
        #当期账户激活状态
        confirm = db.Column(db.Boolean,default=False)
        #参数1模型名称   参数2反向引用的字段名   参数3 加载方式 提供对象
        posts = db.relationship('Posts',backref='user',lazy='dynamic')
        #secondary在多对多关系中指定关联表的名称
        favorite = db.relationship('Posts',secondary='collections',backref=db.backref('users',lazy='dynamic'),lazy='dynamic')
        #添加使用append   删除使用remove
    
        @property
        def password(self):
            raise ValueError
    
        @password.setter
        def password(self, password):
            self.password_hash = generate_password_hash(password)
    
        #生成token的方法
        def generate_token(self):
            s = Seralize(current_app.config['SECRET_KEY'])
            return s.dumps({'id':self.id})
    
        #检测token的方法
        @staticmethod
        def check_token(token):
            s = Seralize(current_app.config['SECRET_KEY'])
            #从当前的token中拿出字典
            try:
                id = s.loads(token)['id']
            except:
                return False
            #根据用户id取出对应用户的对象
            u = User.query.get(id)
            #判断 当期u对象是否存在
            if not u:
                return False
            #判断当期用户的激活状态 如果没有激活 则激活
            if not u.confirm:
                u.confirm = True
                db.session.add(u)
            return True
        #验证密码
        def check_password_hash(self,password):
            return check_password_hash(self.password_hash,password)
    
        def is_favorite(self,postsId):
            all = self.favorite.all()
            for p in all:
                if p.id==postsId:
                    return True
                #lambda表达式
                if list(filter(lambda p:p.id==int(postsId),all)):
                    return True
                return False
    
    #登录认证的回调,保持数据的一致性
    @login_manager.user_loader
    def user_loader(uid):
        return User.query.get(int(uid))
    

    4 Flask工厂函数管理三方,蓝本

    1 extensions.py负责三方模块的导入与

    2 app.__init__.py 负责create_app初始化

    3 view.__init__.py 负责蓝本的注册

    方案一:

    manage.py

    from app import create_app
    from flask_script import Manager
    from flask_migrate import MigrateCommand
    
    #通过函数create_app进行包括蓝本/扩展/系统配置的初始化
    app = create_app('default')
    manager = Manager(app)
    #给manage添加迁移命令db
    manager.add_command('db',MigrateCommand)
    
    if __name__ == '__main__':
        manager.run()
    
    

    extensions.py

    #extensions.py
    from flask_bootstrap import  Bootstrap
    from flask_sqlalchemy import SQLAlchemy
    from flask_migrate import Migrate
    from flask_mail import Mail
    
    from flask_login import LoginManager
    from flask_uploads import UploadSet,IMAGES,patch_request_class,configure_uploads
    from flask_moment import Moment
    from flask_cache import Cache
    
    #实例化db
    db = SQLAlchemy()
    #实例化bootstrap
    bootstrap = Bootstrap()
    #实例化migrate
    migrate = Migrate(db=db)
    #实例化邮箱
    mail = Mail()
    #实例化用户登录模块
    login_manager = LoginManager()
    #实例化文件对象
    file = UploadSet('photos',IMAGES)
    moment = Moment()
    #simple简单缓存
    # cache = Cache(config={'CACHE_TYPE':'simple'})
    cache = Cache(config={'CACHE_TYPE': 'simple'})
    
    def config_extensions(app):
        #bootstrap初始化app
        bootstrap.init_app(app)
        #db初始化app
        db.init_app(app)
        #migrate初始化app
        migrate.init_app(app=app)
        #mail初始化
        mail.init_app(app)
        #登录模块初始化
        login_manager.init_app(app)
        #moment时间模块初始化
        moment.init_app(app)
        cache.init_app(app=app)
    
        #需要指定登录端点
        login_manager.login_view ='user.login'
        #提示信息
        login_manager.login_message = '请登录再访问'
        #设置session保护级别
        login_manager.session_protection = 'strong'
    
        #配置文件上传
        configure_uploads(app,file)
        patch_request_class(app,size=None)
    
    

    app.__init__.py

    #app.__init__.py
    
    from flask import Flask,render_template
    from app.settings import config
    from app.extensions import config_extensions
    from app.views import config_blueprint
    
    #初始化当前整个应用的函数
    def create_app(config_name):
        app = Flask(__name__)
        #导入settings配置信息
        app.config.from_object(config[config_name])
        #第三方库初始化
        config_extensions(app)
        #注册所有的蓝本函数
        config_blueprint(app)
        #错误页面绑定app
        errors(app)
        return app
    
    def errors(app):
        @app.errorhandler(404)
        def page_not_found(e):
            return render_template('errors/error.html',error=e)
    
        @app.errorhandler(500)
        def page_not_found(e):
            return render_template('errors/error.html', error=e)
    

    view.__init__.py

    #view.__init__.py
    
    from .main import main
    from .user import user
    from .posts import posts
    
    BluePrint = [
        (main,''),
        (user,''),
        (posts,'')
    ]
    
    #封装注册蓝本的函数
    def config_blueprint(app):
        #循环注册蓝本
        for blueprint,prefix in BluePrint:
            app.register_blueprint(blueprint,url_prefix=prefix)
    

    方案二

    fisher.py

    from app import create_app
    
    app = create_app()
    
    if __name__ == '__main__':
        from werkzeug.contrib.fixers import ProxyFix
        app.wsgi_app = ProxyFix(app.wsgi_app)
        app.run()
    

    app.__init__.py

    #app.__init__.py
    
    '''创建flask核心对象'''
    from flask import Flask
    from app.models.base import db
    from flask_login import LoginManager
    from flask_mail import Mail
    from app.libs.limiter import Limiter
    
    
    login_manager = LoginManager()
    mail = Mail()
    limiter = Limiter()
    
    
    def create_app():
        '''
        系统配置与蓝图需要绑定app
        :return:
        '''
        app = Flask(__name__)
        app.config.from_object('app.secure')
        app.config.from_object('app.setting')
        register_blueprint(app)
    
        db.init_app(app)
        login_manager.init_app(app)
        login_manager.login_view = 'web.login'
        login_manager.login_message = '请先登录或者注册'
    
        mail.init_app(app)
        with app.app_context():
            db.create_all()
        return app
    
    #注册蓝本
    def register_blueprint(app):
        from app.web.book import web
        app.register_blueprint(web)
    

    view.__init__.py

    #view.__init__.py
    
    from flask import Blueprint,render_template
    
    '''蓝图 blueprint '''
    web = Blueprint('web',__name__)  #__name__代表蓝图所在模块
    
    
    @web.app_errorhandler(404)
    def not_found(e):
        '''
        AOP: 处理所有的404请求
        '''
        return render_template('404.html'),404
    
    @web.app_errorhandler(500)
    def internal_server_error(e):
        '''
        AOP: 处理所有的500请求
        '''
        return render_template('500.html'),500
    
    #在此处导入代表先初始化在导入应用
    #防止循环调用的问题
    from app.web import book  
    from app.web import auth
    from app.web import drift
    from app.web import gift
    from app.web import main
    from app.web import wish
    

    5 生产开发环境下的settings配置

    settings配置

    1 根据不同等级,配置系统设置

    2 配置数据库的连接

    3 配置不同环境下的生产

    方案一

    import os
    base_path = os.path.abspath(os.path.dirname(__file__))
    #配置所有环境的基类
    class Config:
        SECRET_KEY = 'secret_key'
        SQLALCHEMY_COMMIT_ON_TEARDOWN = True
        SQLALCHEMY_TRACK_MODIFICATIONS = False
    
        MAIL_SERVER = os.environ.get('MAIL_SERVER', 'smtp.163.com')
        MAIL_USERNAME = os.environ.get('MAIL_USERNAME', 'xxx@163.com')
        MAIL_PASSWORD = os.environ.get('MAIL_PASSWORD', 'xxx')
    
        #配置上传文件
        MAX_CONTENT_LENGTH = 1024*1024*64
        UPLOADED_PHOTOS_DEST = os.path.join(base_path,'static/upload')
    
        PAGE_NUM = 3
        
    
    #测试
    class TestingConfig(Config):
        SQLALCHEMY_DATABASE_URI = 'mysql+pymysql://root:xxx@127.0.0.1:3306/blog'
    #开发
    class DevelopmentConfig(Config):
        SQLALCHEMY_DATABASE_URI = 'mysql+pymysql://root:xxx@127.0.0.1:3306/blog'
        # SQLALCHEMY_DATABASE_URI = 'sqlite:///'+ os.path.join(base_path,'develop.sqlite')
    #生产
    class ProductionConfig(Config):
        SQLALCHEMY_DATABASE_URI = 'mysql+pymysql://root:xxx@127.0.0.1:3306/blog'
    
    #设置字典
    config = {
        'development':DevelopmentConfig,
        'production':ProductionConfig,
        'test':TestingConfig,
        'default':DevelopmentConfig
    }
    

    方案二

    settings.py

    '''
    配置文件: 生产环境与开发环境相同的配置,setting可以上传git
    '''
    PER_PAGE = 15
    BEANS_UPLOAD_ONE_BOOK = 0.5
    RECENT_BOOK_COUNT = 20
    

    secure.py

    '''
    配置文件:保存单独的加密信息,secure不要上传git
    '''
    DEBUG = True
    
    # 单数据库
    SQLALCHEMY_DATABASE_URI = 'mysql+cymysql://root:xxx@xxx:3306/fisher'
    SECRET_KEY = 'SAAFsdfsdf:sdadzxcsd,./.dasdafasd'
    
    #Email相关配置
    MAIL_SERVER = 'smtp.163.com'
    MAIL_PORT = 465
    MAIL_USE_SSL = True
    MAIL_USE_TSL = False
    MAIL_USERNAME = 'xxx@163.com'
    MAIL_PASSWORD = 'xxx'
    

    6 分页展示

    def calulate_start(self,page):
            return (page-1)* current_app.config.get('PER_PAGE')
    

    7 模型基类的设计

    模型类的设置

    1 创建基类继承db.Model

    2 db实例化通过继承添加容错自动提交数据库功能

    3 封装了快速设置属性功能

    base.py

    from flask_sqlalchemy import SQLAlchemy as _SQLAlchemy, BaseQuery
    from sqlalchemy import Column, Integer, SmallInteger
    from contextlib import contextmanager
    from datetime import datetime
    
    
    class SQLAlchemy(_SQLAlchemy):
        '''
        封装了数据库的自动提交回滚
        '''
        @contextmanager
        def auto_commit(self):
            try:
                yield
                self.session.commit()
            except Exception as e:
                self.session.rollback()
                raise e
    
    
    class Query(BaseQuery):
        '''
        自定义基类(继承,初始化),重写filter_by方法
        '''
        def filter_by(self, **kwargs):
            if 'status' not in kwargs.keys():
                kwargs['status'] = 1
            return super(Query, self).filter_by(**kwargs)
    
    db = SQLAlchemy(query_class=Query)
    
    
    class Base(db.Model):
        '''
        该模型表不想在数据库创建,添加__abstract__ = True不会创建该表
        '''
        __abstract__ = True
        '''类变量在类开始的时候就已经确定了'''
        create_time = Column('create_time', Integer)
        status = Column(SmallInteger, default=1)
    
        def __init__(self):
            '''实例变量保证创建时间的准确性'''
            self.create_time = int(datetime.now().timestamp())
    
        def delete(self):
            self.status = 0
    
        def set_attrs(self, attrs_dict):
            for key, value in attrs_dict.items():
                if hasattr(self, key) and key != 'id':
                    setattr(self, key, value)
                    
        @property
        def create_datetime(self):
            '''时间格式统一,将方法转换成属性调用'''
            if self.create_time:
                return datetime.fromtimestamp(self.create_time)
            else:
                return None
    
  • 相关阅读:
    2019JAVA第三次实验报告
    第二次作业
    第一周作业
    2019年春总结
    第二周作业
    第十二周作业
    第十一周作业
    第十周作业
    第九周作业
    第八周作业
  • 原文地址:https://www.cnblogs.com/why957/p/9285838.html
Copyright © 2011-2022 走看看