python不同模块统一管理操作全局变量
全局变量仅限于单实例模式哈,对于多实例集群如k8s docker则建议使用redis来实现以上请特别留意哈,比如docker中多个pod基本上是没有办法共享全局变量的,一直漂,读的值和设的值会有问题OK,言归正传,对于单实例的模式而言要如何做到不同模块统一管理全局变量呢?一、全局变量管理模块首先定义一个全局变量管理模块# common/gloabalvar.py# -*- coding...
·
全局变量仅限于单实例模式哈,对于多实例集群如k8s docker则建议使用redis来实现
以上请特别留意哈,比如docker中多个pod基本上是没有办法共享全局变量的,一直漂,读的值和设的值会有问题
OK,言归正传,对于单实例的模式而言要如何做到不同模块统一管理全局变量呢?
一、全局变量管理模块
首先定义一个全局变量管理模块
# common/gloabalvar.py
# -*- coding:utf-8 -*-
"""
global var module
"""
def _init():
global _global_dict
_global_dict = {}
def set_value(name, value):
"""定义一个全局变量"""
_global_dict[name] = value
def get_value(name, defValue=None):
"""获取一个全局变量值,不能存在则返回默认值"""
try:
return _global_dict[name]
except KeyError:
return defValue
二、初始化及定义(设置)全局变量
# main.py
import app.common.globalvar as globalvar
if __name__ == "__main__":
# init global var
globalvar._init()
globalvar.set_value('itsm_deadline', SettingConfig.ITSM_DAILY_DEADLINE)
globalvar.set_value('email_subject', SettingConfig.EMAIL_SUBJECT)
globalvar.set_value('email_recipients', SettingConfig.EMAIL_RECIPIENTS)
globalvar.set_value('auto_submit_status', SettingConfig.AUTO_SUBMIT_STATUS)
三、获取全局变量
# business.py
import app.common.globalvar as globalvar
status = globalvar.get_value('auto_submit_status')
附加:API管理全局变量,都是基于上述三点完成
对于flask-restful的web应用则可以通过API接口查看全局变量的值以及通过API接口设置和管理它们
# app/api/v1/test/test.py
from flask_restplus import Resource
import app.common.globalvar as globalvar
from .marshal_model import admin_api_ns as api
@api.route('/')
class GlobalVar(Resource):
"""全局变量管理API"""
@api.doc('Get all the global var')
@api.response(HTTP_GET_SUCCESS, 'query success')
def get(self):
"""查询所有的全局变量"""
glist = globalvar._global_dict
return glist, HTTP_GET_SUCCESS
@api.route('/auto')
class AutoSubmit(Resource):
"""查看及修改自动状态的API"""
@api.doc('check auto submit status')
@api.marshal_with(set_autosubmit)
@api.response(HTTP_GET_SUCCESS, 'query success')
def get(self):
"""测试查看自动状态开关状态"""
status = globalvar.get_value('auto_submit_status')
# status = query_autojob_status()
return {'status': status}, HTTP_GET_SUCCESS
@login_required
@admin_required
@api.doc('set auto submit the planlist everyday')
@api.expect(set_autosubmit)
def post(self):
"""设置每天定时自动开关"""
data = request.json
status = data.get('status')
globalvar.set_value('auto_submit_status', status)
# resp, returncode = update_autojob_status(status)
return {'message': 'success'}, HTTP_POST_SUCCESS
更多推荐
已为社区贡献4条内容
所有评论(0)