from typing import Any, List, Optional, cast
from LOGS.Auxiliary.Decorators import Endpoint
from LOGS.Auxiliary.Exceptions import LOGSException
from LOGS.Auxiliary.MinimalModelGenerator import MinimalModelGenerator
from LOGS.Auxiliary.Tools import Tools
from LOGS.Entities.EntitiesRequestParameter import EntitiesRequestParameter
from LOGS.Entity.Entity import Entity
from LOGS.Entity.EntityIterator import EntityIterator
from LOGS.Entity.EntityMinimal import EntityMinimal
from LOGS.Entity.EntityMinimalWithType import EntityMinimalWithType
from LOGS.Entity.EntityRequestParameter import EntityRequestParameter
from LOGS.Entity.ImportExportEvent import ImportExportEvent
from LOGS.Entity.SerializableContent import SerializableContent
from LOGS.LOGSConnection import LOGSConnection
[docs]
@Endpoint("entities")
class Entities(EntityIterator[Entity, EntityRequestParameter]):
"""LOGS connected multi Entity iterator"""
_generatorType = EntityMinimalWithType # type: ignore
_parameterType = EntitiesRequestParameter # type: ignore
_parametersInput: Optional[EntitiesRequestParameter]
def __init__(
self,
connection: Optional[LOGSConnection],
parameters: Optional[EntitiesRequestParameter] = None,
):
super().__init__(connection=connection, parameters=cast(Any, parameters))
self._parametersInput = parameters
def _convertToEntity(self, entity: EntityMinimalWithType):
return MinimalModelGenerator.MinimalFromSingle(
entity.toDict(), entity.type, None, connection=self._getConnection()
)
def __next__(self) -> EntityMinimalWithType:
entity = cast(EntityMinimalWithType, super().__next__())
result = self._convertToEntity(entity)
if not result:
raise LOGSException(
"Unknown entity %a of type %a." % (str(entity), entity.type)
)
return result
[docs]
def withType(self):
iterator = EntitiesWithType(
connection=self._getConnection(), parameters=self._parametersInput
)
iterator._listEndpoint = "list_with_events"
return iterator
[docs]
def fetch(self, uid: str):
self._entityIterator = 0
if not self._connection:
raise LOGSException(
"Entity connector %a is not connected" % type(self).__name__
)
if not self._endpoint:
raise NotImplementedError(
"Endpoint missing for of entity type %a."
% (
type(self).__name__
if type(self).__name__ != Entity.__name__
else "unknown"
)
)
ref, responseError = self._connection.getEndpoint(self._endpoint + [str(uid)])
if responseError:
raise LOGSException(
message="Could not fetch entity with uid %a: %s"
% (uid, responseError.errorString()),
responseError=responseError,
)
entity = Tools.checkAndConvert(ref, EntityMinimalWithType)
return self._convertToEntity(entity)
[docs]
class EntityWithEvents(SerializableContent):
_entity: EntityMinimal
_importExportEvents: Optional[List[ImportExportEvent]]
def __init__(
self,
entity: EntityMinimal,
importExportEvents: Optional[List[ImportExportEvent]],
):
self._entity = entity
self._importExportEvents = importExportEvents
@property
def entity(self) -> EntityMinimal:
return self._entity
@property
def importExportEvents(self) -> Optional[List[ImportExportEvent]]:
return self._importExportEvents
[docs]
class EntitiesWithType(Entities):
def _convertToEntity(self, entity: EntityMinimalWithType):
slack = entity._slack
# print (slack)
importExportList = (
[
ImportExportEvent(event, connection=self._getConnection())
for event in slack["importExportEvents"]
]
if "importExportEvents" in slack
else None
)
return EntityWithEvents(
MinimalModelGenerator.MinimalFromSingle(
entity.toDict(), entity.type, None, connection=self._getConnection()
),
importExportList,
)
def __next__(self):
return cast(EntityWithEvents, super().__next__())