shaba/openuds
1
0
Fork 1
mirror of https://github.com/dkmstr/openuds.git synced 2025-10-11 03:33:46 +03:00
Code Issues Releases Wiki Activity

Refactor REST model handlers to enhance type safety and consistency

Browse Source 
- Updated `AuthenticatorItem`, `NetworkItem`, `NotifierItem`, `AccessCalendarItem`, `ActionCalendarItem`, `OsManagerItem`, `ProviderItem`, `ReportItem`, `TokenItem`, `ServerItem`, `GroupItem`, `ServiceItem`, `ServicePoolItem`, and `TransportItem` to inherit from `ManagedObjectDictType` for improved type safety.
- Refactored `BaseModelHandler` and `ModelHandler` to use generic types for better type inference.
- Modified `DetailHandler` to extend `BaseModelHandler` with generics.
- Adjusted return types in various methods to utilize the new item types, ensuring consistent type usage across the REST API.
- Introduced `ManagedObjectDictType` to represent managed objects in the REST API, including fields for type and instance.
This commit is contained in:
Adolfo Gómez García
2025-07-26 01:21:06 +02:00
parent d74e6daed2
commit be6cfb0ec5
17 changed files with 337 additions and 249 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -58,7 +58,7 @@ if typing.TYPE_CHECKING:
logger = logging.getLogger(__name__)
class AuthenticatorItem(types.rest.ItemDictType):
class AuthenticatorItem(types.rest.ManagedObjectDictType):
numeric_id: int
id: str
name: str
@@ -72,9 +72,6 @@ class AuthenticatorItem(types.rest.ItemDictType):
mfa_id: str
small_name: str
users_count: int
type: str
type_name: str
type_info: types.rest.TypeInfoDict
permission: int
@@ -196,7 +193,6 @@ class Authenticators(ModelHandler[AuthenticatorItem]):
'users_count': item.users.count(),
'type': type_.mod_type(),
'type_name': type_.mod_name(),
'type_info': self.type_as_dict(type_),
'permission': permissions.effective_permissions(self._user, item),
}

18
server/src/uds/REST/methods/networks.py
View File

@@ -48,20 +48,20 @@ logger = logging.getLogger(__name__)
# Enclosed methods under /item path
class NetworkItem(types.rest.ItemDictType):
id: str
name: str
tags: list[str]
net_string: str
transports_count: int
authenticators_count: int
permission: types.permissions.PermissionType
class Networks(ModelHandler):
class Networks(ModelHandler[NetworkItem]):
"""
Processes REST requests about networks
Implements specific handling for network related requests using GUI
"""
class NetworkItem(types.rest.ItemDictType):
id: str
name: str
tags: list[str]
net_string: str
transports_count: int
authenticators_count: int
permission: types.permissions.PermissionType
model = Network

29
server/src/uds/REST/methods/notifiers.py
View File

@@ -53,19 +53,20 @@ logger = logging.getLogger(__name__)
# Enclosed methods under /item path
class Notifiers(ModelHandler):
class NotifierItem(types.rest.ItemDictType):
id: str
name: str
level: str
enabled: bool
tags: list[str]
comments: str
type: str
type_name: str
permission: types.permissions.PermissionType
class NotifierItem(types.rest.ItemDictType):
id: str
name: str
level: str
enabled: bool
tags: list[str]
comments: str
type: str
type_name: str
permission: types.permissions.PermissionType
class Notifiers(ModelHandler[NotifierItem]):
path = 'messaging'
model = Notifier
save_fields = [
@@ -98,9 +99,7 @@ class Notifiers(ModelHandler):
with Environment.temporary_environment() as env:
notifier = notifier_type(env, None)
local_gui = self.add_default_fields(
notifier.gui_description(), ['name', 'comments', 'tags']
)
local_gui = self.add_default_fields(notifier.gui_description(), ['name', 'comments', 'tags'])
for field in [
{
@@ -119,7 +118,7 @@ class Notifiers(ModelHandler):
'type': types.ui.FieldType.CHECKBOX,
'order': 103,
'default': True,
}
},
]:
self.add_field(local_gui, field)

32
server/src/uds/REST/methods/op_calendars.py
View File

@@ -30,6 +30,7 @@
"""
Author: Adolfo Gómez, dkmaster at dkmon dot com
"""
import datetime
import json
import logging
import typing
@@ -51,10 +52,16 @@ logger = logging.getLogger(__name__)
ALLOW = 'ALLOW'
DENY = 'DENY'
class AccessCalendarItem(types.rest.ItemDictType):
id: str
calendar_id: str
calendar: str
access: str
priority: int
class AccessCalendars(DetailHandler):
class AccessCalendars(DetailHandler[AccessCalendarItem]):
@staticmethod
def as_dict(item: 'models.CalendarAccess|models.CalendarAccessMeta') -> types.rest.ItemDictType:
def as_dict(item: 'models.CalendarAccess|models.CalendarAccessMeta') -> AccessCalendarItem:
return {
'id': item.uuid,
'calendar_id': item.calendar.uuid,
@@ -63,7 +70,7 @@ class AccessCalendars(DetailHandler):
'priority': item.priority,
}
def get_items(self, parent: 'Model', item: typing.Optional[str]) -> types.rest.GetItemsResult:
def get_items(self, parent: 'Model', item: typing.Optional[str]) -> types.rest.GetItemsResult[AccessCalendarItem]:
# parent can be a ServicePool or a metaPool
parent = typing.cast(typing.Union['models.ServicePool', 'models.MetaPool'], parent)
@@ -126,7 +133,20 @@ class AccessCalendars(DetailHandler):
log.log(parent, types.log.LogLevel.INFO, log_str, types.log.LogSource.ADMIN)
class ActionsCalendars(DetailHandler):
class ActionCalendarItem(types.rest.ItemDictType):
id: str
calendar_id: str
calendar: str
action: str
description: str
at_start: bool
events_offset: int
params: dict[str, typing.Any]
pretty_params: str
next_execution: typing.Optional[datetime.datetime]
last_execution: typing.Optional[datetime.datetime]
class ActionsCalendars(DetailHandler[ActionCalendarItem]):
"""
Processes the transports detail requests of a Service Pool
"""
@@ -136,7 +156,7 @@ class ActionsCalendars(DetailHandler):
]
@staticmethod
def as_dict(item: 'models.CalendarAction') -> dict[str, typing.Any]:
def as_dict(item: 'models.CalendarAction') -> ActionCalendarItem:
action = consts.calendar.CALENDAR_ACTION_DICT.get(item.action)
descrption = action.get('description') if action is not None else ''
params = json.loads(item.params)
@@ -154,7 +174,7 @@ class ActionsCalendars(DetailHandler):
'last_execution': item.last_execution,
}
def get_items(self, parent: 'Model', item: typing.Optional[str]) -> types.rest.GetItemsResult:
def get_items(self, parent: 'Model', item: typing.Optional[str]) -> types.rest.GetItemsResult[ActionCalendarItem]:
parent = ensure.is_instance(parent, models.ServicePool)
try:
if item is None:

29
server/src/uds/REST/methods/osmanagers.py
View File

@@ -50,17 +50,17 @@ logger = logging.getLogger(__name__)
# Enclosed methods under /osm path
class OsManagers(ModelHandler):
class OsManagerItem(types.rest.ItemDictType):
id: str
name: str
tags: list[str]
deployed_count: int
type: str
type_name: str
servicesTypes: list[str]
comments: str
permission: types.permissions.PermissionType
class OsManagerItem(types.rest.ManagedObjectDictType):
id: str
name: str
tags: list[str]
deployed_count: int
servicesTypes: list[str]
comments: str
permission: types.permissions.PermissionType
class OsManagers(ModelHandler[OsManagerItem]):
model = OSManager
save_fields = ['name', 'comments', 'tags']
@@ -76,19 +76,20 @@ class OsManagers(ModelHandler):
def os_manager_as_dict(self, osm: OSManager) -> OsManagerItem:
type_ = osm.get_type()
return {
ret_value: OsManagerItem = {
'id': osm.uuid,
'name': osm.name,
'tags': [tag.tag for tag in osm.tags.all()],
'deployed_count': osm.deployedServices.count(),
'type': type_.mod_type(),
'type_name': type_.mod_name(),
'servicesTypes': [
type_.services_types
], # A list for backward compatibility. TODO: To be removed when admin interface is changed
'comments': osm.comments,
'permission': permissions.effective_permissions(self._user, osm),
}
# Fill type and type_name
OsManagers.fill_instance_type(osm, ret_value)
return ret_value
def item_as_dict(self, item: 'Model') -> OsManagerItem:
item = ensure.is_instance(item, OSManager)

28
server/src/uds/REST/methods/permissions.py
View File

@@ -94,21 +94,19 @@ class Permissions(Handler):
entity = perm.user
# If entity is None, it means that the permission is not valid anymore (user or group deleted on db manually?)
if not entity:
continue
res.append(
{
'id': perm.uuid,
'type': kind,
'auth': entity.manager.uuid,
'auth_name': entity.manager.name,
'entity_id': entity.uuid,
'entity_name': entity.name,
'perm': perm.permission,
'perm_name': perm.as_str,
}
)
if entity:
res.append(
{
'id': perm.uuid,
'type': kind,
'auth': entity.manager.uuid,
'auth_name': entity.manager.name,
'entity_id': entity.uuid,
'entity_name': entity.name,
'perm': perm.permission,
'perm_name': perm.as_str,
}
)
return sorted(res, key=lambda v: v['auth_name'] + v['entity_name'])

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

@@ -61,20 +61,18 @@ class OfferItem(types.rest.ItemDictType):
description: str
icon: str
class ProviderItem(types.rest.ManagedObjectDictType):
id: str
name: str
tags: list[str]
services_count: int
user_services_count: int
maintenance_mode: bool
offers: list[OfferItem]
comments: str
permission: types.permissions.PermissionType
class Providers(ModelHandler):
class ProviderItem(types.rest.ItemDictType):
id: str
name: str
tags: list[str]
services_count: int
user_services_count: int
maintenance_mode: bool
offers: list[OfferItem]
type: str
type_name: str
comments: str
permission: types.permissions.PermissionType
class Providers(ModelHandler[ProviderItem]):
model = Provider
detail = {'services': DetailServices, 'usage': ServicesUsage}
@@ -117,7 +115,7 @@ class Providers(ModelHandler):
for t in type_.get_provided_services()
]
return {
val: ProviderItem = {
'id': item.uuid,
'name': item.name,
'tags': [tag.vtag for tag in item.tags.all()],
@@ -133,6 +131,9 @@ class Providers(ModelHandler):
'permission': permissions.effective_permissions(self._user, item),
}
Providers.fill_instance_type(item, val)
return val
def validate_delete(self, item: 'Model') -> None:
item = ensure.is_instance(item, Provider)
if item.services.count() > 0:

20
server/src/uds/REST/methods/reports.py
View File

@@ -58,11 +58,21 @@ VALID_PARAMS = (
)
class ReportItem(types.rest.ItemDictType):
id: str
mime_type: str
encoded: bool
group: str
name: str
description: str
# Enclosed methods under /actor path
class Reports(model.BaseModelHandler):
class Reports(model.BaseModelHandler[ReportItem]):
"""
Processes reports requests
"""
min_access_role = consts.UserRole.ADMIN
table_title = _('Available reports')
@@ -75,7 +85,9 @@ class Reports(model.BaseModelHandler):
# Field from where to get "class" and prefix for that class, so this will generate "row-state-A, row-state-X, ....
table_row_style = types.ui.RowStyleInfo(prefix='row-state-', field='state')
def _locate_report(self, uuid: str, values: typing.Optional[typing.Dict[str, typing.Any]] = None) -> 'Report':
def _locate_report(
self, uuid: str, values: typing.Optional[typing.Dict[str, typing.Any]] = None
) -> 'Report':
found = None
logger.debug('Looking for report %s', uuid)
for i in reports.available_reports:
@@ -149,9 +161,7 @@ class Reports(model.BaseModelHandler):
return sorted(report.gui_description(), key=lambda f: f['gui']['order'])
# Returns the list of
def get_items(
self, *args: typing.Any, **kwargs: typing.Any
) -> typing.Generator[types.rest.ItemDictType, None, None]:
def get_items(self, *args: typing.Any, **kwargs: typing.Any) -> typing.Generator[ReportItem, None, None]:
for i in reports.available_reports:
yield {
'id': i.get_uuid(),

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

@@ -48,21 +48,22 @@ if typing.TYPE_CHECKING:
logger = logging.getLogger(__name__)
class TokenItem(types.rest.ItemDictType):
id: str
name: str
stamp: datetime.datetime
username: str
ip: str
hostname: str
listen_port: int
mac: str
token: str
type: str
os: str
# REST API for Server Tokens management (for admin interface)
class ServersTokens(ModelHandler):
class TokenItem(types.rest.ItemDictType):
id: str
name: str
stamp: datetime.datetime
username: str
ip: str
hostname: str
listen_port: int
mac: str
token: str
type: str
os: str
class ServersTokens(ModelHandler[TokenItem]):
# servers/groups/[id]/servers
model = models.Server
@@ -120,28 +121,29 @@ class ServersTokens(ModelHandler):
return consts.OK
class ServerItem(types.rest.ItemDictType):
id: str
hostname: str
ip: str
listen_port: int
mac: str
maintenance_mode: bool
register_username: str
stamp: datetime.datetime
# REST API For servers (except tunnel servers nor actors)
class ServersServers(DetailHandler):
class ServerItem(types.rest.ItemDictType):
id: str
hostname: str
ip: str
listen_port: int
mac: str
maintenance_mode: bool
register_username: str
stamp: datetime.datetime
class ServersServers(DetailHandler[ServerItem]):
custom_methods = ['maintenance', 'importcsv']
def get_items(self, parent: 'Model', item: typing.Optional[str]) -> types.rest.GetItemsResult:
def get_items(self, parent: 'Model', item: typing.Optional[str]) -> types.rest.GetItemsResult[ServerItem]:
parent = typing.cast('models.ServerGroup', parent) # We will receive for sure
try:
if item is None:
q = parent.servers.all()
else:
q = parent.servers.filter(uuid=process_uuid(item))
res: list[ServersServers.ServerItem] = []
res: list[ServerItem] = []
i = None
for i in q:
res.append(
@@ -157,10 +159,10 @@ class ServersServers(DetailHandler):
}
)
if item is None:
return typing.cast(types.rest.GetItemsResult, res)
return res
if not i:
raise Exception('Item not found')
return typing.cast(types.rest.GetItemsResult, res[0])
raise Exception('Item not found') # Threated on the except below
return res[0]
except Exception as e:
logger.exception('REST servers')
raise self.invalid_item_response() from e
@@ -432,18 +434,19 @@ class ServersServers(DetailHandler):
return import_errors
class GroupItem(types.rest.ItemDictType):
id: str
name: str
comments: str
type: str
subtype: str
type_name: str
tags: list[str]
servers_count: int
permission: types.permissions.PermissionType
class ServersGroups(ModelHandler):
class GroupItem(types.rest.ItemDictType):
id: str
name: str
comments: str
type: str
subtype: str
type_name: str
tags: list[str]
servers_count: int
permission: types.permissions.PermissionType
class ServersGroups(ModelHandler[GroupItem ]):
custom_methods = [
types.rest.ModelCustomMethod('stats', True),

71
server/src/uds/REST/methods/services.py
View File

@@ -58,7 +58,43 @@ if typing.TYPE_CHECKING:
logger = logging.getLogger(__name__)
class Services(DetailHandler): # pylint: disable=too-many-public-methods
class ServiceItem(types.rest.ManagedObjectDictType):
id: str
name: str
tags: list[str]
comments: str
deployed_services_count: int
user_services_count: int
max_services_count_type: str
maintenance_mode: bool
permission: int
info: typing.NotRequired['ServiceInfo']
class ServiceInfo(types.rest.ItemDictType):
icon: str
needs_publication: bool
max_deployed: int
uses_cache: bool
uses_cache_l2: bool
cache_tooltip: str
cache_tooltip_l2: str
needs_osmanager: bool
allowed_protocols: list[str]
services_type_provided: str
can_reset: bool
can_list_assignables: bool
class ServicePoolResumeItem(types.rest.ItemDictType):
id: str
name: str
thumb: str
user_services_count: int
state: str
class Services(DetailHandler[ServiceItem]): # pylint: disable=too-many-public-methods
"""
Detail handler for Services, whose parent is a Provider
"""
@@ -66,7 +102,7 @@ class Services(DetailHandler): # pylint: disable=too-many-public-methods
custom_methods = ['servicepools']
@staticmethod
def service_info(item: models.Service) -> dict[str, typing.Any]:
def service_info(item: models.Service) -> ServiceInfo:
info = item.get_type()
overrided_fields = info.overrided_pools_fields or {}
@@ -79,28 +115,24 @@ class Services(DetailHandler): # pylint: disable=too-many-public-methods
'cache_tooltip': _(info.cache_tooltip),
'cache_tooltip_l2': _(info.cache_tooltip_l2),
'needs_osmanager': info.needs_osmanager,
'allowed_protocols': info.allowed_protocols,
'allowed_protocols': [str(i) for i in info.allowed_protocols],
'services_type_provided': info.services_type_provided,
'can_reset': info.can_reset,
'can_list_assignables': info.can_assign(),
}
@staticmethod
def service_to_dict(item: models.Service, perm: int, full: bool = False) -> types.rest.ItemDictType:
def service_to_dict(item: models.Service, perm: int, full: bool = False) -> ServiceItem:
"""
Convert a service db item to a dict for a rest response
:param item: Service item (db)
:param full: If full is requested, add "extra" fields to complete information
"""
item_type = item.get_type()
ret_value: dict[str, typing.Any] = {
ret_value: ServiceItem = {
'id': item.uuid,
'name': item.name,
'tags': [tag.tag for tag in item.tags.all()],
'comments': item.comments,
'type': item.data_type, # Compat with old code
'data_type': item.data_type,
'type_name': _(item_type.mod_name()),
'deployed_services_count': item.deployedServices.count(),
'user_services_count': models.UserService.objects.filter(deployed_service__service=item)
.exclude(state__in=State.INFO_STATES)
@@ -109,12 +141,14 @@ class Services(DetailHandler): # pylint: disable=too-many-public-methods
'maintenance_mode': item.provider.maintenance_mode,
'permission': perm,
}
Services.fill_instance_type(item, ret_value)
if full:
ret_value['info'] = Services.service_info(item)
return ret_value
def get_items(self, parent: 'Model', item: typing.Optional[str]) -> types.rest.GetItemsResult:
def get_items(self, parent: 'Model', item: typing.Optional[str]) -> types.rest.GetItemsResult[ServiceItem]:
parent = ensure.is_instance(parent, models.Provider)
# Check what kind of access do we have to parent provider
perm = permissions.effective_permissions(self._user, parent)
@@ -123,7 +157,9 @@ class Services(DetailHandler): # pylint: disable=too-many-public-methods
return [Services.service_to_dict(k, perm) for k in parent.services.all()]
k = parent.services.get(uuid=process_uuid(item))
val = Services.service_to_dict(k, perm, full=True)
return self.fill_instance_fields(k, val)
# On detail, ne wee to fill the instance fields by hand
self.fill_instance_fields(k, val)
return val
except Exception as e:
logger.error('Error getting services for %s: %s', parent, e)
raise self.invalid_item_response(repr(e)) from e
@@ -141,7 +177,7 @@ class Services(DetailHandler): # pylint: disable=too-many-public-methods
except Exception: # nosec: This is a delete, we don't care about exceptions
pass
def save_item(self, parent: 'Model', item: typing.Optional[str]) -> typing.Any:
def save_item(self, parent: 'Model', item: typing.Optional[str]) -> ServiceItem:
parent = ensure.is_instance(parent, models.Provider)
# Extract item db fields
# We need this fields for all
@@ -188,7 +224,10 @@ class Services(DetailHandler): # pylint: disable=too-many-public-methods
service.data = service_instance.serialize()
service.save()
return {'id': service.uuid}
return Services.service_to_dict(
service, permissions.effective_permissions(self._user, service), full=True
)
except models.Service.DoesNotExist:
raise self.invalid_item_response() from None
except IntegrityError as e: # Duplicate key probably
@@ -293,7 +332,7 @@ class Services(DetailHandler): # pylint: disable=too-many-public-methods
parent_instance = parent.get_instance()
service_type = parent_instance.get_service_by_type(for_type)
if not service_type:
raise self.invalid_item_response(f'Gui for {for_type} not found')
raise self.invalid_item_response(f'Gui for type "{for_type}" not found')
with Environment.temporary_environment() as env:
service = service_type(
env, parent_instance
@@ -335,11 +374,11 @@ class Services(DetailHandler): # pylint: disable=too-many-public-methods
except Exception:
raise self.invalid_item_response() from None
def servicepools(self, parent: 'Model', item: str) -> types.rest.GetItemsResult:
def servicepools(self, parent: 'Model', item: str) -> list[ServicePoolResumeItem]:
parent = ensure.is_instance(parent, models.Provider)
service = parent.services.get(uuid=process_uuid(item))
logger.debug('Got parameters for servicepools: %s, %s', parent, item)
res: types.rest.ItemListType = []
res: list[ServicePoolResumeItem] = []
for i in service.deployedServices.all():
try:
self.ensure_has_access(

50
server/src/uds/REST/methods/services_pool_groups.py
View File

@@ -53,21 +53,16 @@ from uds.core.ui import gui
# Enclosed methods under /item path
class ServicesPoolGroups(ModelHandler):
class ServicePoolGroupItem(types.rest.ItemDictType):
id: str
name: str
comments: str
priority: int
image_id: str|None
class ServicePoolGroupItem(types.rest.ItemDictType):
id: str
name: str
comments: str
priority: int
image_id: typing.NotRequired[str | None]
thumb: typing.NotRequired[str]
class ServicePoolGroupItemOverview(types.rest.ItemDictType):
id: str
name: str
priority: int
comments: str
thumb: str
class ServicesPoolGroups(ModelHandler[ServicePoolGroupItem]):
path = 'gallery'
model = ServicePoolGroup
@@ -104,20 +99,17 @@ class ServicesPoolGroups(ModelHandler):
local_gui = self.add_default_fields([], ['name', 'comments', 'priority'])
for field in [
{
'name': 'image_id',
'choices': [gui.choice_image(-1, '--------', DEFAULT_THUMB_BASE64)]
+ gui.sorted_choices(
[
gui.choice_image(v.uuid, v.name, v.thumb64)
for v in Image.objects.all()
]
),
'label': gettext('Associated Image'),
'tooltip': gettext('Image assocciated with this service'),
'type': types.ui.FieldType.IMAGECHOICE,
'order': 102,
}
{
'name': 'image_id',
'choices': [gui.choice_image(-1, '--------', DEFAULT_THUMB_BASE64)]
+ gui.sorted_choices(
[gui.choice_image(v.uuid, v.name, v.thumb64) for v in Image.objects.all()]
),
'label': gettext('Associated Image'),
'tooltip': gettext('Image assocciated with this service'),
'type': types.ui.FieldType.IMAGECHOICE,
'order': 102,
}
]:
self.add_field(local_gui, field)
@@ -133,9 +125,7 @@ class ServicesPoolGroups(ModelHandler):
'image_id': item.image.uuid if item.image else None,
}
def item_as_dict_overview(
self, item: 'Model'
) -> ServicePoolGroupItemOverview:
def item_as_dict_overview(self, item: 'Model') -> ServicePoolGroupItem:
item = ensure.is_instance(item, ServicePoolGroup)
return {
'id': item.uuid,

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

@@ -50,7 +50,7 @@ from uds.models import Account, Image, OSManager, Service, ServicePool, ServiceP
from uds.REST.model import ModelHandler
from .op_calendars import AccessCalendars, ActionsCalendars
from .services import Services
from .services import Services, ServiceInfo
from .user_services import AssignedUserService, CachedService, Changelog, Groups, Publications, Transports
if typing.TYPE_CHECKING:
@@ -59,7 +59,50 @@ if typing.TYPE_CHECKING:
logger = logging.getLogger(__name__)
class ServicesPools(ModelHandler):
class ServicePoolItem(types.rest.ItemDictType):
id: str
name: str
short_name: str
tags: typing.List[str]
parent: str
parent_type: str
comments: str
state: str
thumb: str
account: str
account_id: str | None
service_id: str
provider_id: str
image_id: str | None
initial_srvs: int
cache_l1_srvs: int
cache_l2_srvs: int
max_srvs: int
show_transports: bool
visible: bool
allow_users_remove: bool
allow_users_reset: bool
ignores_unused: bool
fallbackAccess: str
meta_member: list[dict[str, str]]
calendar_message: str
custom_message: str
display_custom_message: bool
osmanager_id: str | None
user_services_count: typing.NotRequired[int]
user_services_in_preparation: typing.NotRequired[int]
user_services_in_preparation: typing.NotRequired[int]
restrained: typing.NotRequired[bool]
permission: typing.NotRequired[int]
info: typing.NotRequired[ServiceInfo]
pool_group_id: typing.NotRequired[str | None]
pool_group_name: typing.NotRequired[str]
pool_group_thumb: typing.NotRequired[str]
usage: typing.NotRequired[str]
class ServicesPools(ModelHandler[ServicePoolItem]):
"""
Handles Services Pools REST requests
"""
@@ -131,7 +174,7 @@ class ServicesPools(ModelHandler):
def get_items(
self, *args: typing.Any, **kwargs: typing.Any
) -> typing.Generator[types.rest.ItemDictType, None, None]:
) -> typing.Generator[ServicePoolItem, None, None]:
# Optimized query, due that there is a lot of info needed for theee
d = sql_now() - datetime.timedelta(seconds=GlobalConfig.RESTRAINT_TIME.as_int())
return super().get_items(
@@ -178,49 +221,7 @@ class ServicesPools(ModelHandler):
# return super().get_items(overview=kwargs.get('overview', True), prefetch=['service', 'service__provider', 'servicesPoolGroup', 'image', 'tags'])
# return super(ServicesPools, self).get_items(*args, **kwargs)
class SummaryItem(types.rest.ItemDictType):
id: str
name: str
short_name: str
tags: typing.List[str]
parent: str
parent_type: str
comments: str
state: str
thumb: str
account: str
account_id: str | None
service_id: str
provider_id: str
image_id: str | None
initial_srvs: int
cache_l1_srvs: int
cache_l2_srvs: int
max_srvs: int
show_transports: bool
visible: bool
allow_users_remove: bool
allow_users_reset: bool
ignores_unused: bool
fallbackAccess: str
meta_member: list[dict[str, str]]
calendar_message: str
custom_message: str
display_custom_message: bool
osmanager_id: str | None
class FullItem(SummaryItem):
user_services_count: int
user_services_in_preparation: int
restrained: bool
permission: int
info: dict[str, typing.Any]
pool_group_id: str | None
pool_group_name: str
pool_group_thumb: str
usage: str
def item_as_dict(self, item: 'Model') -> SummaryItem | FullItem:
def item_as_dict(self, item: 'Model') -> ServicePoolItem:
item = ensure.is_instance(item, ServicePool)
summary = 'summarize' in self._params
# if item does not have an associated service, hide it (the case, for example, for a removed service)
@@ -241,7 +242,7 @@ class ServicesPools(ModelHandler):
# This needs a lot of queries, and really does not apport anything important to the report
# elif UserServiceManager.manager().canInitiateServiceFromDeployedService(item) is False:
# state = State.SLOWED_DOWN
val: ServicesPools.SummaryItem = {
val: ServicePoolItem = {
'id': item.uuid,
'name': item.name,
'short_name': item.short_name,
@@ -277,9 +278,6 @@ class ServicesPools(ModelHandler):
if summary:
return val
# Recast to complete data
val = typing.cast(ServicesPools.FullItem, val)
if hasattr(item, 'valid_count'):
valid_count = getattr(item, 'valid_count')
preparing_count = getattr(item, 'preparing_count')

35
server/src/uds/REST/methods/transports.py
View File

@@ -51,25 +51,24 @@ logger = logging.getLogger(__name__)
# Enclosed methods under /item path
class TransportItem(types.rest.ManagedObjectDictType):
id: str
name: str
tags: list[str]
comments: str
priority: int
label: str
net_filtering: str
networks: list[str]
allowed_oss: list[str]
pools: list[str]
pools_count: int
deployed_count: int
protocol: str
permission: int
class Transports(ModelHandler):
class TransportItem(types.rest.ItemDictType):
id: str
name: str
tags: list[str]
comments: str
priority: int
label: str
net_filtering: str
networks: list[str]
allowed_oss: list[str]
pools: list[str]
pools_count: int
deployed_count: int
type: str
type_name: str
protocol: str
permission: int
class Transports(ModelHandler[TransportItem]):
model = Transport
save_fields = [

33
server/src/uds/REST/model/base.py
View File

@@ -55,7 +55,7 @@ logger = logging.getLogger(__name__)
# pylint: disable=unused-argument
class BaseModelHandler(Handler):
class BaseModelHandler(Handler, typing.Generic[types.rest.T_Item]):
"""
Base Handler for Master & Detail Handlers
"""
@@ -329,7 +329,24 @@ class BaseModelHandler(Handler):
return args
def fill_instance_fields(self, item: 'models.Model', dct: types.rest.ItemDictType) -> None:
@staticmethod
def fill_instance_type(item: 'models.Model', dct: types.rest.ManagedObjectDictType) -> None:
"""
For Managed Objects (db element that contains a serialized object), fills a dictionary with the "type" and "type_name" parameters values.
For non managed objects, it does nothing
Args:
item: Item to fill type
dct: Dictionary to fill with type
"""
if isinstance(item, ManagedObjectModel):
kind = item.get_type()
typing.cast(dict[str, typing.Any], dct)['type'] = kind.mod_type()
typing.cast(dict[str, typing.Any], dct)['type_name'] = kind.mod_name()
@staticmethod
def fill_instance_fields(item: 'models.Model', dct: types.rest.ItemDictType) -> None:
"""
For Managed Objects (db element that contains a serialized object), fills a dictionary with the "field" parameters values.
For non managed objects, it does nothing
@@ -341,11 +358,19 @@ class BaseModelHandler(Handler):
"""
# Cast to allow override typing
res = typing.cast(dict[str, typing.Any], dct)
if isinstance(item, ManagedObjectModel):
res = typing.cast(types.rest.ManagedObjectDictType, dct)
i = item.get_instance()
i.init_gui() # Defaults & stuff
res.update(i.get_fields_as_dict())
fields = i.get_fields_as_dict()
# TODO: This will be removed in future versions, as it will be overseed by "instance" key
typing.cast(typing.Any, res).update(fields) # Add fields to dict
res['type'] = i.mod_type() # Add type
res['type_name'] = i.mod_name() # Add type name
# Future inmplementation wil insert instace fields into "instance" key
# For now, just repeat the fields
res['instance'] = fields
# Exceptions
def invalid_request_response(self, message: typing.Optional[str] = None) -> exceptions.rest.HandlerError:

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

@@ -57,7 +57,7 @@ logger = logging.getLogger(__name__)
# Details do not have types at all
# so, right now, we only process details petitions for Handling & tables info
# noinspection PyMissingConstructor
class DetailHandler(BaseModelHandler, typing.Generic[types.rest.T_Item]):
class DetailHandler(BaseModelHandler[types.rest.T_Item], typing.Generic[types.rest.T_Item]):
"""
Detail handler (for relations such as provider-->services, authenticators-->users,groups, deployed services-->cache,assigned, groups, transports
Urls recognized for GET are:

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

@@ -54,7 +54,7 @@ if typing.TYPE_CHECKING:
logger = logging.getLogger(__name__)
class ModelHandler(BaseModelHandler, typing.Generic[types.rest.T_Item]):
class ModelHandler(BaseModelHandler[types.rest.T_Item], typing.Generic[types.rest.T_Item]):
"""
Basic Handler for a model
Basically we will need same operations for all models, so we can

9
server/src/uds/core/types/rest/__init__.py
View File

@@ -112,6 +112,15 @@ class ItemDictType(typing.TypedDict):
pass
class ManagedObjectDictType(typing.TypedDict):
"""
Represents a managed object type, with its name and type.
This is used to represent the type of a managed object in the REST API.
"""
type: typing.NotRequired[str] # Type of the managed object
type_name: typing.NotRequired[str] # Name of the type of the managed object
instance: typing.NotRequired[typing.Any] # Instance of the managed object, if available
# Alias for item type
# ItemDictType = dict[str, typing.Any]
T_Item = typing.TypeVar("T_Item", bound=ItemDictType)