mirror of
https://github.com/jumpserver/jumpserver.git
synced 2025-09-06 18:00:57 +00:00
reactor&remove: 重构applications模块 & 移除applications、perms中已不再使用的模块 (#5374)
* reactor: 重构applications模块 & 删除applications、perms中已不再使用的模块 * reactor: 1. 针对application.attrs字段的view-serializer映射逻辑,采用DynamicMapping的方案重写; * reactor: 2. 删除applications和perms模块中已不再使用的database-app/k8s-app/remote-app模块; * reactor: 添加迁移文件(删除perms/databaseperrmission/remoteapppermission/k8sapppermission) * reactor: 修改细节 Co-authored-by: Bai <bugatti_it@163.com>
This commit is contained in:
@@ -1,6 +1,7 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
#
|
||||
import re
|
||||
import data_tree
|
||||
from collections import OrderedDict
|
||||
from itertools import chain
|
||||
import logging
|
||||
@@ -10,6 +11,8 @@ from functools import wraps
|
||||
import time
|
||||
import ipaddress
|
||||
import psutil
|
||||
from django.utils.translation import ugettext_lazy as _
|
||||
from ..exceptions import JMSException
|
||||
|
||||
|
||||
UUID_PATTERN = re.compile(r'\w{8}(-\w{4}){3}-\w{12}')
|
||||
@@ -251,3 +254,152 @@ def get_disk_usage():
|
||||
mount_points = [p.mountpoint for p in partitions]
|
||||
usages = {p: psutil.disk_usage(p) for p in mount_points}
|
||||
return usages
|
||||
|
||||
|
||||
# Verify that `value` is in `choices` and throw an `JMSException`
|
||||
# ---------------------------------------------------------------
|
||||
|
||||
|
||||
def check_value_in_choices(value, choices, **kwargs):
|
||||
# get raise parameters from kwargs
|
||||
raise_exception = kwargs.get('raise_exception', False)
|
||||
raise_error_msg = kwargs.get('raise_error_msg', None)
|
||||
raise_reverse = kwargs.get('raise_reverse', False)
|
||||
|
||||
def should_raise():
|
||||
"""
|
||||
Simplify the following logic:
|
||||
|
||||
if raise_exception:
|
||||
if raise_reverse and value_in_choices:
|
||||
return True
|
||||
else:
|
||||
return False
|
||||
|
||||
if not raise_reverse and not value_in_choices:
|
||||
return True
|
||||
else:
|
||||
return False
|
||||
else:
|
||||
return False
|
||||
"""
|
||||
return raise_exception and raise_reverse == value_in_choices
|
||||
|
||||
value_in_choices = True if value in choices else False
|
||||
|
||||
if not should_raise():
|
||||
return value_in_choices
|
||||
|
||||
if raise_error_msg is None:
|
||||
raise_error_msg = _('Value `{}` is not in Choices: `{}`'.format(value, choices))
|
||||
|
||||
raise JMSException(raise_error_msg)
|
||||
|
||||
|
||||
# Quick lookup dict
|
||||
# -----------------
|
||||
|
||||
|
||||
class QuickLookupDict(object):
|
||||
"""
|
||||
说明:
|
||||
dict 类型数据的快速查找
|
||||
作用:
|
||||
可根据指定 key 的深度 path 快速查找出对应的 value 值
|
||||
依赖:
|
||||
data-tree==0.0.1
|
||||
实现:
|
||||
通过对 data-tree 库的封装来实现
|
||||
"""
|
||||
|
||||
def __init__(self, data, key_delimiter='.'):
|
||||
self._check_data_type(data, type_choices=(dict, ), error='Expected `data` type is dict')
|
||||
self.data = data
|
||||
self.key_delimiter = key_delimiter
|
||||
self._data_tree = self._get_data_tree(data, key_delimiter)
|
||||
|
||||
# Method encapsulated of `data-tree`
|
||||
# ----------------------------------
|
||||
|
||||
@staticmethod
|
||||
def _get_data_tree(data, key_delimiter):
|
||||
tree = data_tree.Data_tree_node(
|
||||
arg_data=data, arg_string_delimiter_for_path=key_delimiter
|
||||
)
|
||||
return tree
|
||||
|
||||
def _get_data_tree_node(self, path):
|
||||
return self._data_tree.get(arg_path=path)
|
||||
|
||||
@staticmethod
|
||||
def _get_data_tree_node_original_data(tree_node):
|
||||
if isinstance(tree_node, data_tree.Data_tree_node):
|
||||
data = tree_node.get_data_in_format_for_export()
|
||||
else:
|
||||
data = tree_node
|
||||
return data
|
||||
|
||||
# Method called internally
|
||||
# ------------------------
|
||||
|
||||
@staticmethod
|
||||
def _check_data_type(data, type_choices, error=None):
|
||||
error = error or '`data` type error, {} => {}'.format(type(data), type_choices)
|
||||
assert isinstance(data, type_choices), error
|
||||
|
||||
@staticmethod
|
||||
def _check_object_callable(_object):
|
||||
if _object is None:
|
||||
return False
|
||||
if not callable(_object):
|
||||
return False
|
||||
return True
|
||||
|
||||
# Method called externally
|
||||
# ------------------------
|
||||
|
||||
def get(self, key_path, default=None):
|
||||
error = 'key_path - can be either a list of keys, or a delimited string.'
|
||||
self._check_data_type(key_path, (list, str,), error=error)
|
||||
|
||||
tree_node = self._get_data_tree_node(key_path)
|
||||
if tree_node is None:
|
||||
return default
|
||||
value = self._get_data_tree_node_original_data(tree_node)
|
||||
return value
|
||||
|
||||
def get_many(self, key_paths, default=None):
|
||||
values = [
|
||||
self.get(key_path, default=default) for key_path in key_paths
|
||||
]
|
||||
return values
|
||||
|
||||
def find_one(self, key_paths, default=None, callable_filter=None):
|
||||
"""
|
||||
按照 key_paths 顺序查找,返回第一个满足 `callable_filter` 规则的值
|
||||
"""
|
||||
|
||||
def get_data_filter():
|
||||
if self._check_object_callable(callable_filter):
|
||||
return callable_filter
|
||||
return self.__default_find_callable_filter
|
||||
|
||||
_filter = get_data_filter()
|
||||
|
||||
for key_path in key_paths:
|
||||
value = self.get(key_path=key_path)
|
||||
if _filter(key_path, value):
|
||||
return value
|
||||
|
||||
return default
|
||||
|
||||
# Method default
|
||||
# --------------
|
||||
|
||||
@staticmethod
|
||||
def __default_find_callable_filter(key_path, value):
|
||||
return value is not None
|
||||
|
||||
|
||||
|
||||
|
||||
|
Reference in New Issue
Block a user