魚書項目模塊化總結
阿新 • • 發佈:2018-07-09
pen open spider reject blue all return 重寫 except
魚書項目模塊化總結
項目總體思路
模型類
視圖藍本
表單驗證數據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
魚書項目模塊化總結