魚書項目模塊化總結

項目總體思路

模型類

視圖藍本

表單驗證數據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‘, ‘[email protected]‘)
    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:[email protected]:3306/blog‘
#開發
class DevelopmentConfig(Config):
    SQLALCHEMY_DATABASE_URI = ‘mysql+pymysql://root:[email protected]:3306/blog‘
    # SQLALCHEMY_DATABASE_URI = ‘sqlite:///‘+ os.path.join(base_path,‘develop.sqlite‘)
#生產
class ProductionConfig(Config):
    SQLALCHEMY_DATABASE_URI = ‘mysql+pymysql://root:[email protected]: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 = ‘[email protected]‘
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

魚書項目模塊化總結