120 lines
3.7 KiB
Python
120 lines
3.7 KiB
Python
from typing import Optional, Type
|
|
from pydantic import BaseModel
|
|
|
|
|
|
class EventCluster:
|
|
"""
|
|
EventCluster
|
|
"""
|
|
|
|
def __init__(self, endpoint_uu_id: str):
|
|
self.endpoint_uu_id = endpoint_uu_id
|
|
self.events = []
|
|
|
|
def add_event(self, list_of_events: list["Event"]):
|
|
"""
|
|
Add an event to the cluster
|
|
"""
|
|
for event in list_of_events:
|
|
self.events.append(event)
|
|
self.events = list(set(self.events))
|
|
|
|
def get_event(self, event_key: str):
|
|
"""
|
|
Get an event by its key
|
|
"""
|
|
|
|
for event in self.events:
|
|
if event.key == event_key:
|
|
return event
|
|
return None
|
|
|
|
def set_events_to_database(self):
|
|
from Schemas import Events, EndpointRestriction
|
|
|
|
with Events.new_session() as db_session:
|
|
if to_save_endpoint := EndpointRestriction.filter_one(
|
|
EndpointRestriction.operation_uu_id == self.endpoint_uu_id,
|
|
db=db_session,
|
|
).data:
|
|
for event in self.events:
|
|
event_to_save_database = Events.find_or_create(
|
|
function_code=event.key,
|
|
function_class=event.name,
|
|
description=event.description,
|
|
endpoint_code=self.endpoint_uu_id,
|
|
endpoint_id=to_save_endpoint.id,
|
|
endpoint_uu_id=str(to_save_endpoint.uu_id),
|
|
is_confirmed=True,
|
|
active=True,
|
|
db=db_session,
|
|
)
|
|
if event_to_save_database.meta_data.created:
|
|
event_to_save_database.save(db=db_session)
|
|
print(
|
|
f"UUID: {event_to_save_database.uu_id} event is saved to {to_save_endpoint.uu_id}"
|
|
)
|
|
|
|
def match_event(self, event_key: str) -> "Event":
|
|
"""
|
|
Match an event by its key
|
|
"""
|
|
# print('set(event_keys)', set(event_keys))
|
|
# print('event.keys', set([event.key for event in self.events]))
|
|
# intersection_of_key: set[str] = set(event_key) & set([event.key for event in self.events])
|
|
# if not len(intersection_of_key) == 1:
|
|
# raise ValueError(
|
|
# f"Event key not found or multiple matches found: {intersection_of_key}"
|
|
# )
|
|
return self.get_event(event_key=event_key)
|
|
|
|
|
|
class Event:
|
|
|
|
def __init__(
|
|
self,
|
|
name: str,
|
|
key: str,
|
|
request_validator: Optional[Type[BaseModel]] = None,
|
|
response_validator: Optional[Type[BaseModel]] = None,
|
|
description: str = "",
|
|
):
|
|
self.name = name
|
|
self.key = key
|
|
self.request_validator = request_validator
|
|
self.response_validator = response_validator
|
|
self.description = description
|
|
|
|
def event_callable(self):
|
|
"""
|
|
Example callable method
|
|
"""
|
|
print(self.name)
|
|
return {}
|
|
|
|
|
|
class SetEventCluster:
|
|
"""
|
|
SetEventCluster
|
|
"""
|
|
|
|
list_of_event_clusters: list[EventCluster] = []
|
|
|
|
def add_event_cluster(self, event_cluster: EventCluster):
|
|
"""
|
|
Add an event cluster to the set
|
|
"""
|
|
endpoint_uu_id_list = [
|
|
event_cluster_uuid.endpoint_uu_id
|
|
for event_cluster_uuid in self.list_of_event_clusters
|
|
]
|
|
if event_cluster.endpoint_uu_id not in endpoint_uu_id_list:
|
|
self.list_of_event_clusters.append(event_cluster)
|
|
|
|
@property
|
|
def retrieve_all_event_clusters(self) -> list[EventCluster]:
|
|
"""
|
|
Retrieve all event clusters
|
|
"""
|
|
return self.list_of_event_clusters
|