253 lines
8.2 KiB
Python
253 lines
8.2 KiB
Python
"""
|
|
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
|