api-automation-test/ApiAutomationTest/app/cores/variable.py

116 lines
3.4 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

# coding=utf-8
from typing import Dict
from flask import session
from app.extensions import session_id_manager
from app.cores.dictionaries import DISPATCHER_TRIGGER_TYPE
class Variable:
"""
VariablePool = {
session_id: {
project_id: {
key: value,
key1: value1,
...
},
project_id2: {
...
}
},
session_id2:{
...
},
...
}
ScheduleVariablePool = {
project_id: {
key: value,
key1: value1,
...
},
project_id2: {
...
},
...
}
"""
# 变量池
VariablePool = {}
# 定时任务变量池
ScheduleVariablePool = {}
@classmethod
def get_project_variable(cls, project_id):
"""
获取项目中所有定义的变量
:param id: 项目id
:type id: int
:return: 项目级所有变量
:rtype: Dict
"""
cls._set_default_project_variable(project_id=project_id)
if session.get('dispatcher_trigger_type') == DISPATCHER_TRIGGER_TYPE.BY_SCHEDULE:
project_id = session.get('project_id')
if project_id:
return cls.ScheduleVariablePool.get(project_id)
else:
session_id = session_id_manager.get_session_id()
return cls.VariablePool.get(session_id).get(project_id)
@classmethod
def set_project_variable(cls, project_id, key, value):
"""
设置项目级别变量
:param project_id: 项目id
:type project_id: int
:param key: 变量名
:type key: str
:param value: 变量值
:type value: str
"""
cls._set_default_project_variable(project_id=project_id)
if session.get('dispatcher_trigger_type') == DISPATCHER_TRIGGER_TYPE.BY_SCHEDULE:
project_id = session.get('project_id')
if project_id:
cls.ScheduleVariablePool.get(project_id).update({
key: value
})
else:
session_id = session_id_manager.get_session_id()
cls.VariablePool.get(session_id).get(project_id).update({
key: value
})
@classmethod
def _set_default_project_variable(cls, project_id):
"""当VariablePool为空时为其设置默认值避免取出值为None的项目级变量池"""
if session.get('dispatcher_trigger_type') == DISPATCHER_TRIGGER_TYPE.BY_SCHEDULE:
project_id = session.get('project_id')
if project_id:
project_var = cls.ScheduleVariablePool.get(project_id)
if project_var is None:
cls.ScheduleVariablePool.update({
project_id: {}
})
else:
session_id = session_id_manager.get_session_id()
session_var = cls.VariablePool.get(session_id)
if session_var is None:
cls.VariablePool.update({
session_id: {
project_id: {}
}
})
else:
project_var = cls.VariablePool.get(session_id).get(project_id)
if project_var is None:
cls.VariablePool.get(session_id).update({
project_id: {}
})