shaba/openuds
1
0
Fork 1
mirror of https://github.com/dkmstr/openuds.git synced 2025-03-20 06:50:23 +03:00
Code Issues Releases Wiki Activity

Refactor custom methods to use ModelCustomMethod for improved clarity and consistency

Browse Source
This commit is contained in:
Adolfo Gómez García 2025-01-25 20:03:35 +01:00
parent b9f4e7f2ea
commit beccee144a
No known key found for this signature in database
GPG Key ID: DD1ABF20724CDA23
14 changed files with 124 additions and 105 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
server/src/uds/REST/__init__.py
View File

@ -31,6 +31,6 @@ Author: Adolfo Gómez, dkmaster at dkmon dot com
"""
# pyright: reportUnusedImport=false
# Convenience imports, must be present before initializing handlers
from .handlers import Handler, HelpPath
from .handlers import Handler
from .dispatcher import Dispatcher
from .documentation import Documentation

80
server/src/uds/REST/dispatcher.py
View File

@ -41,12 +41,13 @@ from django.utils.decorators import method_decorator
from django.views.decorators.csrf import csrf_exempt
from django.views.generic.base import View
from uds.core.types.rest import HandlerNode
from uds.core import consts, exceptions, types
from uds.core.util import modfinder
from . import processors, log
from .handlers import Handler
from .model import DetailHandler, ModelHandler
from .model import DetailHandler
# Not imported at runtime, just for type checking
if typing.TYPE_CHECKING:
@ -57,73 +58,6 @@ logger = logging.getLogger(__name__)
__all__ = ['Handler', 'Dispatcher']
@dataclasses.dataclass(frozen=True)
class HandlerNode:
"""
Represents a node on the handler tree
"""
name: str
handler: typing.Optional[type[Handler]]
parent: typing.Optional['HandlerNode']
children: dict[str, 'HandlerNode']
def __str__(self) -> str:
return f'HandlerNode({self.name}, {self.handler}, {self.children})'
def __repr__(self) -> str:
return str(self)
def tree(self, level: int = 0) -> str:
"""
Returns a string representation of the tree
"""
if self.handler is None:
return f'{" " * level}|- {self.name}\n' + ''.join(
child.tree(level + 1) for child in self.children.values()
)
ret = f'{" " * level}{self.name} ({self.handler.__name__} {self.full_path()})\n'
if issubclass(self.handler, ModelHandler):
# Add custom_methods
for method in self.handler.custom_methods:
ret += f'{" " * level} |- {method}\n'
# Add detail methods
if self.handler.detail:
for method in self.handler.detail.keys():
ret += f'{" " * level} |- {method}\n'
return ret + ''.join(child.tree(level + 1) for child in self.children.values())
def find_path(self, path: str | list[str]) -> typing.Optional['HandlerNode']:
"""
Returns the node for a given path, or None if not found
"""
if not path or not self.children:
return self
path = path.split('/') if isinstance(path, str) else path
if path[0] not in self.children:
return None
return self.children[path[0]].find_path(path[1:]) # Recursive call
def full_path(self) -> str:
"""
Returns the full path of this node
"""
if self.name == '' or self.parent is None:
return ''
parent_full_path = self.parent.full_path()
if parent_full_path == '':
return self.name
return f'{parent_full_path}/{self.name}'
class Dispatcher(View):
"""
This class is responsible of dispatching REST requests
@ -172,7 +106,7 @@ class Dispatcher(View):
handler_node = Dispatcher.base_handler_node.find_path(path)
if not handler_node:
return http.HttpResponseNotFound('Service not found', content_type="text/plain")
logger.debug("REST request: %s (%s)", handler_node, handler_node.full_path())
# Now, service points to the class that will process the request
@ -192,7 +126,9 @@ class Dispatcher(View):
return http.HttpResponseNotAllowed(['GET', 'POST', 'PUT', 'DELETE'], content_type="text/plain")
# Path here has "remaining" path, that is, method part has been removed
args = path[len(handler_node.full_path()):].split('/')[1:] # First element is always empty, so we skip it
args = path[len(handler_node.full_path()) :].split('/')[
1:
] # First element is always empty, so we skip it
handler: typing.Optional[Handler] = None
@ -207,7 +143,9 @@ class Dispatcher(View):
)
operation: collections.abc.Callable[[], typing.Any] = getattr(handler, http_method)
except processors.ParametersException as e:
logger.debug('Path: %s', )
logger.debug(
'Path: %s',
)
logger.debug('Error: %s', e)
log.log_operation(handler, 400, types.log.LogLevel.ERROR)

9
server/src/uds/REST/handlers.py
View File

@ -52,13 +52,6 @@ if typing.TYPE_CHECKING:
logger = logging.getLogger(__name__)
class HelpPath(typing.NamedTuple):
"""
Help path class
"""
path: str
help: str
class Handler:
"""
REST requests handler base class
@ -80,7 +73,7 @@ class Handler:
# For implementing help
# A list of pairs of (path, help) for subpaths on this handler
help_paths: typing.ClassVar[list[HelpPath]] = []
help_paths: typing.ClassVar[list[types.rest.HelpPath]] = []
help_text: typing.ClassVar[str] = 'No help available'
_request: 'ExtendedHttpRequestWithUser' # It's a modified HttpRequest

5
server/src/uds/REST/methods/accounts.py
View File

@ -59,7 +59,10 @@ class Accounts(ModelHandler):
model = Account
detail = {'usage': AccountsUsage}
custom_methods = [('clear', True), ('timemark', True)]
custom_methods = [
types.rest.ModelCustomMethod('clear', True),
types.rest.ModelCustomMethod('timemark', True),
]
save_fields = ['name', 'comments', 'tags']

2
server/src/uds/REST/methods/authenticators.py
View File

@ -62,7 +62,7 @@ logger = logging.getLogger(__name__)
class Authenticators(ModelHandler):
model = Authenticator
# Custom get method "search" that requires authenticator id
custom_methods = [('search', True)]
custom_methods = [types.rest.ModelCustomMethod('search', True)]
detail = {'users': Users, 'groups': Groups}
save_fields = ['name', 'comments', 'tags', 'priority', 'small_name', 'mfa_id:_']

10
server/src/uds/REST/methods/meta_pools.py
View File

@ -108,7 +108,10 @@ class MetaPools(ModelHandler):
{'tags': {'title': _('tags'), 'visible': False}},
]
custom_methods = [('setFallbackAccess', True), ('getFallbackAccess', True)]
custom_methods = [
types.rest.ModelCustomMethod('setFallbackAccess', True),
types.rest.ModelCustomMethod('getFallbackAccess', True),
]
def item_as_dict(self, item: 'Model') -> dict[str, typing.Any]:
item = ensure.is_instance(item, MetaPool)
@ -205,10 +208,7 @@ class MetaPools(ModelHandler):
'name': 'servicesPoolGroup_id',
'choices': [gui.choice_image(-1, _('Default'), DEFAULT_THUMB_BASE64)]
+ gui.sorted_choices(
[
gui.choice_image(v.uuid, v.name, v.thumb64)
for v in ServicePoolGroup.objects.all()
]
[gui.choice_image(v.uuid, v.name, v.thumb64) for v in ServicePoolGroup.objects.all()]
),
'label': gettext('Pool group'),
'tooltip': gettext('Pool group for this pool (for pool classify on display)'),

6
server/src/uds/REST/methods/providers.py
View File

@ -62,7 +62,11 @@ class Providers(ModelHandler):
model = Provider
detail = {'services': DetailServices, 'usage': ServicesUsage}
custom_methods = [('allservices', False), ('service', False), ('maintenance', True)]
custom_methods = [
types.rest.ModelCustomMethod('allservices', False),
types.rest.ModelCustomMethod('service', False),
types.rest.ModelCustomMethod('maintenance', True),
]
save_fields = ['name', 'comments', 'tags']

7
server/src/uds/REST/methods/servers_management.py
View File

@ -403,7 +403,9 @@ class ServersServers(DetailHandler):
class ServersGroups(ModelHandler):
custom_methods = [('stats', True)]
custom_methods = [
types.rest.ModelCustomMethod('stats', True),
]
model = models.ServerGroup
model_filter = {
'type__in': [
@ -511,8 +513,7 @@ class ServersGroups(ModelHandler):
def stats(self, item: 'Model') -> typing.Any:
# Avoid circular imports
from uds.core.managers.servers import ServerManager
item = ensure.is_instance(item, models.ServerGroup)
return [

10
server/src/uds/REST/methods/services_pools.py
View File

@ -119,11 +119,11 @@ class ServicesPools(ModelHandler):
table_row_style = types.ui.RowStyleInfo(prefix='row-state-', field='state')
custom_methods = [
('set_fallback_access', True),
('get_fallback_access', True),
('actions_list', True),
('list_assignables', True),
('create_from_assignable', True),
types.rest.ModelCustomMethod('set_fallback_access', True),
types.rest.ModelCustomMethod('get_fallback_access', True),
types.rest.ModelCustomMethod('actions_list', True),
types.rest.ModelCustomMethod('list_assignables', True),
types.rest.ModelCustomMethod('create_from_assignable', True),
]
def get_items(

3
server/src/uds/REST/methods/stats.py
View File

@ -34,8 +34,9 @@ import logging
import datetime
import typing
from uds.core.types.rest import HelpPath
from uds.core import types
from uds.REST import Handler, HelpPath
from uds.REST import Handler
from uds import models
from uds.core.util.stats import counters

3
server/src/uds/REST/methods/system.py
View File

@ -38,13 +38,14 @@ import pickletools
import typing
from uds import models
from uds.core.types.rest import HelpPath
from uds.core import exceptions, types
from uds.core.util import permissions
from uds.core.util.cache import Cache
from uds.core.util.model import process_uuid, sql_now
from uds.core.types.states import State
from uds.core.util.stats import counters
from uds.REST import Handler, HelpPath
from uds.REST import Handler
logger = logging.getLogger(__name__)

6
server/src/uds/REST/methods/user_services.py
View File

@ -56,10 +56,6 @@ class AssignedService(DetailHandler):
Rest handler for Assigned Services, wich parent is Service
"""
custom_methods = [
'reset',
]
custom_methods = ['reset']
@staticmethod
@ -270,7 +266,7 @@ class CachedService(AssignedService):
Rest handler for Cached Services, wich parent is Service
"""
custom_methods: typing.ClassVar[list[str]] = [] # Remove custom methods from assigned services
custom_methods = [] # Remove custom methods from assigned services
def get_items(self, parent: 'Model', item: typing.Optional[str]) -> types.rest.ManyItemsDictType:
parent = ensure.is_instance(parent, models.ServicePool)

2
server/src/uds/REST/model/model.py
View File

@ -88,7 +88,7 @@ class ModelHandler(BaseModelHandler):
# This is an array of tuples of two items, where first is method and second inticates if method needs parent id (normal behavior is it needs it)
# For example ('services', True) -- > .../id_parent/services
# ('services', False) --> ..../services
custom_methods: typing.ClassVar[list[tuple[str, bool]]] = (
custom_methods: typing.ClassVar[list[types.rest.ModelCustomMethod]] = (
[]
) # If this model respond to "custom" methods, we will declare them here
# If this model has details, which ones

84
server/src/uds/core/types/rest.py
View File

@ -34,6 +34,10 @@ import typing
import dataclasses
import collections.abc
if typing.TYPE_CHECKING:
from uds.REST.handlers import Handler
TypeInfoDict = dict[str, typing.Any] # Alias for type info dict
@ -106,4 +110,82 @@ ItemGeneratorType = typing.Generator[ItemDictType, None, None]
ManyItemsDictType = typing.Union[ItemListType, ItemDictType, ItemGeneratorType]
#
FieldType = collections.abc.Mapping[str, typing.Any]
FieldType = collections.abc.Mapping[str, typing.Any]
class HelpPath(typing.NamedTuple):
"""
Help helper class
"""
path: str
help: str
@dataclasses.dataclass(frozen=True)
class HandlerNode:
"""
Represents a node on the handler tree for rest services
"""
name: str
handler: typing.Optional[type['Handler']]
parent: typing.Optional['HandlerNode']
children: dict[str, 'HandlerNode']
def __str__(self) -> str:
return f'HandlerNode({self.name}, {self.handler}, {self.children})'
def __repr__(self) -> str:
return str(self)
def tree(self, level: int = 0) -> str:
"""
Returns a string representation of the tree
"""
from uds.REST.model import ModelHandler
if self.handler is None:
return f'{" " * level}|- {self.name}\n' + ''.join(
child.tree(level + 1) for child in self.children.values()
)
ret = f'{" " * level}{self.name} ({self.handler.__name__} {self.full_path()})\n'
if issubclass(self.handler, ModelHandler):
# Add custom_methods
for method in self.handler.custom_methods:
ret += f'{" " * level} |- {method}\n'
# Add detail methods
if self.handler.detail:
for method in self.handler.detail.keys():
ret += f'{" " * level} |- {method}\n'
return ret + ''.join(child.tree(level + 1) for child in self.children.values())
def find_path(self, path: str | list[str]) -> typing.Optional['HandlerNode']:
"""
Returns the node for a given path, or None if not found
"""
if not path or not self.children:
return self
path = path.split('/') if isinstance(path, str) else path
if path[0] not in self.children:
return None
return self.children[path[0]].find_path(path[1:]) # Recursive call
def full_path(self) -> str:
"""
Returns the full path of this node
"""
if self.name == '' or self.parent is None:
return ''
parent_full_path = self.parent.full_path()
if parent_full_path == '':
return self.name
return f'{parent_full_path}/{self.name}'