created design pattern

This commit is contained in:
2025-04-23 12:26:15 +03:00
parent e5f88f2eb4
commit 9511f81bc0
33 changed files with 1511 additions and 39 deletions

View File

@@ -0,0 +1,29 @@
FROM python:3.12-slim
WORKDIR /
# Install system dependencies and Poetry
RUN apt-get update && apt-get install -y --no-install-recommends gcc && rm -rf /var/lib/apt/lists/* && pip install --no-cache-dir poetry
# Copy Poetry configuration
COPY /pyproject.toml ./pyproject.toml
# Configure Poetry and install dependencies with optimizations
RUN poetry config virtualenvs.create false && poetry install --no-interaction --no-ansi --no-root --only main \
&& pip cache purge && rm -rf ~/.cache/pypoetry
# Copy application code
COPY /ApiControllers /ApiControllers
COPY /ApiDefaults /ApiDefaults
COPY /Controllers /Controllers
COPY /Schemas /Schemas
COPY /ApiServices/BuildingService/Endpoints /ApiDefaults/Endpoints
COPY /ApiServices/BuildingService/Events /ApiDefaults/Events
COPY /ApiServices/BuildingService/Validations /ApiDefaults/Validations
# Set Python path to include app directory
ENV PYTHONPATH=/ PYTHONUNBUFFERED=1 PYTHONDONTWRITEBYTECODE=1
# Run the application using the configured uvicorn server
CMD ["poetry", "run", "python", "ApiDefaults/app.py"]

View File

@@ -0,0 +1,70 @@
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.area.cluster import BuildAreaRouterCluster
area_route = APIRouter(prefix="/building/area", tags=["Building Area"])
@area_route.post(
path="/list",
description="List building area endpoint",
operation_id="f2831efc-5493-448c-bcf6-4d52d744ba7e",
)
def area_list_route(
data: PaginateOnly,
headers: CommonHeaders = Depends(CommonHeaders.as_dependency),
):
"""
List building area 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 = BuildAreaRouterCluster.get_event_cluster("BuildAreaList")
event_cluster_matched = FoundCluster.match_event(event_key=event_key)
return event_cluster_matched.event_callable(data=data)
@area_route.post(
path="/create",
description="Create building area endpoint",
operation_id="66e8b310-7a5d-41f4-977c-ee5f3b0603da",
)
def area_create_route(
data: dict,
headers: CommonHeaders = Depends(CommonHeaders.as_dependency),
):
"""
Create building area 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 = BuildAreaRouterCluster.get_event_cluster("BuildAreaCreate")
event_cluster_matched = FoundCluster.match_event(event_key=event_key)
return event_cluster_matched.event_callable(data=data)
@area_route.post(
path="/update",
description="Update building area endpoint",
operation_id="d7769899-156d-49ff-85f4-06e3faa23e30",
)
def area_update_route(
data: dict,
headers: CommonHeaders = Depends(CommonHeaders.as_dependency),
):
"""
Update building area 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 = BuildAreaRouterCluster.get_event_cluster("BuildAreaUpdate")
event_cluster_matched = FoundCluster.match_event(event_key=event_key)
return event_cluster_matched.event_callable(data=data)

View File

@@ -0,0 +1,70 @@
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.building.cluster import BuildRouterCluster
building_route = APIRouter(prefix="/building", tags=["Building"])
@building_route.post(
path="/list",
description="List buildings endpoint",
operation_id="72003d1f-79cc-43c3-9150-3a94f318186e",
)
def building_list_route(
data: PaginateOnly,
headers: CommonHeaders = Depends(CommonHeaders.as_dependency),
):
"""
List buildings 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 = BuildRouterCluster.get_event_cluster("BuildList")
event_cluster_matched = FoundCluster.match_event(event_key=event_key)
return event_cluster_matched.event_callable(data=data)
@building_route.post(
path="/create",
description="Create buildings endpoint",
operation_id="13b86c9c-702b-411f-acb9-9ff50ef86507",
)
def building_create_route(
data: dict,
headers: CommonHeaders = Depends(CommonHeaders.as_dependency),
):
"""
Create buildings 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 = BuildRouterCluster.get_event_cluster("BuildCreate")
event_cluster_matched = FoundCluster.match_event(event_key=event_key)
return event_cluster_matched.event_callable(data=data)
@building_route.post(
path="/update",
description="Update buildings endpoint",
operation_id="b09fb6fd-a6c9-4c03-8637-9af0570a84e8",
)
def building_update_route(
data: dict,
headers: CommonHeaders = Depends(CommonHeaders.as_dependency),
):
"""
Update buildings 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 = BuildRouterCluster.get_event_cluster("BuildUpdate")
event_cluster_matched = FoundCluster.match_event(event_key=event_key)
return event_cluster_matched.event_callable(data=data)

View File

@@ -0,0 +1,22 @@
from fastapi import APIRouter
def get_routes() -> list[APIRouter]:
from .building.route import building_route
from .area.route import area_route
from .sites.route import sites_route
from .parts.route import parts_route
from .spaces.route import spaces_route
from .type.route import build_types_route
return [building_route, area_route, sites_route, parts_route, spaces_route, build_types_route]
def get_safe_endpoint_urls() -> list[tuple[str, str]]:
return [
("/", "GET"),
("/docs", "GET"),
("/redoc", "GET"),
("/openapi.json", "GET"),
("/metrics", "GET"),
]

View File

@@ -0,0 +1,70 @@
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.sites.cluster import BuildSitesRouterCluster
sites_route = APIRouter(prefix="/building/sites", tags=["Build Sites"])
@sites_route.post(
path="/list",
description="List build sites endpoint",
operation_id="b64a2479-4160-43bf-9676-cd762a48bf62",
)
def sites_list_route(
data: PaginateOnly,
headers: CommonHeaders = Depends(CommonHeaders.as_dependency),
):
"""
List build sites 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 = BuildSitesRouterCluster.get_event_cluster("BuildSitesList")
event_cluster_matched = FoundCluster.match_event(event_key=event_key)
return event_cluster_matched.event_callable(data=data)
@sites_route.post(
path="/create",
description="Create build sites endpoint",
operation_id="e63dab4b-d446-430c-ac7d-0cb47a66390c",
)
def sites_create_route(
data: dict,
headers: CommonHeaders = Depends(CommonHeaders.as_dependency),
):
"""
Create build sites 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 = BuildSitesRouterCluster.get_event_cluster("BuildSitesCreate")
event_cluster_matched = FoundCluster.match_event(event_key=event_key)
return event_cluster_matched.event_callable(data=data)
@sites_route.post(
path="/update",
description="Update build sites endpoint",
operation_id="34ec902c-5ebe-41f6-8294-dfe634d47d98",
)
def sites_update_route(
data: dict,
headers: CommonHeaders = Depends(CommonHeaders.as_dependency),
):
"""
Update build sites 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 = BuildSitesRouterCluster.get_event_cluster("BuildSitesUpdate")
event_cluster_matched = FoundCluster.match_event(event_key=event_key)
return event_cluster_matched.event_callable(data=data)

View File

@@ -0,0 +1,70 @@
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.spaces.cluster import BuildLivingSpaceRouterCluster
spaces_route = APIRouter(prefix="/building/spaces", tags=["Build Spaces"])
@spaces_route.post(
path="/list",
description="List build spaces endpoint",
operation_id="9d072fd3-d727-4f71-bf0c-fd74e25ba507",
)
def spaces_list_route(
data: PaginateOnly,
headers: CommonHeaders = Depends(CommonHeaders.as_dependency),
):
"""
List build spaces 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 = BuildLivingSpaceRouterCluster.get_event_cluster("BuildLivingSpaceList")
event_cluster_matched = FoundCluster.match_event(event_key=event_key)
return event_cluster_matched.event_callable(data=data)
@spaces_route.post(
path="/create",
description="Create build spaces endpoint",
operation_id="4520cfb7-ed45-44d2-a706-e2a8be777419",
)
def spaces_create_route(
data: dict,
headers: CommonHeaders = Depends(CommonHeaders.as_dependency),
):
"""
Create build spaces 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 = BuildLivingSpaceRouterCluster.get_event_cluster("BuildLivingSpaceCreate")
event_cluster_matched = FoundCluster.match_event(event_key=event_key)
return event_cluster_matched.event_callable(data=data)
@spaces_route.post(
path="/update",
description="Update build spaces endpoint",
operation_id="d8c90f64-0a4c-4a7c-a685-0e1894cbe1c4",
)
def spaces_update_route(
data: dict,
headers: CommonHeaders = Depends(CommonHeaders.as_dependency),
):
"""
Update build spaces 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 = BuildLivingSpaceRouterCluster.get_event_cluster("BuildLivingSpaceUpdate")
event_cluster_matched = FoundCluster.match_event(event_key=event_key)
return event_cluster_matched.event_callable(data=data)

View File

@@ -0,0 +1,70 @@
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.type.cluster import BuildTypeRouterCluster
build_types_route = APIRouter(prefix="/building/types", tags=["Build Types"])
@build_types_route.post(
path="/list",
description="List build types endpoint",
operation_id="916d2e02-ee83-4099-ba02-63e2a3f83ec4",
)
def build_types_list_route(
data: PaginateOnly,
headers: CommonHeaders = Depends(CommonHeaders.as_dependency),
):
"""
List build types 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 = BuildTypeRouterCluster.get_event_cluster("BuildTypeList")
event_cluster_matched = FoundCluster.match_event(event_key=event_key)
return event_cluster_matched.event_callable(data=data)
@build_types_route.post(
path="/create",
description="Create build types endpoint",
operation_id="ae7d9e5d-3933-4613-a686-c887d74b85b3",
)
def build_types_create_route(
data: dict,
headers: CommonHeaders = Depends(CommonHeaders.as_dependency),
):
"""
Create build types 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 = BuildTypeRouterCluster.get_event_cluster("BuildTypeCreate")
event_cluster_matched = FoundCluster.match_event(event_key=event_key)
return event_cluster_matched.event_callable(data=data)
@build_types_route.post(
path="/update",
description="Update build types endpoint",
operation_id="0f304075-efb0-4f9d-a3b2-e60eeb4d9a87",
)
def build_types_update_route(
data: dict,
headers: CommonHeaders = Depends(CommonHeaders.as_dependency),
):
"""
Update build types 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 = BuildTypeRouterCluster.get_event_cluster("BuildTypeUpdate")
event_cluster_matched = FoundCluster.match_event(event_key=event_key)
return event_cluster_matched.event_callable(data=data)

View File

@@ -0,0 +1,15 @@
from .building.cluster import BuildRouterCluster
from .area.cluster import BuildAreaRouterCluster
from .parts.cluster import BuildPartsRouterCluster
from .spaces.cluster import BuildLivingSpaceRouterCluster
from .type.cluster import BuildTypeRouterCluster
from .sites.cluster import BuildSitesRouterCluster
__all__ = [
"BuildRouterCluster",
"BuildAreaRouterCluster",
"BuildPartsRouterCluster",
"BuildLivingSpaceRouterCluster",
"BuildTypeRouterCluster",
"BuildSitesRouterCluster",
]

View File

@@ -0,0 +1,20 @@
from ApiControllers.abstracts.event_clusters import EventCluster, RouterCluster
from .supers_events import (
BuildAreaListEvent,
BuildAreaCreateEvent,
BuildAreaUpdateEvent,
)
BuildAreaRouterCluster = RouterCluster(name="BuildAreaRouterCluster")
BuildAreaEventClusterList = EventCluster(name="BuildAreaList", endpoint_uu_id="cc487a4f-9a45-4072-89c1-a1ad504c79ad")
BuildAreaEventClusterCreate = EventCluster(name="BuildAreaCreate", endpoint_uu_id="bdd58d68-3a7c-4150-9f5b-e322db35b804")
BuildAreaEventClusterUpdate = EventCluster(name="BuildAreaUpdate", endpoint_uu_id="cad0c4e2-36e3-4f80-9ad2-b06bf8cd8d1c")
BuildAreaEventClusterList.add_event(BuildAreaListEvent)
BuildAreaEventClusterCreate.add_event(BuildAreaCreateEvent)
BuildAreaEventClusterUpdate.add_event(BuildAreaUpdateEvent)
BuildAreaRouterCluster.set_event_cluster(BuildAreaEventClusterList)
BuildAreaRouterCluster.set_event_cluster(BuildAreaEventClusterCreate)
BuildAreaRouterCluster.set_event_cluster(BuildAreaEventClusterUpdate)

View File

@@ -0,0 +1,96 @@
from ApiControllers.abstracts.event_clusters import Event
from Controllers.Postgres.pagination import Pagination, PaginationResult, PaginateOnly
from Controllers.Postgres.response import EndpointResponse
from Schemas import BuildArea
# List endpoint
BuildAreaListEvent = Event(
name="build_area_list",
key="306e6e11-21da-4fbb-b3d6-3a5179ecfe71",
request_validator=None, # TODO: Add request validator
response_validator=None, # TODO: Add response validator
description="List events of build area endpoint",
)
# Create endpoint
BuildAreaCreateEvent = Event(
name="build_area_create",
key="aa173f0f-7a97-4e91-97da-173626dd3257",
request_validator=None, # TODO: Add request validator
response_validator=None, # TODO: Add response validator
description="Create events of build area endpoint",
)
# Update endpoint
BuildAreaUpdateEvent = Event(
name="build_area_update",
key="358404d0-fb80-4d58-8907-0c04948b364e",
request_validator=None, # TODO: Add request validator
response_validator=None, # TODO: Add response validator
description="Update events of build area endpoint",
)
def build_area_list_callable(list_options: PaginateOnly):
"""
Example callable method
"""
list_options = PaginateOnly(**list_options.model_dump())
with BuildArea.new_session() as db_session:
if list_options.query:
buildings_list = BuildArea.filter_all(
*BuildArea.convert(list_options.query), db=db_session
)
else:
buildings_list = BuildArea.filter_all(db=db_session)
pagination = Pagination(data=buildings_list)
pagination.change(**list_options.model_dump())
pagination_result = PaginationResult(
data=buildings_list,
pagination=pagination,
# response_model="",
).pagination.as_dict
return EndpointResponse(
message="MSG0003-LIST",
pagination_result=pagination_result,
).response
BuildAreaListEvent.event_callable = build_area_list_callable
def build_area_create_callable(data: dict):
"""
Example callable method
"""
with BuildArea.new_session() as db_session:
return {
"completed": True,
"message": "Example callable method 2",
"info": {
"host": "example_host",
"user_agent": "example_user_agent",
},
}
BuildAreaCreateEvent.event_callable = build_area_create_callable
def build_area_update_callable(data: dict):
"""
Example callable method
"""
with BuildArea.new_session() as db_session:
return {
"completed": True,
"message": "Example callable method 2",
"info": {
"host": "example_host",
"user_agent": "example_user_agent",
},
}
BuildAreaUpdateEvent.event_callable = build_area_update_callable

View File

@@ -0,0 +1,27 @@
from ApiControllers.abstracts.event_clusters import EventCluster, RouterCluster
from .supers_events import (
BuildingListEvent,
BuildingCreateEvent,
BuildingUpdateEvent,
)
BuildRouterCluster = RouterCluster(name="BuildRouterCluster")
BuildEventClusterList = EventCluster(
name="BuildList", endpoint_uu_id="1a186985-ed2a-434b-af28-22d6773382e9"
)
BuildEventClusterList.add_event(BuildingListEvent)
BuildEventClusterCreate = EventCluster(
name="BuildCreate", endpoint_uu_id="d545c5a9-bbbf-4795-a4de-467db0809a04"
)
BuildEventClusterCreate.add_event(BuildingCreateEvent)
BuildEventClusterUpdate = EventCluster(
name="BuildUpdate", endpoint_uu_id="6166a28e-8da8-4a2c-96c8-0a1651739cf3"
)
BuildEventClusterUpdate.add_event(BuildingUpdateEvent)
BuildRouterCluster.set_event_cluster(BuildEventClusterList)
BuildRouterCluster.set_event_cluster(BuildEventClusterCreate)
BuildRouterCluster.set_event_cluster(BuildEventClusterUpdate)

View File

@@ -0,0 +1,98 @@
from ApiControllers.abstracts.event_clusters import Event
from Validations.building.building.validations import (
REQUESTEWFAZCDMPVZHIWOKZEJBIEUDAFBNXFEEAEGSELVGGCDMWLQPYMRAEEABSRQJUFBIMFEEADXK,
)
from Controllers.Postgres.pagination import Pagination, PaginationResult, PaginateOnly
from Controllers.Postgres.response import EndpointResponse
from Schemas import Build
# List endpoint
BuildingListEvent = Event(
name="building_list",
key="e675d36a-6772-44a8-b153-9a8d55e0f560",
request_validator=None, # TODO: Add request validator
response_validator=None, # TODO: Add response validator
description="List events of buildings endpoint",
)
# Create endpoint
BuildingCreateEvent = Event(
name="building_create",
key="d79e3bdc-8f75-41d0-8eeb-dd9893d7ea0d",
request_validator=None, # TODO: Add request validator
response_validator=None, # TODO: Add response validator
description="Create events of buildings endpoint",
)
# Update endpoint
BuildingUpdateEvent = Event(
name="building_update",
key="45e1bb8c-c521-4013-9e99-d3a58204bc5f",
request_validator=None, # TODO: Add request validator
response_validator=None, # TODO: Add response validator
description="Update events of buildings endpoint",
)
def building_list_callable(list_options: PaginateOnly):
"""
Example callable method
"""
list_options = PaginateOnly(**list_options.model_dump())
with Build.new_session() as db_session:
if list_options.query:
buildings_list = Build.filter_all(
*Build.convert(list_options.query), db=db_session
)
else:
buildings_list = Build.filter_all(db=db_session)
pagination = Pagination(data=buildings_list)
pagination.change(**list_options.model_dump())
pagination_result = PaginationResult(
data=buildings_list,
pagination=pagination,
# response_model="",
).pagination.as_dict
return EndpointResponse(
message="MSG0003-LIST",
pagination_result=pagination_result,
request=REQUESTEWFAZCDMPVZHIWOKZEJBIEUDAFBNXFEEAEGSELVGGCDMWLQPYMRAEEABSRQJUFBIMFEEADXK,
).response
BuildingListEvent.event_callable = building_list_callable
def building_create_callable():
"""
Example callable method
"""
return {
"completed": True,
"message": "Example callable method 2",
"info": {
"host": "example_host",
"user_agent": "example_user_agent",
},
}
BuildingCreateEvent.event_callable = building_create_callable
def building_update_callable():
"""
Example callable method
"""
return {
"completed": True,
"message": "Example callable method 2",
"info": {
"host": "example_host",
"user_agent": "example_user_agent",
},
}
BuildingUpdateEvent.event_callable = building_update_callable

View File

@@ -0,0 +1,27 @@
from ApiControllers.abstracts.event_clusters import EventCluster, RouterCluster
from .supers_events import (
BuildSitesListEvent,
BuildSitesCreateEvent,
BuildSitesUpdateEvent,
)
BuildSitesRouterCluster = RouterCluster(name="BuildSitesRouterCluster")
BuildSitesEventClusterList = EventCluster(
name="BuildSitesList", endpoint_uu_id="1159fb39-ad5d-46fb-8b6b-461021a39da3"
)
BuildSitesEventClusterList.add_event(BuildSitesListEvent)
BuildSitesEventClusterCreate = EventCluster(
name="BuildCreate", endpoint_uu_id="bc71d276-fd51-43bb-90f2-6f4245b59d72"
)
BuildSitesEventClusterCreate.add_event(BuildSitesCreateEvent)
BuildSitesEventClusterUpdate = EventCluster(
name="BuildUpdate", endpoint_uu_id="eca87985-3074-44cb-a947-f01e30769019"
)
BuildSitesEventClusterUpdate.add_event(BuildSitesUpdateEvent)
BuildSitesRouterCluster.set_event_cluster(BuildSitesEventClusterList)
BuildSitesRouterCluster.set_event_cluster(BuildSitesEventClusterCreate)
BuildSitesRouterCluster.set_event_cluster(BuildSitesEventClusterUpdate)

View File

@@ -0,0 +1,96 @@
from ApiControllers.abstracts.event_clusters import Event
from Controllers.Postgres.pagination import Pagination, PaginationResult, PaginateOnly
from Controllers.Postgres.response import EndpointResponse
from Schemas import BuildSites
# List endpoint
BuildSitesListEvent = Event(
name="build_sites_list",
key="d16c692f-5597-493a-90fe-b2bfe0d1c163",
request_validator=None, # TODO: Add request validator
response_validator=None, # TODO: Add response validator
description="List events of build sites endpoint",
)
# Create endpoint
BuildSitesCreateEvent = Event(
name="build_sites_create",
key="8ed0929e-9c96-4e81-b073-e3578ca13f37",
request_validator=None, # TODO: Add request validator
response_validator=None, # TODO: Add response validator
description="Create events of build sites endpoint",
)
# Update endpoint
BuildSitesUpdateEvent = Event(
name="build_sites_update",
key="ece55fa6-afba-4c34-bd2e-f39a4cdd6b5c",
request_validator=None, # TODO: Add request validator
response_validator=None, # TODO: Add response validator
description="Update events of build sites endpoint",
)
def build_sites_list_callable(list_options: PaginateOnly):
"""
Example callable method
"""
list_options = PaginateOnly(**list_options.model_dump())
with BuildSites.new_session() as db_session:
if list_options.query:
buildings_list = BuildSites.filter_all(
*BuildSites.convert(list_options.query), db=db_session
)
else:
buildings_list = BuildSites.filter_all(db=db_session)
pagination = Pagination(data=buildings_list)
pagination.change(**list_options.model_dump())
pagination_result = PaginationResult(
data=buildings_list,
pagination=pagination,
# response_model="",
).pagination.as_dict
return EndpointResponse(
message="MSG0003-LIST",
pagination_result=pagination_result,
).response
BuildSitesListEvent.event_callable = build_sites_list_callable
def build_sites_create_callable(data: dict):
"""
Example callable method
"""
with BuildSites.new_session() as db_session:
return {
"completed": True,
"message": "Example callable method 2",
"info": {
"host": "example_host",
"user_agent": "example_user_agent",
},
}
BuildSitesCreateEvent.event_callable = build_sites_create_callable
def build_sites_update_callable(data: dict):
"""
Example callable method
"""
with BuildSites.new_session() as db_session:
return {
"completed": True,
"message": "Example callable method 2",
"info": {
"host": "example_host",
"user_agent": "example_user_agent",
},
}
BuildSitesUpdateEvent.event_callable = build_sites_update_callable

View File

@@ -0,0 +1,27 @@
from ApiControllers.abstracts.event_clusters import EventCluster, RouterCluster
from .supers_events import (
BuildLivingSpaceListEvent,
BuildLivingSpaceCreateEvent,
BuildLivingSpaceUpdateEvent,
)
BuildLivingSpaceRouterCluster = RouterCluster(name="BuildLivingSpaceRouterCluster")
BuildLivingSpaceEventClusterList = EventCluster(
name="BuildLivingSpaceList", endpoint_uu_id="ad17c019-3050-4c41-ab6f-9ce43290e81a"
)
BuildLivingSpaceEventClusterList.add_event(BuildLivingSpaceListEvent)
BuildLivingSpaceEventClusterCreate = EventCluster(
name="BuildCreate", endpoint_uu_id="ed9a0303-14e2-46fe-bec0-d395c29801ff"
)
BuildLivingSpaceEventClusterCreate.add_event(BuildLivingSpaceCreateEvent)
BuildLivingSpaceEventClusterUpdate = EventCluster(
name="BuildUpdate", endpoint_uu_id="c62bb3dc-03c0-406c-8bbc-0e1f75a6420c"
)
BuildLivingSpaceEventClusterUpdate.add_event(BuildLivingSpaceUpdateEvent)
BuildLivingSpaceRouterCluster.set_event_cluster(BuildLivingSpaceEventClusterList)
BuildLivingSpaceRouterCluster.set_event_cluster(BuildLivingSpaceEventClusterCreate)
BuildLivingSpaceRouterCluster.set_event_cluster(BuildLivingSpaceEventClusterUpdate)

View File

@@ -0,0 +1,96 @@
from ApiControllers.abstracts.event_clusters import Event
from Controllers.Postgres.pagination import Pagination, PaginationResult, PaginateOnly
from Controllers.Postgres.response import EndpointResponse
from Schemas import BuildLivingSpace
# List endpoint
BuildLivingSpaceListEvent = Event(
name="build_living_space_list",
key="d16c692f-5597-493a-90fe-b2bfe0d1c163",
request_validator=None, # TODO: Add request validator
response_validator=None, # TODO: Add response validator
description="List events of build living spaces endpoint",
)
# Create endpoint
BuildLivingSpaceCreateEvent = Event(
name="build_living_space_create",
key="8ed0929e-9c96-4e81-b073-e3578ca13f37",
request_validator=None, # TODO: Add request validator
response_validator=None, # TODO: Add response validator
description="Create events of build living spaces endpoint",
)
# Update endpoint
BuildLivingSpaceUpdateEvent = Event(
name="build_living_space_update",
key="ece55fa6-afba-4c34-bd2e-f39a4cdd6b5c",
request_validator=None, # TODO: Add request validator
response_validator=None, # TODO: Add response validator
description="Update events of build living spaces endpoint",
)
def build_living_space_list_callable(list_options: PaginateOnly):
"""
List callable method
"""
list_options = PaginateOnly(**list_options.model_dump())
with BuildLivingSpace.new_session() as db_session:
if list_options.query:
buildings_list = BuildLivingSpace.filter_all(
*BuildLivingSpace.convert(list_options.query), db=db_session
)
else:
buildings_list = BuildLivingSpace.filter_all(db=db_session)
pagination = Pagination(data=buildings_list)
pagination.change(**list_options.model_dump())
pagination_result = PaginationResult(
data=buildings_list,
pagination=pagination,
# response_model="",
).pagination.as_dict
return EndpointResponse(
message="MSG0003-LIST",
pagination_result=pagination_result,
).response
BuildLivingSpaceListEvent.event_callable = build_living_space_list_callable
def build_living_space_create_callable(data: dict):
"""
Create callable method
"""
with BuildLivingSpace.new_session() as db_session:
return {
"completed": True,
"message": "Build living space created",
"info": {
"host": "example_host",
"user_agent": "example_user_agent",
},
}
BuildLivingSpaceCreateEvent.event_callable = build_living_space_create_callable
def build_living_space_update_callable(data: dict):
"""
Update callable method
"""
with BuildLivingSpace.new_session() as db_session:
return {
"completed": True,
"message": "Build living space updated",
"info": {
"host": "example_host",
"user_agent": "example_user_agent",
},
}
BuildLivingSpaceUpdateEvent.event_callable = build_living_space_update_callable

View File

@@ -0,0 +1,27 @@
from ApiControllers.abstracts.event_clusters import EventCluster, RouterCluster
from .supers_events import (
BuildTypeListEvent,
BuildTypeCreateEvent,
BuildTypeUpdateEvent,
)
BuildTypeRouterCluster = RouterCluster(name="BuildTypeRouterCluster")
BuildTypeEventClusterList = EventCluster(
name="BuildTypeList", endpoint_uu_id="14f076c2-b118-4cfb-b679-9f2168a2658c"
)
BuildTypeEventClusterList.add_event(BuildTypeListEvent)
BuildTypeEventClusterCreate = EventCluster(
name="BuildCreate", endpoint_uu_id="4c90c4cf-8873-479a-a56b-d12f276efa9d"
)
BuildTypeEventClusterCreate.add_event(BuildTypeCreateEvent)
BuildTypeEventClusterUpdate = EventCluster(
name="BuildUpdate", endpoint_uu_id="cde9db69-6795-4562-802d-540d0b03ceaa"
)
BuildTypeEventClusterUpdate.add_event(BuildTypeUpdateEvent)
BuildTypeRouterCluster.set_event_cluster(BuildTypeEventClusterList)
BuildTypeRouterCluster.set_event_cluster(BuildTypeEventClusterCreate)
BuildTypeRouterCluster.set_event_cluster(BuildTypeEventClusterUpdate)

View File

@@ -0,0 +1,96 @@
from ApiControllers.abstracts.event_clusters import Event
from Controllers.Postgres.pagination import Pagination, PaginationResult, PaginateOnly
from Controllers.Postgres.response import EndpointResponse
from Schemas import BuildTypes
# List endpoint
BuildTypeListEvent = Event(
name="build_type_list",
key="e013b43e-3845-4a95-b5b0-409343659c5f",
request_validator=None, # TODO: Add request validator
response_validator=None, # TODO: Add response validator
description="List events of build types endpoint",
)
# Create endpoint
BuildTypeCreateEvent = Event(
name="build_type_create",
key="8119a07e-bbb2-4d4c-b529-f923e8cdcc60",
request_validator=None, # TODO: Add request validator
response_validator=None, # TODO: Add response validator
description="Create events of build types endpoint",
)
# Update endpoint
BuildTypeUpdateEvent = Event(
name="build_type_update",
key="6b7b89d5-6406-4614-aba4-c942f0c081ac",
request_validator=None, # TODO: Add request validator
response_validator=None, # TODO: Add response validator
description="Update events of build types endpoint",
)
def build_type_list_callable(list_options: PaginateOnly):
"""
List callable method
"""
list_options = PaginateOnly(**list_options.model_dump())
with BuildTypes.new_session() as db_session:
if list_options.query:
buildings_list = BuildTypes.filter_all(
*BuildTypes.convert(list_options.query), db=db_session
)
else:
buildings_list = BuildTypes.filter_all(db=db_session)
pagination = Pagination(data=buildings_list)
pagination.change(**list_options.model_dump())
pagination_result = PaginationResult(
data=buildings_list,
pagination=pagination,
# response_model="",
).pagination.as_dict
return EndpointResponse(
message="MSG0003-LIST",
pagination_result=pagination_result,
).response
BuildTypeListEvent.event_callable = build_type_list_callable
def build_type_create_callable(data: dict):
"""
Create callable method
"""
with BuildTypes.new_session() as db_session:
return {
"completed": True,
"message": "Build type created",
"info": {
"host": "example_host",
"user_agent": "example_user_agent",
},
}
BuildTypeCreateEvent.event_callable = build_type_create_callable
def build_type_update_callable(data: dict):
"""
Update callable method
"""
with BuildTypes.new_session() as db_session:
return {
"completed": True,
"message": "Build type updated",
"info": {
"host": "example_host",
"user_agent": "example_user_agent",
},
}
BuildTypeUpdateEvent.event_callable = build_type_update_callable

View File

View File

@@ -0,0 +1,33 @@
from pydantic import BaseModel
from typing import Optional
class REQUESTEWFAZCDMPVZHIWOKZEJBIEUDAFBNXFEEAEGSELVGGCDMWLQPYMRAEEABSRQJUFBIMFEEADXK(BaseModel):
gov_address_code: str
build_name: str
build_no: str
max_floor: int
underground_floor: int
build_date: str
decision_period_date: str
tax_no: str
lift_count: int
heating_system: bool
cooling_system: bool
hot_water_system: bool
block_service_man_count: int
security_service_man_count: int
garage_count: int
management_room_id: int
site_uu_id: str
address_uu_id: str
class REQUESTOCARDAJXLDANCXQAJWDBDIWXHUAEKQNUOSBZOCWXDAFGLAAVRBSADHUBDXAREUSESYGNGKBR(BaseModel):
pass
class REQUESTAEKUDAILSFMCCLOERBHBFRCIKFCSNCBOSENCAEOIDACPRFZCCWGEDBHBFBMZBFCJHCBVKEFC(BaseModel):
pass

View File

@@ -0,0 +1,26 @@
from pydantic import BaseModel
from typing import Optional
class ListOptions(BaseModel):
"""
Query for list option abilities
"""
page: Optional[int] = 1
size: Optional[int] = 10
order_field: Optional[str] = "id"
order_type: Optional[str] = "asc"
# include_joins: Optional[list] = None
query: Optional[dict] = None
class PaginateOnly(BaseModel):
"""
Query for list option abilities
"""
page: Optional[int] = 1
size: Optional[int] = 10
order_field: Optional[str] = "id"
order_type: Optional[str] = "asc"

View File

@@ -26,7 +26,7 @@ def people_route_list(
event_key = TokenProvider.retrieve_event_codes(**event_founder_dict)
FoundCluster = PeopleRouterCluster.get_event_cluster("PeopleList")
event_cluster_matched = FoundCluster.match_event(event_key=event_key)
return event_cluster_matched.event_callable()
return event_cluster_matched.event_callable(data=data)
@people_route.post(

View File

@@ -2,7 +2,7 @@ from ApiControllers.abstracts.event_clusters import Event
from Validations.people.validations import (
REQUESTAWMXNTKMGPPOJWRCTZUBADNFLQDBDYVQAORFAVCSXUUHEBQHCEPCSKFBADBODFDBPYKOVINV,
)
from Controllers.Postgres.pagination import Pagination, PaginationResult, PaginateOnly
from Controllers.Postgres.pagination import ListOptions, Pagination, PaginationResult, PaginateOnly
from Controllers.Postgres.response import EndpointResponse
from Schemas import People
@@ -34,7 +34,7 @@ SupersPeopleListEvent = Event(
)
def supers_people_create_callable(list_options):
def supers_people_create_callable():
"""
Example callable method
"""
@@ -70,11 +70,11 @@ def supers_people_update_callable():
SupersPeopleUpdateEvent.event_callable = supers_people_update_callable
def supers_people_list_callable(list_options: PaginateOnly):
def supers_people_list_callable(data: PaginateOnly):
"""
Example callable method
"""
list_options = PaginateOnly(**list_options.model_dump())
list_options = PaginateOnly(**data.model_dump())
with People.new_session() as db_session:
if list_options.query:
people_list = People.filter_all(

View File

@@ -1,10 +1,10 @@
from ApiControllers.abstracts.event_clusters import Event
from Controllers.Postgres.pagination import Pagination, PaginationResult, PaginateOnly
from Controllers.Postgres.response import EndpointResponse
from Schemas import (
Users,
)
from Schemas import Users
# List endpoint
SuperUsersListEvent = Event(
name="supers_users_list",
key="341b394f-9f11-4abb-99e7-4b27fa6bf012",
@@ -13,6 +13,24 @@ SuperUsersListEvent = Event(
description="List events of users endpoint",
)
# Create endpoint
SuperUsersCreateEvent = Event(
name="supers_users_create",
key="4e7e189e-e015-4ff8-902d-60138cbc77a6",
request_validator=None, # TODO: Add request validator
response_validator=None, # TODO: Add response validator
description="Create events of users endpoint",
)
# Update endpoint
SuperUsersUpdateEvent = Event(
name="supers_users_update",
key="efa4aa4a-d414-4391-91ee-97eb617b7755",
request_validator=None, # TODO: Add request validator
response_validator=None, # TODO: Add response validator
description="Update events of users endpoint",
)
def supers_users_list_callable(list_options: PaginateOnly):
"""
@@ -32,31 +50,15 @@ def supers_users_list_callable(list_options: PaginateOnly):
data=users_list,
pagination=pagination,
# response_model="",
)
).pagination.as_dict
return EndpointResponse(
message="MSG0003-LIST",
pagination_result=pagination_result,
).response
# return {
# "completed": True,
# "message": "Example callable method 2",
# "info": {
# "host": "example_host",
# "user_agent": "example_user_agent",
# },
# }
SuperUsersListEvent.event_callable = supers_users_list_callable
SuperUsersCreateEvent = Event(
name="supers_users_create",
key="4e7e189e-e015-4ff8-902d-60138cbc77a6",
request_validator=None, # TODO: Add request validator
response_validator=None, # TODO: Add response validator
description="Create events of users endpoint",
)
def supers_users_create_callable():
"""
@@ -74,15 +76,6 @@ def supers_users_create_callable():
SuperUsersCreateEvent.event_callable = supers_users_create_callable
# Update endpoint
SuperUsersUpdateEvent = Event(
name="supers_users_update",
key="efa4aa4a-d414-4391-91ee-97eb617b7755",
request_validator=None, # TODO: Add request validator
response_validator=None, # TODO: Add response validator
description="Update events of users endpoint",
)
def supers_users_update_callable():
"""