""" Endpoint Structure Reference This file serves as a reference for the Endpoint Structure pattern used in EVYOS backend services. When addressed, this file helps to understand the pattern structure and implementation. The pattern consists of three main components: 1. Events Component 2. Endpoints Component 3. Validations Component Below is a simplified implementation example that demonstrates the structure. """ # ============================================================================ # EVENTS COMPONENT # ============================================================================ # cluster.py """ from ApiControllers.abstracts.event_clusters import EventCluster, RouterCluster from .supers_events import ( EntityListEvent, EntityCreateEvent, EntityUpdateEvent, ) # Create a router cluster to contain all event clusters EntityRouterCluster = RouterCluster(name="EntityRouterCluster") # Create event clusters for different operations EntityEventClusterList = EventCluster( name="EntityList", endpoint_uu_id="xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx" ) EntityEventClusterList.add_event(EntityListEvent) EntityEventClusterCreate = EventCluster( name="EntityCreate", endpoint_uu_id="xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx" ) EntityEventClusterCreate.add_event(EntityCreateEvent) EntityEventClusterUpdate = EventCluster( name="EntityUpdate", endpoint_uu_id="xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx" ) EntityEventClusterUpdate.add_event(EntityUpdateEvent) # Register event clusters with the router cluster EntityRouterCluster.set_event_cluster(EntityEventClusterList) EntityRouterCluster.set_event_cluster(EntityEventClusterCreate) EntityRouterCluster.set_event_cluster(EntityEventClusterUpdate) """ # supers_events.py """ from ApiControllers.abstracts.event_clusters import Event from Controllers.Postgres.pagination import Pagination, PaginationResult, PaginateOnly from Controllers.Postgres.response import EndpointResponse from Schemas import Entity from Validations.entity.entity.validations import ( REQUESTVALIDATIONMODEL, ) # Define events EntityListEvent = Event( name="entity_list", key="xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx", request_validator=None, # TODO: Add request validator response_validator=None, # TODO: Add response validator description="List events of entities endpoint", ) EntityCreateEvent = Event( name="entity_create", key="xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx", request_validator=None, # TODO: Add request validator response_validator=None, # TODO: Add response validator description="Create events of entities endpoint", ) EntityUpdateEvent = Event( name="entity_update", key="xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx", request_validator=None, # TODO: Add request validator response_validator=None, # TODO: Add response validator description="Update events of entities endpoint", ) # Define callable methods for events def entity_list_callable(list_options: PaginateOnly): """ Example callable method for list operation """ list_options = PaginateOnly(**list_options.model_dump()) with Entity.new_session() as db_session: if list_options.query: entities_list = Entity.filter_all( *Entity.convert(list_options.query), db=db_session ) else: entities_list = Entity.filter_all(db=db_session) pagination = Pagination(data=entities_list) pagination.change(**list_options.model_dump()) pagination_result = PaginationResult( data=entities_list, pagination=pagination, ).pagination.as_dict return EndpointResponse( message="MSG0003-LIST", pagination_result=pagination_result, request=REQUESTVALIDATIONMODEL, ).response # Assign callable methods to events EntityListEvent.event_callable = entity_list_callable def entity_create_callable(): """ Example callable method for create operation """ # Implementation here pass EntityCreateEvent.event_callable = entity_create_callable def entity_update_callable(): """ Example callable method for update operation """ # Implementation here pass EntityUpdateEvent.event_callable = entity_update_callable """ # ============================================================================ # ENDPOINTS COMPONENT # ============================================================================ # route.py """ from fastapi import APIRouter, Depends from ApiControllers.abstracts.default_validations import CommonHeaders from ApiControllers.providers.token_provider import TokenProvider from Controllers.Postgres.pagination import PaginateOnly from Events.entity.cluster import EntityRouterCluster # Create API router entity_route = APIRouter(prefix="/entity", tags=["Entity"]) @entity_route.post( path="/list", description="List entities endpoint", operation_id="xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx", ) def entity_list_route( data: PaginateOnly, headers: CommonHeaders = Depends(CommonHeaders.as_dependency), ): """ List entities endpoint """ token_object = TokenProvider.get_dict_from_redis(token=headers.token) event_founder_dict = dict(endpoint_code=headers.operation_id, token=token_object) event_key = TokenProvider.retrieve_event_codes(**event_founder_dict) FoundCluster = EntityRouterCluster.get_event_cluster("EntityList") event_cluster_matched = FoundCluster.match_event(event_key=event_key) return event_cluster_matched.event_callable(data=data) @entity_route.post( path="/create", description="Create entity endpoint", operation_id="xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx", ) def entity_create_route( data: dict, headers: CommonHeaders = Depends(CommonHeaders.as_dependency), ): """ Create entity endpoint """ token_object = TokenProvider.get_dict_from_redis(token=headers.token) event_founder_dict = dict(endpoint_code=headers.operation_id, token=token_object) event_key = TokenProvider.retrieve_event_codes(**event_founder_dict) FoundCluster = EntityRouterCluster.get_event_cluster("EntityCreate") event_cluster_matched = FoundCluster.match_event(event_key=event_key) return event_cluster_matched.event_callable(data=data) @entity_route.post( path="/update", description="Update entity endpoint", operation_id="xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx", ) def entity_update_route( data: dict, headers: CommonHeaders = Depends(CommonHeaders.as_dependency), ): """ Update entity endpoint """ token_object = TokenProvider.get_dict_from_redis(token=headers.token) event_founder_dict = dict(endpoint_code=headers.operation_id, token=token_object) event_key = TokenProvider.retrieve_event_codes(**event_founder_dict) FoundCluster = EntityRouterCluster.get_event_cluster("EntityUpdate") event_cluster_matched = FoundCluster.match_event(event_key=event_key) return event_cluster_matched.event_callable(data=data) """ # ============================================================================ # VALIDATIONS COMPONENT # ============================================================================ # validations.py """ from pydantic import BaseModel from typing import Optional class REQUESTVALIDATIONMODEL(BaseModel): field1: str field2: str field3: int field4: bool field5: Optional[str] = None # Additional fields as needed class RESPONSEVALIDATIONMODEL(BaseModel): # Response validation fields pass """ # ============================================================================ # IMPLEMENTATION GUIDE # ============================================================================ def create_endpoint_structure(service_name, entity_name): """ Function to create a new endpoint structure for a given service and entity. Args: service_name (str): Name of the service (e.g., "BuildingService") entity_name (str): Name of the entity (e.g., "building") Steps: 1. Create directory structure 2. Create validation models 3. Create events and callable methods 4. Create event clusters 5. Create API routes """ # Implementation would create the necessary files and directories pass