cengal.parallel_execution.coroutines.coro_standard_services.remote_nodes.versions.v_0.request_class_info

Module Docstring Docstrings: http://www.python.org/dev/peps/pep-0257/

View Source

  1#!/usr/bin/env python
  2# coding=utf-8
  3
  4# Copyright © 2012-2024 ButenkoMS. All rights reserved. Contacts: <gtalk@butenkoms.space>
  5# 
  6# Licensed under the Apache License, Version 2.0 (the "License");
  7# you may not use this file except in compliance with the License.
  8# You may obtain a copy of the License at
  9# 
 10#     http://www.apache.org/licenses/LICENSE-2.0
 11# 
 12# Unless required by applicable law or agreed to in writing, software
 13# distributed under the License is distributed on an "AS IS" BASIS,
 14# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 15# See the License for the specific language governing permissions and
 16# limitations under the License.
 17
 18
 19"""
 20Module Docstring
 21Docstrings: http://www.python.org/dev/peps/pep-0257/
 22"""
 23
 24
 25__author__ = "ButenkoMS <gtalk@butenkoms.space>"
 26__copyright__ = "Copyright © 2012-2024 ButenkoMS. All rights reserved. Contacts: <gtalk@butenkoms.space>"
 27__credits__ = ["ButenkoMS <gtalk@butenkoms.space>", ]
 28__license__ = "Apache License, Version 2.0"
 29__version__ = "4.4.1"
 30__maintainer__ = "ButenkoMS <gtalk@butenkoms.space>"
 31__email__ = "gtalk@butenkoms.space"
 32# __status__ = "Prototype"
 33__status__ = "Development"
 34# __status__ = "Production"
 35
 36
 37from enum import Enum
 38from cengal.parallel_execution.coroutines.coro_scheduler import *
 39from cengal.parallel_execution.coroutines.coro_tools.await_coro import *
 40from cengal.parallel_execution.coroutines.coro_standard_services.asyncio_loop import *
 41from cengal.parallel_execution.coroutines.coro_standard_services.loop_yield import CoroPriority
 42from cengal.parallel_execution.coroutines.coro_standard_services.put_coro import *
 43from cengal.parallel_execution.coroutines.coro_standard_services.timer_func_runner import *
 44from cengal.file_system.file_manager import path_relative_to_current_dir
 45from cengal.time_management.cpu_clock_cycles import perf_counter
 46from cengal.data_manipulation.serialization import *
 47from typing import Hashable, Tuple, List, Any, Dict, Callable, Type
 48from cengal.introspection.inspect import get_exception, entity_owning_module_importable_str, entity_owning_module_info_and_owning_path, entity_properties
 49from cengal.io.core.memory_management import IOCoreMemoryManagement
 50from cengal.parallel_execution.asyncio.efficient_streams import StreamManagerIOCoreMemoryManagement, TcpStreamManager, UdpStreamManager, StreamManagerAbstract
 51from cengal.code_flow_control.smart_values import ValueExistence
 52from cengal.io.named_connections.named_connections_manager import NamedConnectionsManager
 53from cengal.code_flow_control.args_manager import number_of_provided_args
 54from cengal.data_manipulation.serialization import Serializer, Serializers, best_serializer_for_standard_data
 55from cengal.code_flow_control.args_manager import find_arg_position_and_value, UnknownArgumentError
 56from cengal.data_generation.id_generator import IDGenerator, GeneratorType
 57from cengal.system import PLATFORM_NAME, PYTHON_VERSION
 58from importlib import import_module
 59import sys
 60import os
 61import asyncio
 62
 63from .exceptions import *
 64from .commands import *
 65from .class_info import *
 66
 67
 68class LocalRequestClassInfo(LocalClassInfo):
 69    def __init__(self, local_id: Hashable, request: Request) -> None:
 70        super().__init__(local_id, type(request))
 71        self._properties: Dict[str, Hashable] = {property_name: index for index, property_name in enumerate(entity_properties(request))}  # key: property name; value: property id
 72        self._properties_tuple: Tuple[Tuple[str, Hashable]] = tuple(self._properties.items())
 73    
 74    def __call__(self) -> Type:
 75        return {
 76            CommandDataFieldsDeclareServiceRequestClass.local_id.value: self._local_id,
 77            CommandDataFieldsDeclareServiceRequestClass.class_name.value: self._class_name,
 78            CommandDataFieldsDeclareServiceRequestClass.module_importable_str.value: self._module_importable_str,
 79            CommandDataFieldsDeclareServiceRequestClass.properties_tuple.value: self._properties_tuple,
 80        }
 81    
 82    @property
 83    def properties(self):
 84        return self._properties
 85    
 86    @property
 87    def properties_tuple(self):
 88        return self._properties_tuple
 89    
 90    def request_to_data(self, request: Request) -> Dict:
 91        return {
 92            CommandDataFieldsServiceRequestWithRequestClass.request_class_id.value: self._local_id,
 93            CommandDataFieldsServiceRequestWithRequestClass.properties_tuple.value: tuple(((property_id, getattr(request, property_name)) for property_name, property_id in self._properties_tuple)),
 94        }
 95
 96
 97class RemoteRequestClassInfo(RemoteClassInfo):
 98    def __init__(self, local_id: Hashable, class_name: str, module_importable_str: str, properties_tuple: Tuple[Tuple[str, Hashable]]) -> None:
 99        super().__init__(local_id, class_name, module_importable_str)
100        self._properties_tuple: Tuple[Tuple[str, Hashable]] = properties_tuple
101        self._properties: Dict[Hashable, str] = {index: property_name for property_name, index in properties_tuple}  # key: property id; value: property name
102    
103    @classmethod
104    def from_data(cls, data: Dict[Hashable, Any]) -> 'RemoteRequestClassInfo':
105        local_id: Hashable = data[CommandDataFieldsDeclareServiceRequestClass.local_id.value]
106        class_name: str = data[CommandDataFieldsDeclareServiceRequestClass.class_name.value]
107        module_importable_str: str = data[CommandDataFieldsDeclareServiceRequestClass.module_importable_str.value]
108        properties_tuple: Tuple[Tuple[str, Hashable]] = data[CommandDataFieldsDeclareServiceRequestClass.properties_tuple.value]
109        return cls(local_id, class_name, module_importable_str, properties_tuple)
110    
111    def __call__(self, data: Dict) -> Request:
112        request: Request = self.class_type()
113        properties_tuple: Tuple[Tuple[Hashable, Any]] = data[CommandDataFieldsDeclareServiceRequestClass.properties_tuple.value]
114        for index, value in properties_tuple:
115            name: str = self._properties[index]
116            setattr(request, name, value)
117        
118        return request

class LocalRequestClassInfo(cengal.parallel_execution.coroutines.coro_standard_services.remote_nodes.versions.v_0.class_info.LocalClassInfo): View Source

69class LocalRequestClassInfo(LocalClassInfo):
70    def __init__(self, local_id: Hashable, request: Request) -> None:
71        super().__init__(local_id, type(request))
72        self._properties: Dict[str, Hashable] = {property_name: index for index, property_name in enumerate(entity_properties(request))}  # key: property name; value: property id
73        self._properties_tuple: Tuple[Tuple[str, Hashable]] = tuple(self._properties.items())
74    
75    def __call__(self) -> Type:
76        return {
77            CommandDataFieldsDeclareServiceRequestClass.local_id.value: self._local_id,
78            CommandDataFieldsDeclareServiceRequestClass.class_name.value: self._class_name,
79            CommandDataFieldsDeclareServiceRequestClass.module_importable_str.value: self._module_importable_str,
80            CommandDataFieldsDeclareServiceRequestClass.properties_tuple.value: self._properties_tuple,
81        }
82    
83    @property
84    def properties(self):
85        return self._properties
86    
87    @property
88    def properties_tuple(self):
89        return self._properties_tuple
90    
91    def request_to_data(self, request: Request) -> Dict:
92        return {
93            CommandDataFieldsServiceRequestWithRequestClass.request_class_id.value: self._local_id,
94            CommandDataFieldsServiceRequestWithRequestClass.properties_tuple.value: tuple(((property_id, getattr(request, property_name)) for property_name, property_id in self._properties_tuple)),
95        }

LocalRequestClassInfo( local_id: Hashable, request: cengal.parallel_execution.coroutines.coro_scheduler.versions.v_0.coro_scheduler.Request) View Source

70    def __init__(self, local_id: Hashable, request: Request) -> None:
71        super().__init__(local_id, type(request))
72        self._properties: Dict[str, Hashable] = {property_name: index for index, property_name in enumerate(entity_properties(request))}  # key: property name; value: property id
73        self._properties_tuple: Tuple[Tuple[str, Hashable]] = tuple(self._properties.items())

properties View Source

83    @property
84    def properties(self):
85        return self._properties

properties_tuple View Source

87    @property
88    def properties_tuple(self):
89        return self._properties_tuple

def request_to_data( self, request: cengal.parallel_execution.coroutines.coro_scheduler.versions.v_0.coro_scheduler.Request) -> Dict: View Source

91    def request_to_data(self, request: Request) -> Dict:
92        return {
93            CommandDataFieldsServiceRequestWithRequestClass.request_class_id.value: self._local_id,
94            CommandDataFieldsServiceRequestWithRequestClass.properties_tuple.value: tuple(((property_id, getattr(request, property_name)) for property_name, property_id in self._properties_tuple)),
95        }

Inherited Members

cengal.parallel_execution.coroutines.coro_standard_services.remote_nodes.versions.v_0.class_info.LocalClassInfo

class_type

class_name

module_importable_str

local_id

class RemoteRequestClassInfo(cengal.parallel_execution.coroutines.coro_standard_services.remote_nodes.versions.v_0.class_info.RemoteClassInfo): View Source

 98class RemoteRequestClassInfo(RemoteClassInfo):
 99    def __init__(self, local_id: Hashable, class_name: str, module_importable_str: str, properties_tuple: Tuple[Tuple[str, Hashable]]) -> None:
100        super().__init__(local_id, class_name, module_importable_str)
101        self._properties_tuple: Tuple[Tuple[str, Hashable]] = properties_tuple
102        self._properties: Dict[Hashable, str] = {index: property_name for property_name, index in properties_tuple}  # key: property id; value: property name
103    
104    @classmethod
105    def from_data(cls, data: Dict[Hashable, Any]) -> 'RemoteRequestClassInfo':
106        local_id: Hashable = data[CommandDataFieldsDeclareServiceRequestClass.local_id.value]
107        class_name: str = data[CommandDataFieldsDeclareServiceRequestClass.class_name.value]
108        module_importable_str: str = data[CommandDataFieldsDeclareServiceRequestClass.module_importable_str.value]
109        properties_tuple: Tuple[Tuple[str, Hashable]] = data[CommandDataFieldsDeclareServiceRequestClass.properties_tuple.value]
110        return cls(local_id, class_name, module_importable_str, properties_tuple)
111    
112    def __call__(self, data: Dict) -> Request:
113        request: Request = self.class_type()
114        properties_tuple: Tuple[Tuple[Hashable, Any]] = data[CommandDataFieldsDeclareServiceRequestClass.properties_tuple.value]
115        for index, value in properties_tuple:
116            name: str = self._properties[index]
117            setattr(request, name, value)
118        
119        return request

RemoteRequestClassInfo( local_id: Hashable, class_name: str, module_importable_str: str, properties_tuple: Tuple[Tuple[str, Hashable]]) View Source

 99    def __init__(self, local_id: Hashable, class_name: str, module_importable_str: str, properties_tuple: Tuple[Tuple[str, Hashable]]) -> None:
100        super().__init__(local_id, class_name, module_importable_str)
101        self._properties_tuple: Tuple[Tuple[str, Hashable]] = properties_tuple
102        self._properties: Dict[Hashable, str] = {index: property_name for property_name, index in properties_tuple}  # key: property id; value: property name

@classmethod

def from_data( cls, data: Dict[Hashable, Any]) -> RemoteRequestClassInfo: View Source

104    @classmethod
105    def from_data(cls, data: Dict[Hashable, Any]) -> 'RemoteRequestClassInfo':
106        local_id: Hashable = data[CommandDataFieldsDeclareServiceRequestClass.local_id.value]
107        class_name: str = data[CommandDataFieldsDeclareServiceRequestClass.class_name.value]
108        module_importable_str: str = data[CommandDataFieldsDeclareServiceRequestClass.module_importable_str.value]
109        properties_tuple: Tuple[Tuple[str, Hashable]] = data[CommandDataFieldsDeclareServiceRequestClass.properties_tuple.value]
110        return cls(local_id, class_name, module_importable_str, properties_tuple)

Inherited Members

cengal.parallel_execution.coroutines.coro_standard_services.remote_nodes.versions.v_0.class_info.RemoteClassInfo

local_id

class_name

module_importable_str

module

class_type