wechat-admin:Flask使用篇
/ / / 阅读数:4412在 Flask 最佳实践 里面有三项在本项目也有应用:
- 怎么用扩展
- 自定义 RESTAPI 的处理
- local_settings.py
这我就不再复述了,看些不一样的内容吧。
Flask 处理静态资源
理论上应该使用 Nginx 来处理静态资源,但是 wechat-admin 不是面向用户的产品,所以为了便利直接使用 Flask 提供的 SharedDataMiddleware 中间件:
from werkzeug.wsgi import SharedDataMiddleware app = Flask(__name__) app.add_url_rule('/uploads/<filename>', 'uploaded_file', build_only=True) app.wsgi_app = SharedDataMiddleware(app.wsgi_app, { '/uploads': app.config['UPLOAD_FOLDER'] }) |
在给对应群聊 / 用户发消息时添加的文件保存在服务器上面,可以通过/uploads/<filename>
访问到。
CORS
本地开发时,前端运行的端口和后端 API 接口用的端口不一致,为了方便本地开发可以添加一个钩子,利用 CORS 实现跨域请求:
# For local test @app.after_request def after_request(response): response.headers.add('Access-Control-Allow-Origin', '*') response.headers.add( 'Access-Control-Allow-Headers', 'Content-Type,Authorization') response.headers.add('Access-Control-Allow-Methods', 'GET,PUT,POST,DELETE') return response |
注意,为了简单省事 Access-Control-Allow-Origin 的值设成了'*'。不过事实上,如果你使用 Webpack,可以在 config/index.js 文件的 dev 键下使用 changeOrigin,举个例子:
dev: { ... proxyTable: { '/j/admin': { target: 'http://localhost:8100, changeOrigin: true } }, cssSourceMap: false } |
自定义报错状态码
如果项目是商业级别,我通常建议在 API 接口管理上自定义一些状态码:
unknown_error = (1000, 'unknown_error', 400) access_forbidden = (1001, 'access_forbidden', 403) unimplemented_error = (1002, 'unimplemented_error', 400) not_found = (1003, 'not_found', 404) illegal_state = (1004, 'illegal_state', 400) |
这样做的好处的是在文档中能让使用者清晰的知道某种错误的意义,也能让开发者了解在什么情况下抛出什么样的错误。
对于本项目的设计,自定义的异常类型要接受这样的三个参数:
from views.utils import ApiResult class ApiException(Exception): def __init__(self, error, real_message=None): self.code, self.message, self.status = error if real_message is not None: self.message = real_message def to_result(self): return ApiResult({'msg': self.message, 'r': self.code}, status=self.status) |
为此还需要通过 errorhandler 指定 ApiException 类型异常如何处理:
@json_api.errorhandler(ApiException) def api_error_handler(error): return error.to_result() |
现在在业务中在应该抛错误的地方直接这么用就好了:
raise ApiException(errors.not_found) |
另外我们也要通过 errorhandler 指定对于其他引起 404、500 以及 403 的地方做类似的错误:
@json_api.errorhandler(403) @json_api.errorhandler(404) @json_api.errorhandler(500) def error_handler(error): if hasattr(error, 'name'): msg = error.name code = error.code else: msg = error.message code = 500 return ApiResult({'message': msg}, status=code) |
这样就保证了 API 返回的都是统一格式的结果,而不会抛出错误了。
登录的逻辑
登陆的视图比较简单:
from ext import sse from libs.globals import current_bot @json_api.route('/login', methods=['post']) def login(): user = get_logged_in_user(current_bot) from wechat.tasks import retrieve_data retrieve_data.delay() sse.publish({'type': 'logged_in', 'user': user}, type='login') return {'msg': ''} |
其中的 get_logged_in_user 是业务逻辑就不展示了,有兴趣的可以看源码了解。retrieve_data 就是之前说的获取微信联系人、群聊、公众号的任务,在视图内部调用 delay 方法就可以让它异步执行了。同时 sse.publish 会发一个登录的推送让前端页面做对应的处理。
上述逻辑里面最不好理解的是 current_bot。我们都知道 Flask 自带了 4 个上下文,比如 request、current_app、session 和 g。使用 wxpy 获得 bot 是这样的:
def get_bot(): bot = Bot('bot.pkl', qr_path=os.path.join( here, '../static/img/qr_code.png'), console_qr=None) bot.enable_puid() bot.messages.max_history = 0 return bot |
本来应该是把这个对象序列化存在 Redis 中的,但是由于设计的时候没有考虑序列化的问题,我对它的代码也不熟,所以就直接使用上下文来做了:
from werkzeug.local import LocalStack, LocalProxy def _find_bot(): from .wx import get_bot top = _wx_ctx_stack.top if top is None: top = get_bot() _wx_ctx_stack.push(top) return top _wx_ctx_stack = LocalStack() current_bot = LocalProxy(_find_bot) |
不过要注意,启动多进程的话,理论上每个进程都会创建多个 bot,不过由于 bot 存在的意义是执行,另外 get_bot 执行成功一次后在重复执行会使用之前生成的 pkl 文件,所以这样用也是没有问题。
在使用 wechat-admin 项目的 README 文档中,我特意说了先扫码登陆后才能启动 Celery,这是因为不这样做的话,current_bot 还是阻塞状态,会在启动 Celery 的时候先卡在让你扫码登录上,这点要注意。
总结
今天为止,整个 wechat-admin 就介绍完啦~