from __future__ import annotations from typing import TYPE_CHECKING, Dict from common.database.api.context.Keys import KEY_SERVICE, KEY_SERVICE_CONSTRAINTS, KEY_SERVICE_ENDPOINTS from common.database.api.context.service.Constraint import Constraint from common.database.api.context.service.Endpoint import Endpoint from common.database.api.context.service.ServiceState import ServiceState, to_servicestate_enum from common.database.api.context.service.ServiceType import ServiceType, to_servicetype_enum from common.database.api.entity._Entity import _Entity from common.database.api.entity.EntityCollection import EntityCollection if TYPE_CHECKING: from common.database.api.context.Context import Context VALIDATORS = { 'service_type': lambda v: v is not None and isinstance(v, ServiceType), 'service_config': lambda v: v is not None and isinstance(v, str) and (len(v) > 0), 'service_state': lambda v: v is not None and isinstance(v, ServiceState), } TRANSCODERS = { 'service_type': { ServiceType: lambda v: v.value, int : lambda v: to_servicetype_enum(v), str : lambda v: to_servicetype_enum(v), }, 'service_state': { ServiceState: lambda v: v.value, int : lambda v: to_servicestate_enum(v), str : lambda v: to_servicestate_enum(v), }, } class Service(_Entity): def __init__(self, service_uuid : str, parent : 'Context'): super().__init__(parent, service_uuid, KEY_SERVICE, VALIDATORS, TRANSCODERS) self._endpoints = EntityCollection(self, KEY_SERVICE_ENDPOINTS) self._constraints = EntityCollection(self, KEY_SERVICE_CONSTRAINTS) @property def parent(self) -> 'Context': return self._parent @property def context(self) -> 'Context': return self._parent @property def context_uuid(self) -> str: return self.context.context_uuid @property def service_uuid(self) -> str: return self._entity_uuid @property def endpoints(self) -> EntityCollection: return self._endpoints @property def constraints(self) -> EntityCollection: return self._constraints def endpoint(self, endpoint_uuid : str) -> Endpoint: return Endpoint(endpoint_uuid, self) def constraint(self, constraint_type : str) -> Constraint: return Constraint(constraint_type, self) def create(self, service_type : ServiceType, service_config : str, service_state : ServiceState) -> 'Service': self.update(update_attributes={ 'service_type': service_type, 'service_config': service_config, 'service_state': service_state, }) self.parent.services.add(self.service_uuid) return self def update(self, update_attributes={}, remove_attributes=[]) -> 'Service': self.attributes.update(update_attributes=update_attributes, remove_attributes=remove_attributes) return self def delete(self) -> None: for endpoint_uuid in self.endpoints.get(): self.endpoint(endpoint_uuid).delete() for constraint_uuid in self.constraints.get(): self.constraint(constraint_uuid).delete() self.attributes.delete() self.parent.services.delete(self.service_uuid) def dump_id(self) -> Dict: return { 'contextId': self.context.dump_id(), 'cs_id': {'uuid': self.service_uuid}, } def dump(self) -> Dict: attributes = self.attributes.get() service_type = attributes.get('service_type', None) if isinstance(service_type, ServiceType): service_type = service_type.value service_state = attributes.get('service_state', None) if isinstance(service_state, ServiceState): service_state = service_state.value service_config = attributes.get('device_config', None) endpoints = [self.endpoint(endpoint_uuid).dump() for endpoint_uuid in self.endpoints.get()] constraints = [self.constraint(constraint_type).dump() for constraint_type in self.constraints.get()] return { 'cs_id': self.dump_id(), 'serviceType': service_type, 'endpointList': endpoints, 'constraint': constraints, 'serviceState': {'serviceState': service_state}, 'serviceConfig': {'serviceConfig': service_config} }