发布时间:2019/02/09 22:40:49   更新时间:2020/07/31 19:52:00
from flask import Flask
from Config import config
from Config import datas
from flask_uploads import configure_uploads
from flask_uploads import patch_request_class
from flask_socketio import SocketIO
from flask_sqlalchemy import SQLAlchemy
from flask_login import LoginManager
login_manager = LoginManager()
#session_protection 属性可以设为 None、 'basic' 或 'strong',以提供不同的安全等级防止用户会话遭篡改。 设为 'strong' 时, Flask-Login 会记录客户端 IP地址和浏览器的用户代理信息, 如果发现异动就登出用户。
login_manager.session_protection = "strong"
#login_view 属性设置登录页面的端点
login_manager.login_view = "page.login"#数据库的ORM。db = SQLAlchemy(use_native_unicode="utf8")#socketio。
socketio = SocketIO()
#工厂函数
def creat_app(config_name):
#static_url_path设置以static为root目录。这个很重要,前端知道了根目录是哪个。
app = Flask(__name__, template_folder = "./templates/", static_url_path = "")
app.config.from_object(config[config_name])
config[config_name].init_app(app)
#注册文件上传(安全起见,先不打开)
#configure_uploads(app, datas)
#限制上传文件大小为默认的16M
#patch_request_class(app)
#注意,app工厂函数内部,app还没有产生,使用app.route或者app.errorhandler是不行的。
#这种情况,就要用蓝图注册的方法。这种方法就针对用函数产生app的场景。
#注册page的蓝图
from page import page_bp
app.register_blueprint(page_bp)
#注册api的蓝图
from api import api_bp
app.register_blueprint(api_bp)
#进行必要的组件初始化
login_manager.init_app(app)
db.init_app(app)
socketio.init_app(app)
#注册socketio
from Socketio import handler_recv_json
socketio.on_event("message", handler_recv_json)
#进行必要的初始化,比如数据库的初始化。
from Init import db_init
db_init()
#app产生。供Mangage.py调用。
return app
如下提供一个典型的装饰函数,用来打印每个request的一些信息。
class Decorator(object):
def __init__(self):
return
@staticmethod
def request_pretreat(f):
@wraps(f)
def decorated_function(*args, **kwargs):
#打印request信息
log_app.logger.critical("The request has:\n\
url : %r\n\
method : %r\n\
form : %r\n\
files : %r\n\
args : %r\n."
% (pprint.pformat(request.url),
pprint.pformat(request.method),
pprint.pformat(request.form),
pprint.pformat(request.files),
pprint.pformat(request.args)))
return f(*args, **kwargs)
return decorated_function
本文件提供一些公共的函数使用。
class Common(object):
def __init__(self):
return
'''
在app起来或者没起来两种情况下,获取app。
'''
@staticmethod
def get_app():
try:
print("The current_app.name is %s." % current_app.name)
app = current_app
except:
app = global_app
return app'''
获取当前本地时间。
'''
def get_datetime_now(self):
return datetime.datetime.now()
'''
获取当前utc时间。
'''
def get_datetime_utcnow(self):
return datetime.datetime.utcnow()
'''
获取当前本地时间字符串后缀。
'''
def get_datetime_now_suffix(self):
#获取datetime当前时间,
datetime_now = self.get_datetime_now()
#转换自定义格式时间字符串
time_string = datetime_now.strftime("%Y-%m-%d-%H-%M-%S")
#获取datetime当前时间,微秒。
time_string = "_" + time_string + "-" + str(datetime.datetime.now().microsecond)
return time_string
'''
根据时间戳(timestamp),获取UTC显示时间(datetime)。
'''
def get_datetime_utc_form_timestamp(self, timestamp):
return datetime.datetime.utcfromtimestamp(timestamp)
'''
根据显示时间(datetime),获取UTC时间戳(timestamp)。。
'''
def get_timestamp_from_datetime_utc(self, datetime_utc):
#先给出time.struct_time对象。然后通过calendar.timegm转换时间戳。
return calendar.timegm(datetime_utc.timetuple())
提供数据库表的格式,给出一个用户表的典型例子。
#flask_login要的回调,作用是把user实例的user_id的str类型写入cookie。
@login_manager.user_loader
def load_user(user_id):
return User.query.get(int(user_id))
class User(UserMixin, db.Model):
__tablename__ = "users"
#primary_key不能为空,unique可以为空。
id = db.Column(db.Integer, primary_key=True)
#用email注册,还是用tel注册,前端可以控制。
username = db.Column(db.String(64), unique=True, index=True)
password_hash = db.Column(db.String(128))
tel = db.Column(db.String(16), unique=True, index=True)
email = db.Column(db.String(64))
role = db.Column(db.String(64))
posts = db.relationship("Post", backref="author", lazy="dynamic")
#__repr()__方法,返回一个具有可读性的字符串表示模型,可在调试和测试时使用。只要打印User实例,就会出现。
def __repr__(self):
return '<User %r>' % self.username
@property
#password的读属性返回
def password(self):
raise AttributeError('password is not a readable attribute')
@password.setter
#password的写属性返回。注意,这个是属性值,使用方式是user.password = "xxx",password_hash自动生成。
def password(self, password):
self.password_hash = generate_password_hash(password)
#校验不同用户注册存储的密码。返回值为 True 表明密码正确。
def verify_password(self, password):
return check_password_hash(self.password_hash, password)
#将所有信息转化为dict/json消息。
def get_user_dict(self):
user_dict = {}
user_dict["username"] = self.username
user_dict["tel"] = self.tel
user_dict["email"] = self.email
user_dict["role"] = self.role
return user_dict
给出一个文章表的典型例子。
class Post(db.Model):
__tablename__ = "posts"
id = db.Column(db.Integer, primary_key=True)
title = db.Column(db.Text)
body = db.Column(db.Text)
#国际时间的时间戳
timestamp = db.Column(db.DateTime, index=True, default=common_obj.get_datetime_utcnow())
author_id = db.Column(db.Integer, db.ForeignKey("users.id"))
# 将所有信息转化为dict/json消息。
def get_post_dict(self):
post_dict = {}
#注意,url_for()使用,必须有一条route相对应。
#url_for() 函数最简单的用法是以视图函数名(或者app.add_url_route() 定义路由时使用的端点名)作为参数,返回对应的URL。
#id=000,是动态加上id。
#_external绝对路径。
post_dict["url"] = url_for("page.post", id=self.id, _external=True)
post_dict["id"] = self.id
post_dict["title"] = self.title
post_dict["body"] = self.body
#*1000,得到毫秒。
post_dict["timestamp"] = common_obj.get_timestamp_from_datetime_utc(self.timestamp)*1000
user = User.query.filter_by(id=self.author_id).first()
post_dict["author"] = user.username
return post_dict