github/jumpserver
1
0
Fork 0
mirror of https://github.com/jumpserver/jumpserver.git synced 2026-01-29 21:51:31 +00:00
Code Issues Projects Releases Wiki Activity

refactor: 重构用户授权资产查询工具(重构中..) (#9225)

Browse Source 
* refactor: 重构用户授权资产查询工具(重构中..)

* perf: 修改 get_perm_nodes_assets 名称

* refactor: 优化用户授权节点查询工具; 删除UnionQuerySet工具

Co-authored-by: Bai <baijiangjie@gmail.com>
This commit is contained in:
fit2bot
2022-12-21 17:36:44 +08:00
committed by GitHub
parent 510ca9a5b8
commit 34cc3b233d
11 changed files with 269 additions and 415 deletions
Download Patch File Download Diff File Expand all files Collapse all files

84
apps/common/db/models.py
View File

@@ -9,9 +9,8 @@
- 此文件中添加代码的时候，注意不要跟 `django.db.models` 中的命名冲突
"""
import inspect
import uuid
from functools import reduce, partial
from functools import reduce
from django.db import models
from django.db import transaction
@@ -96,87 +95,6 @@ def output_as_string(field_name):
return ExpressionWrapper(F(field_name), output_field=models.CharField())
class UnionQuerySet(QuerySet):
after_union = ['order_by']
not_return_qs = [
'query', 'get', 'create', 'get_or_create',
'update_or_create', 'bulk_create', 'count',
'latest', 'earliest', 'first', 'last', 'aggregate',
'exists', 'update', 'delete', 'as_manager', 'explain',
]
def __init__(self, *queryset_list):
self.queryset_list = queryset_list
self.after_union_items = []
self.before_union_items = []
def __execute(self):
queryset_list = []
for qs in self.queryset_list:
for attr, args, kwargs in self.before_union_items:
qs = getattr(qs, attr)(*args, **kwargs)
queryset_list.append(qs)
union_qs = reduce(lambda x, y: x.union(y), queryset_list)
for attr, args, kwargs in self.after_union_items:
union_qs = getattr(union_qs, attr)(*args, **kwargs)
return union_qs
def __before_union_perform(self, item, *args, **kwargs):
self.before_union_items.append((item, args, kwargs))
return self.__clone(*self.queryset_list)
def __after_union_perform(self, item, *args, **kwargs):
self.after_union_items.append((item, args, kwargs))
return self.__clone(*self.queryset_list)
def __clone(self, *queryset_list):
uqs = UnionQuerySet(*queryset_list)
uqs.after_union_items = self.after_union_items
uqs.before_union_items = self.before_union_items
return uqs
def __getattribute__(self, item):
if item.startswith('__') or item in UnionQuerySet.__dict__ or item in [
'queryset_list', 'after_union_items', 'before_union_items'
]:
return object.__getattribute__(self, item)
if item in UnionQuerySet.not_return_qs:
return getattr(self.__execute(), item)
origin_item = object.__getattribute__(self, 'queryset_list')[0]
origin_attr = getattr(origin_item, item, None)
if not inspect.ismethod(origin_attr):
return getattr(self.__execute(), item)
if item in UnionQuerySet.after_union:
attr = partial(self.__after_union_perform, item)
else:
attr = partial(self.__before_union_perform, item)
return attr
def __getitem__(self, item):
return self.__execute()[item]
def __iter__(self):
return iter(self.__execute())
def __str__(self):
return str(self.__execute())
def __repr__(self):
return repr(self.__execute())
@classmethod
def test_it(cls):
from assets.models import Asset
assets1 = Asset.objects.filter(hostname__startswith='a')
assets2 = Asset.objects.filter(hostname__startswith='b')
qs = cls(assets1, assets2)
return qs
class MultiTableChildQueryset(QuerySet):
def bulk_create(self, objs, batch_size=None):