diff --git a/.idea/workspace.xml b/.idea/workspace.xml index d3e7e5a..7143707 100644 --- a/.idea/workspace.xml +++ b/.idea/workspace.xml @@ -6,32 +6,21 @@ - + + - - - - - - - - - - - - - - - - - - - - - - - - + + + + + + + + + + + + - - - - - + + + + + diff --git a/Events/AllEvents/events/__init__.py b/Events/AllEvents/events/__init__.py index 79a28e9..06e4ad3 100644 --- a/Events/AllEvents/events/__init__.py +++ b/Events/AllEvents/events/__init__.py @@ -2,7 +2,10 @@ Events package initialization. """ from .account.cluster import AccountCluster +from .address.cluster import AddressCluster + __all__ = [ "AccountCluster", + "AddressCluster" ] diff --git a/Events/AllEvents/events/account/account_records.py b/Events/AllEvents/events/account/account_records.py index 738c7cd..3a3e52c 100644 --- a/Events/AllEvents/events/account/account_records.py +++ b/Events/AllEvents/events/account/account_records.py @@ -1,5 +1,5 @@ """ -template related API endpoints. +Account related API endpoints. """ from typing import Any, Dict from fastapi import Request @@ -9,43 +9,120 @@ from Events.base_request_model import EndpointBaseRequestModel, ContextRetriever from ApiLayers.Middleware.token_event_middleware import TokenEventMiddleware from ApiLayers.ApiValidations.Response.default_response import EndpointSuccessListResponse -from .api_events import account_insert_super_user_event from .function_handlers import AccountListEventMethods +from .api_events import SuperUserAccountEvents -AccountRecordsEventMethods = MethodToEvent( - name="AccountRecordsEventMethods", +AccountRecordsListEventMethods = MethodToEvent( + name="AccountRecordsListEventMethods", events={ - account_insert_super_user_event.key: account_insert_super_user_event, + SuperUserAccountEvents.SuperUserListEvent.key: SuperUserAccountEvents.SuperUserListEvent, }, headers=[], errors=[], decorators_list=[TokenEventMiddleware.event_required], url="/list", method="POST", - summary="Login via domain and access key : [email] | [phone]", - description="Login to the system via domain, access key : [email] | [phone]", + summary="List all accounts by given previligous", + description="List all accounts by given previligous", ) -def account_insert_event_endpoint( +def account_list_event_endpoint( request: Request, data: EndpointBaseRequestModel ) -> Dict[str, Any]: - context_retriever = ContextRetrievers(func=account_insert_event_endpoint) - event_2_catch = AccountRecordsEventMethods.retrieve_event( - event_function_code=f"{account_insert_super_user_event.key}" + context_retriever = ContextRetrievers(func=account_list_event_endpoint) + event_2_catch = AccountRecordsListEventMethods.retrieve_event( + event_function_code=f"{SuperUserAccountEvents.SuperUserListEvent.key}" ) context_retriever.RESPONSE_VALIDATOR = event_2_catch.RESPONSE_VALIDATOR data = event_2_catch.REQUEST_VALIDATOR(**data.data) AccountListEventMethods.context_retriever = context_retriever pagination_result = event_2_catch.endpoint_callable(data=data) return EndpointSuccessListResponse( - code="ACCOUNTS_LIST", lang=context_retriever.token.lang + code=event_2_catch.static_key, lang=context_retriever.token.lang ).as_dict( data=pagination_result.data, pagination=pagination_result.pagination.as_dict() ) -AccountRecordsEventMethods.endpoint_callable = ( - account_insert_event_endpoint +AccountRecordsListEventMethods.endpoint_callable = ( + account_list_event_endpoint ) + + +AccountRecordsCreateEventMethods = MethodToEvent( + name="AccountRecordsCreateEventMethods", + events={ + SuperUserAccountEvents.SuperUserCreateEvent.key: SuperUserAccountEvents.SuperUserCreateEvent, + }, + headers=[], + errors=[], + decorators_list=[TokenEventMiddleware.event_required], + url="/create", + method="POST", + summary="Create Account via given data and previligous", + description="Create Account via given data and previligous", +) + + +def account_create_event_endpoint( + request: Request, data: EndpointBaseRequestModel +) -> Dict[str, Any]: + context_retriever = ContextRetrievers(func=account_create_event_endpoint) + event_2_catch = AccountRecordsCreateEventMethods.retrieve_event( + event_function_code=f"{SuperUserAccountEvents.SuperUserCreateEvent.key}" + ) + context_retriever.RESPONSE_VALIDATOR = event_2_catch.RESPONSE_VALIDATOR + data = event_2_catch.REQUEST_VALIDATOR(**data.data) + AccountListEventMethods.context_retriever = context_retriever + pagination_result = event_2_catch.endpoint_callable(data=data) + return EndpointSuccessListResponse( + code=event_2_catch.static_key, lang=context_retriever.token.lang + ).as_dict( + data=pagination_result.data, pagination=pagination_result.pagination.as_dict() + ) + + +AccountRecordsCreateEventMethods.endpoint_callable = ( + account_create_event_endpoint +) + + +AccountRecordsUpdateEventMethods = MethodToEvent( + name="AccountRecordsUpdateEventMethods", + events={ + SuperUserAccountEvents.SuperUserUpdateEvent.key: SuperUserAccountEvents.SuperUserUpdateEvent, + }, + headers=[], + errors=[], + decorators_list=[TokenEventMiddleware.event_required], + url="/update", + method="POST", + summary="Update Account via given data and previligous", + description="Update Account via given data and previligous", +) + + +def account_update_event_endpoint( + request: Request, data: EndpointBaseRequestModel +) -> Dict[str, Any]: + context_retriever = ContextRetrievers(func=account_update_event_endpoint) + event_2_catch = AccountRecordsUpdateEventMethods.retrieve_event( + event_function_code=f"{SuperUserAccountEvents.SuperUserUpdateEvent.key}" + ) + context_retriever.RESPONSE_VALIDATOR = event_2_catch.RESPONSE_VALIDATOR + data = event_2_catch.REQUEST_VALIDATOR(**data.data) + AccountListEventMethods.context_retriever = context_retriever + pagination_result = event_2_catch.endpoint_callable(data=data) + return EndpointSuccessListResponse( + code=event_2_catch.static_key, lang=context_retriever.token.lang + ).as_dict( + data=pagination_result.data, pagination=pagination_result.pagination.as_dict() + ) + + +AccountRecordsUpdateEventMethods.endpoint_callable = ( + account_update_event_endpoint +) + diff --git a/Events/AllEvents/events/account/api_events.py b/Events/AllEvents/events/account/api_events.py index 0070c93..6b27663 100644 --- a/Events/AllEvents/events/account/api_events.py +++ b/Events/AllEvents/events/account/api_events.py @@ -1,19 +1,13 @@ -from pydantic import BaseModel - -from ApiLayers.ApiValidations.Request import ListOptions -from ApiLayers.Schemas import AccountRecords -# from ApiLayers.LanguageModels.Request import ( -# LoginRequestLanguageModel, -# SelectRequestLanguageModel, -# ) from Events.Engine.abstract_class import Event from .models import AccountRequestValidators from .function_handlers import ( AccountListEventMethods, + AccountCreateEventMethods, + AccountUpdateEventMethods, ) -# + # class SelectResponseAccount(BaseModel): # """ # Response model for account list. @@ -24,8 +18,9 @@ from .function_handlers import ( # type_description: str # + # Auth Login -account_insert_super_user_event = Event( +account_list_super_user_event = Event( name="account_insert_super_user_event", key="36a165fe-a2f3-437b-80ee-1ee44670fe70", request_validator=AccountRequestValidators.ListAccountRecord, @@ -37,6 +32,45 @@ account_insert_super_user_event = Event( ) -account_insert_super_user_event.endpoint_callable = ( +account_list_super_user_event.endpoint_callable = ( AccountListEventMethods.account_records_list ) + + +account_insert_super_user_event = Event( + name="account_insert_super_user_event", + key="3aa46155-72bc-4370-b4e7-b937b0f9b893", + request_validator=AccountRequestValidators.InsertAccountRecord, + # response_validator=SelectResponseAccount, + # language_models=[AccountRecords.__language_model__], + language_models=[], + statics="ACCOUNT_CREATED", + description="Create a new account by validation list options and queries.", +) + + +account_insert_super_user_event.endpoint_callable = ( + AccountCreateEventMethods.account_records_create +) + + +account_update_super_user_event = Event( + name="account_insert_super_user_event", + key="3aa46155-72bc-4370-b4e7-b937b0f9b893", + request_validator=AccountRequestValidators.UpdateAccountRecord, + # response_validator=SelectResponseAccount, + # language_models=[AccountRecords.__language_model__], + language_models=[], + statics="ACCOUNT_UPDATED", + description="Update a specific account by validation list options and queries.", +) + +account_update_super_user_event.endpoint_callable = ( + AccountUpdateEventMethods.account_records_update +) + + +class SuperUserAccountEvents: + SuperUserListEvent = account_list_super_user_event + SuperUserCreateEvent = account_insert_super_user_event + SuperUserUpdateEvent = account_update_super_user_event diff --git a/Events/AllEvents/events/account/cluster.py b/Events/AllEvents/events/account/cluster.py index 3fe6b91..2a2257f 100644 --- a/Events/AllEvents/events/account/cluster.py +++ b/Events/AllEvents/events/account/cluster.py @@ -1,18 +1,25 @@ from Events.Engine.abstract_class import CategoryCluster -from .account_records import AccountRecordsEventMethods +from .account_records import ( + AccountRecordsListEventMethods, + AccountRecordsCreateEventMethods, + AccountRecordsUpdateEventMethods, +) from .info import account_page_info AccountCluster = CategoryCluster( name="AccountCluster", - tags=["accounts"], + tags=["Account Records"], prefix="/accounts", description="Account Cluster", pageinfo=account_page_info, endpoints={ - "AccountRecordsEventMethods": AccountRecordsEventMethods, + "AccountRecordsListEventMethods": AccountRecordsListEventMethods, + "AccountRecordsCreateEventMethods": AccountRecordsCreateEventMethods, + "AccountRecordsUpdateEventMethods": AccountRecordsUpdateEventMethods, }, include_in_schema=True, sub_category=[], + is_client=True, ) diff --git a/Events/AllEvents/events/account/function_handlers.py b/Events/AllEvents/events/account/function_handlers.py index 4f00439..edf01f9 100644 --- a/Events/AllEvents/events/account/function_handlers.py +++ b/Events/AllEvents/events/account/function_handlers.py @@ -291,3 +291,4 @@ class AccountUpdateEventMethods(BaseRouteModel): # cls_object=AccountRecords, # response_model=UpdateAccountRecord, # ) + diff --git a/Events/AllEvents/events/address/address.py b/Events/AllEvents/events/address/address.py index 7d3d19b..b51dbd2 100644 --- a/Events/AllEvents/events/address/address.py +++ b/Events/AllEvents/events/address/address.py @@ -1,351 +1,173 @@ """ - request models. +Account related API endpoints. """ +from typing import Any, Dict +from fastapi import Request -from typing import TYPE_CHECKING, Dict, Any, List, Optional, TypedDict, Union -from pydantic import BaseModel, Field, model_validator, RootModel, ConfigDict -from ApiEvents.abstract_class import MethodToEvent -from ApiEvents.base_request_model import BaseRequestModel, DictRequestModel -from ApiValidations.Custom.token_objects import EmployeeTokenObject, OccupantTokenObject -from ApiValidations.Request.address import SearchAddress, UpdateAddress -from ApiValidations.Request.base_validations import ListOptions -from ErrorHandlers.Exceptions.api_exc import HTTPExceptionApi -from Schemas.identity.identity import ( - AddressPostcode, - AddressStreet, - Addresses, - RelationshipEmployee2PostCode, +from Events.Engine.abstract_class import MethodToEvent +from Events.base_request_model import EndpointBaseRequestModel, ContextRetrievers + +from ApiLayers.Middleware.token_event_middleware import TokenEventMiddleware +from ApiLayers.ApiValidations.Response.default_response import EndpointSuccessListResponse + +from .function_handlers import ( + AddressListFunctions, + AddressUpdateFunctions, + AddressSearchFunctions, + AddressCreateFunctions, ) -from ApiValidations.Request import ( - InsertAddress, -) -from ApiValidations.Response import ( - ListAddressResponse, +from .api_events import AddressSuperUserEvents + + +AddressListEventMethods = MethodToEvent( + name="AddressListEventMethods", + events={ + AddressSuperUserEvents.AddressListEvents.key: AddressSuperUserEvents.AddressListEvents, + }, + headers=[], + errors=[], + decorators_list=[TokenEventMiddleware.event_required], + url="/list", + method="POST", + summary="List all accounts by given previligous", + description="List all accounts by given previligous", ) -if TYPE_CHECKING: - from fastapi import Request +def account_list_event_endpoint( + request: Request, data: EndpointBaseRequestModel +) -> Dict[str, Any]: + context_retriever = ContextRetrievers(func=account_list_event_endpoint) + event_2_catch = AddressListEventMethods.retrieve_event( + event_function_code=f"{AddressSuperUserEvents.AddressListEvents.key}" + ) + context_retriever.RESPONSE_VALIDATOR = event_2_catch.RESPONSE_VALIDATOR + data = event_2_catch.REQUEST_VALIDATOR(**data.data) + AddressListFunctions.context_retriever = context_retriever + pagination_result = event_2_catch.endpoint_callable(data=data) + return EndpointSuccessListResponse( + code=event_2_catch.static_key, lang=context_retriever.token.lang + ).as_dict( + data=pagination_result.data, pagination=pagination_result.pagination.as_dict() + ) -class AddressListEventMethod(MethodToEvent): - - event_type = "SELECT" - event_description = "List Address records" - event_category = "Address" - - __event_keys__ = { - "9c251d7d-da70-4d63-a72c-e69c26270442": "address_list_super_user", - "52afe375-dd95-4f4b-aaa2-4ec61bc6de52": "address_list_employee", - } - __event_validation__ = { - "9c251d7d-da70-4d63-a72c-e69c26270442": ( - ListAddressResponse, - [Addresses.__language_model__], - ), - "52afe375-dd95-4f4b-aaa2-4ec61bc6de52": ( - ListAddressResponse, - [Addresses.__language_model__], - ), - } - - @classmethod - def address_list_super_user( - cls, - list_options: ListOptions, - token_dict: Union[EmployeeTokenObject, OccupantTokenObject], - ): - db = RelationshipEmployee2PostCode.new_session() - post_code_list = RelationshipEmployee2PostCode.filter_all( - RelationshipEmployee2PostCode.company_id - == token_dict.selected_company.company_id, - db=db, - ).data - post_code_id_list = [post_code.member_id for post_code in post_code_list] - if not post_code_id_list: - raise HTTPExceptionApi( - status_code=404, - detail="User has no post code registered. User can not list addresses.", - ) - get_street_ids = [ - street_id[0] - for street_id in AddressPostcode.select_only( - AddressPostcode.id.in_(post_code_id_list), - select_args=[AddressPostcode.street_id], - order_by=AddressPostcode.street_id.desc(), - db=db, - ).data - ] - if not get_street_ids: - raise HTTPExceptionApi( - status_code=404, - detail="User has no street registered. User can not list addresses.", - ) - Addresses.pre_query = Addresses.filter_all( - Addresses.street_id.in_(get_street_ids), - db=db, - ).query - Addresses.filter_attr = list_options - records = Addresses.filter_all(db=db).data - return {} - # return AlchemyJsonResponse( - # completed=True, message="List Address records", result=records - # ) - - @classmethod - def address_list_employee( - cls, - list_options: ListOptions, - token_dict: Union[EmployeeTokenObject, OccupantTokenObject], - ): - # Addresses.filter_attr = list_options - Addresses.pre_query = Addresses.filter_all( - Addresses.street_id.in_(get_street_ids), - ) - records = Addresses.filter_all().data - return - # return AlchemyJsonResponse( - # completed=True, message="List Address records", result=records - # ) +AddressListEventMethods.endpoint_callable = ( + account_list_event_endpoint +) -class AddressCreateEventMethod(MethodToEvent): - - event_type = "CREATE" - event_description = "" - event_category = "" - - __event_keys__ = { - "ffdc445f-da10-4ce4-9531-d2bdb9a198ae": "create_address", - } - __event_validation__ = { - "ffdc445f-da10-4ce4-9531-d2bdb9a198ae": ( - InsertAddress, - [Addresses.__language_model__], - ), - } - - @classmethod - def create_address( - cls, - data: InsertAddress, - token_dict: Union[EmployeeTokenObject, OccupantTokenObject], - ): - post_code = AddressPostcode.filter_one( - AddressPostcode.uu_id == data.post_code_uu_id, - ).data - if not post_code: - raise HTTPExceptionApi( - status_code=404, - detail="Post code not found. User can not create address without post code.", - ) - - data_dict = data.excluded_dump() - data_dict["street_id"] = post_code.street_id - data_dict["street_uu_id"] = str(post_code.street_uu_id) - del data_dict["post_code_uu_id"] - address = Addresses.find_or_create(**data_dict) - address.save() - address.update(is_confirmed=True) - address.save() - return AlchemyJsonResponse( - completed=True, - message="Address created successfully", - result=address.get_dict(), - ) +AddressCreateEventMethods = MethodToEvent( + name="AddressCreateEventMethods", + events={ + AddressSuperUserEvents.AddressCreateEvents.key: AddressSuperUserEvents.AddressCreateEvents, + }, + headers=[], + errors=[], + decorators_list=[TokenEventMiddleware.event_required], + url="/create", + method="POST", + summary="Create Address via given data and previligous", + description="Create Address via given data and previligous", +) -class AddressSearchEventMethod(MethodToEvent): - """Event methods for searching addresses. - - This class handles address search functionality including text search - and filtering. - """ - - event_type = "SEARCH" - event_description = "Search for addresses using text and filters" - event_category = "Address" - - __event_keys__ = { - "e0ac1269-e9a7-4806-9962-219ac224b0d0": "search_address", - } - __event_validation__ = { - "e0ac1269-e9a7-4806-9962-219ac224b0d0": ( - SearchAddress, - [Addresses.__language_model__], - ), - } - - @classmethod - def _build_order_clause( - cls, filter_list: Dict[str, Any], schemas: List[str], filter_table: Any - ) -> Any: - """Build the ORDER BY clause for the query. - - Args: - filter_list: Dictionary of filter options - schemas: List of available schema fields - filter_table: SQLAlchemy table to query - - Returns: - SQLAlchemy order_by clause - """ - # Default to ordering by UUID if field not in schema - if filter_list.get("order_field") not in schemas: - filter_list["order_field"] = "uu_id" - else: - # Extract table and field from order field - table_name, field_name = str(filter_list.get("order_field")).split(".") - filter_table = getattr(databases.sql_models, table_name) - filter_list["order_field"] = field_name - - # Build order clause - field = getattr(filter_table, filter_list.get("order_field")) - return ( - field.desc() - if str(filter_list.get("order_type"))[0] == "d" - else field.asc() - ) - - @classmethod - def _format_record(cls, record: Any, schemas: List[str]) -> Dict[str, str]: - """Format a database record into a dictionary. - - Args: - record: Database record to format - schemas: List of schema fields - - Returns: - Formatted record dictionary - """ - result = {} - for index, schema in enumerate(schemas): - value = str(record[index]) - # Special handling for UUID fields - if "uu_id" in value: - value = str(value) - result[schema] = value - return result - - @classmethod - def search_address( - cls, - data: SearchAddress, - token_dict: Union[EmployeeTokenObject, OccupantTokenObject], - ) -> Any: - """Search for addresses using text search and filters. - - Args: - data: Search parameters including text and filters - token_dict: Authentication token - - Returns: - JSON response with search results - - Raises: - HTTPExceptionApi: If search fails - """ - try: - # Start performance measurement - start_time = perf_counter() - - # Get initial query - search_result = AddressStreet.search_address_text(search_text=data.search) - if not search_result: - raise HTTPExceptionApi( - status_code=status.HTTP_404_NOT_FOUND, - detail="No addresses found matching search criteria", - ) - - query = search_result.get("query") - schemas = search_result.get("schema") - - # Apply filters - filter_list = data.list_options.dump() - filter_table = AddressStreet - - # Build and apply order clause - order = cls._build_order_clause(filter_list, schemas, filter_table) - - # Apply pagination - page_size = int(filter_list.get("size")) - offset = (int(filter_list.get("page")) - 1) * page_size - - # Execute query - query = ( - query.order_by(order) - .limit(page_size) - .offset(offset) - .populate_existing() - ) - records = list(query.all()) - - # Format results - results = [cls._format_record(record, schemas) for record in records] - - # Log performance - duration = perf_counter() - start_time - print(f"Address search completed in {duration:.3f}s") - - return AlchemyJsonResponse( - completed=True, message="Address search results", result=results - ) - - except HTTPExceptionApi as e: - # Re-raise HTTP exceptions - raise e - except Exception as e: - # Log and wrap other errors - print(f"Address search error: {str(e)}") - raise HTTPExceptionApi( - status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, - detail="Failed to search addresses", - ) from e +def account_create_event_endpoint( + request: Request, data: EndpointBaseRequestModel +) -> Dict[str, Any]: + context_retriever = ContextRetrievers(func=account_create_event_endpoint) + event_2_catch = AddressCreateEventMethods.retrieve_event( + event_function_code=f"{AddressSuperUserEvents.AddressCreateEvents.key}" + ) + context_retriever.RESPONSE_VALIDATOR = event_2_catch.RESPONSE_VALIDATOR + data = event_2_catch.REQUEST_VALIDATOR(**data.data) + AddressCreateFunctions.context_retriever = context_retriever + pagination_result = event_2_catch.endpoint_callable(data=data) + return EndpointSuccessListResponse( + code=event_2_catch.static_key, lang=context_retriever.token.lang + ).as_dict( + data=pagination_result.data, pagination=pagination_result.pagination.as_dict() + ) -class AddressUpdateEventMethod(MethodToEvent): +AddressCreateEventMethods.endpoint_callable = ( + account_create_event_endpoint +) - event_type = "UPDATE" - event_description = "" - event_category = "" - __event_keys__ = { - "1f9c3a9c-e5bd-4dcd-9b9a-3742d7e03a27": "update_address", - } - __event_validation__ = { - "1f9c3a9c-e5bd-4dcd-9b9a-3742d7e03a27": ( - UpdateAddress, - [Addresses.__language_model__], - ), - } +AddressUpdateEventMethods = MethodToEvent( + name="AddressUpdateEventMethods", + events={ + AddressSuperUserEvents.AddressUpdateEvents.key: AddressSuperUserEvents.AddressUpdateEvents, + }, + headers=[], + errors=[], + decorators_list=[TokenEventMiddleware.event_required], + url="/update", + method="POST", + summary="Update Address via given data and previligous", + description="Update Address via given data and previligous", +) + + +def account_update_event_endpoint( + request: Request, data: EndpointBaseRequestModel +) -> Dict[str, Any]: + context_retriever = ContextRetrievers(func=account_update_event_endpoint) + event_2_catch = AddressUpdateEventMethods.retrieve_event( + event_function_code=f"{AddressSuperUserEvents.AddressUpdateEvents.key}" + ) + context_retriever.RESPONSE_VALIDATOR = event_2_catch.RESPONSE_VALIDATOR + data = event_2_catch.REQUEST_VALIDATOR(**data.data) + AddressUpdateFunctions.context_retriever = context_retriever + pagination_result = event_2_catch.endpoint_callable(data=data) + return EndpointSuccessListResponse( + code=event_2_catch.static_key, lang=context_retriever.token.lang + ).as_dict( + data=pagination_result.data, pagination=pagination_result.pagination.as_dict() + ) + + +AddressUpdateEventMethods.endpoint_callable = ( + account_update_event_endpoint +) + + +AddressSearchEventMethods = MethodToEvent( + name="AddressSearchEventMethods", + events={ + AddressSuperUserEvents.AddressSearchEvents.key: AddressSuperUserEvents.AddressSearchEvents, + }, + headers=[], + errors=[], + decorators_list=[TokenEventMiddleware.event_required], + url="/search", + method="POST", + summary="Search Address via given data and previligous", + description="Search Address via given data and previligous", +) + + +def address_search_event_endpoint( + request: Request, data: EndpointBaseRequestModel +) -> Dict[str, Any]: + context_retriever = ContextRetrievers(func=account_update_event_endpoint) + event_2_catch = AddressUpdateEventMethods.retrieve_event( + event_function_code=f"{AddressSuperUserEvents.AddressSearchEvents.key}" + ) + context_retriever.RESPONSE_VALIDATOR = event_2_catch.RESPONSE_VALIDATOR + data = event_2_catch.REQUEST_VALIDATOR(**data.data) + AddressSearchFunctions.context_retriever = context_retriever + pagination_result = event_2_catch.endpoint_callable(data=data) + return EndpointSuccessListResponse( + code=event_2_catch.static_key, lang=context_retriever.token.lang + ).as_dict( + data=pagination_result.data, pagination=pagination_result.pagination.as_dict() + ) + +AddressSearchEventMethods.endpoint_callable = ( + address_search_event_endpoint +) + - @classmethod - def update_address( - cls, - address_uu_id: str, - data: UpdateAddress, - token_dict: Union[EmployeeTokenObject, OccupantTokenObject], - ): - if isinstance(token_dict, EmployeeTokenObject): - address = Addresses.filter_one( - Addresses.uu_id == address_uu_id, - ).data - if not address: - raise HTTPExceptionApi( - status_code=404, - detail=f"Address not found. User can not update with given address uuid : {address_uu_id}", - ) - data_dict = data.excluded_dump() - updated_address = address.update(**data_dict) - updated_address.save() - return AlchemyJsonResponse( - completed=True, - message="Address updated successfully", - result=updated_address.get_dict(), - ) - elif isinstance(token_dict, OccupantTokenObject): - raise HTTPExceptionApi( - status_code=403, - detail="Occupant can not update address.", - ) diff --git a/Events/AllEvents/events/address/api_events.py b/Events/AllEvents/events/address/api_events.py index be66917..2fafdbd 100644 --- a/Events/AllEvents/events/address/api_events.py +++ b/Events/AllEvents/events/address/api_events.py @@ -3,19 +3,82 @@ from ApiLayers.LanguageModels.Request import ( LoginRequestLanguageModel, ) -from models import TemplateResponseModels, TemplateRequestModels -from function_handlers import TemplateFunctions +# from models import TemplateResponseModels, TemplateRequestModels +from .function_handlers import AddressSuperUserFunctions -# Auth Login -template_event = Event( - name="authentication_login_super_user_event", - key="a5d2d0d1-3e9b-4b0f-8c7d-6d4a4b4c4d4e", - request_validator=TemplateRequestModels.TemplateRequestModelX, - language_models=[LoginRequestLanguageModel], - response_validation_static="LOGIN_SUCCESS", - description="Login super user", +# Address List for super_user event +address_list_super_user_event = Event( + name="account_insert_super_user_event", + key="7ce855ce-db79-4397-b0ec-f5e408ea6447", + # request_validator=AccountRequestValidators.ListAccountRecord, + # response_validator=SelectResponseAccount, + # language_models=[AccountRecords.__language_model__], + language_models=[], + statics="", + description="List address by validation list options and queries.", ) -template_event.endpoint_callable = TemplateFunctions.template_example_function() +address_list_super_user_event.endpoint_callable = ( + AddressSuperUserFunctions.AddressListFunctions.template_example_function_list +) + +# Address Create for super_user event +address_create_super_user_event = Event( + name="account_insert_super_user_event", + key="d638a6b2-cf2e-4361-99a4-021183b75ec1", + # request_validator=AccountRequestValidators.ListAccountRecord, + # response_validator=SelectResponseAccount, + # language_models=[AccountRecords.__language_model__], + language_models=[], + statics="", + description="Create address by validation list options and queries.", +) + + +address_create_super_user_event.endpoint_callable = ( + AddressSuperUserFunctions.AddressCreateFunctions.template_example_function_list +) + + +# Address Update for super_user event +address_update_super_user_event = Event( + name="account_insert_super_user_event", + key="d638a6b2-cf2e-4361-99a4-021183b75ec1", + # request_validator=AccountRequestValidators.ListAccountRecord, + # response_validator=SelectResponseAccount, + # language_models=[AccountRecords.__language_model__], + language_models=[], + statics="", + description="Update address by validation list options and queries.", +) + + +address_update_super_user_event.endpoint_callable = ( + AddressSuperUserFunctions.AddressUpdateFunctions.template_example_function_list +) + + +# Address Update for super_user event +address_search_super_user_event = Event( + name="account_insert_super_user_event", + key="d638a6b2-cf2e-4361-99a4-021183b75ec1", + # request_validator=AccountRequestValidators.ListAccountRecord, + # response_validator=SelectResponseAccount, + # language_models=[AccountRecords.__language_model__], + language_models=[], + statics="", + description="Search address by validation list options and queries.", +) + + +address_search_super_user_event.endpoint_callable = ( + AddressSuperUserFunctions.AddressSearchFunctions.template_example_function_list +) + +class AddressSuperUserEvents: + AddressListEvents = address_list_super_user_event + AddressCreateEvents = address_create_super_user_event + AddressUpdateEvents = address_update_super_user_event + AddressSearchEvents = address_search_super_user_event \ No newline at end of file diff --git a/Events/AllEvents/events/address/cluster.py b/Events/AllEvents/events/address/cluster.py index de5066b..db9f8de 100644 --- a/Events/AllEvents/events/address/cluster.py +++ b/Events/AllEvents/events/address/cluster.py @@ -1,14 +1,27 @@ from Events.Engine.abstract_class import CategoryCluster -from info import template_page_info + +from .address import ( +AddressListEventMethods, +AddressCreateEventMethods, +AddressUpdateEventMethods, +AddressSearchEventMethods, +) +from .info import address_page_info -TemplateCluster = CategoryCluster( - name="TemplateCluster", - tags=["template"], - prefix="/template", - description="Template cluster", - pageinfo=template_page_info, - endpoints={}, +AddressCluster = CategoryCluster( + name="AddressCluster", + tags=["Address"], + prefix="/address", + description="Address Cluster", + pageinfo=address_page_info, + endpoints={ + "AddressListEventMethods": AddressListEventMethods, + "AddressCreateEventMethods": AddressCreateEventMethods, + "AddressUpdateEventMethods": AddressUpdateEventMethods, + "AddressSearchEventMethods": AddressSearchEventMethods, + }, include_in_schema=True, sub_category=[], + is_client=True, ) diff --git a/Events/AllEvents/events/address/function_handlers.py b/Events/AllEvents/events/address/function_handlers.py index 4c8fbcd..acdb195 100644 --- a/Events/AllEvents/events/address/function_handlers.py +++ b/Events/AllEvents/events/address/function_handlers.py @@ -3,6 +3,7 @@ from typing import Union, Optional from ApiLayers.ApiValidations.Request import ListOptions from Events.base_request_model import BaseRouteModel, ListOptionsBase from Services.PostgresDb.Models.pagination import PaginationResult +from ApiLayers.Schemas import AddressNeighborhood class Handlers: @@ -14,38 +15,32 @@ class Handlers: return -class TemplateFunctions(BaseRouteModel): - """ - Class for handling authentication functions - Is a template 4 TokenMiddleware.event_required decorator function groups. - results as : - STATIC_MESSAGE & LANG retrieved from redis - { - "completed": true, - "message": STATIC_MESSAGE, - "lang": LANG, - "pagination": { - "size": 10, - "page": 2, - "allCount": 28366, - "totalCount": 18, - "totalPages": 2, - "pageCount": 8, - "orderField": ["type_code", "neighborhood_name"], - "orderType": ["asc", "desc"] - }, - "data": [ - { - "created_at": "2025-01-12 09:39:48 +00:00", - "active": true, - "expiry_starts": "2025-01-12 09:39:48 +00:00", - "locality_uu_id": "771fd152-aca1-4d75-a42e-9b29ea7112b5", - "uu_id": "e1baa3bc-93ce-4099-a078-a11b71d3b1a8" - }, - ... - ] - } - """ +class AddressListFunctions(BaseRouteModel): + + + @classmethod + def template_example_function_list(cls, data: Optional[Union[dict, ListOptions]]) -> PaginationResult: + list_options_base = ListOptionsBase( + table=AddressNeighborhood, list_options=data, model_query=None, + ) + db_session, query_options = list_options_base.init_list_options() + if cls.context_retriever.token.is_occupant: + AddressNeighborhood.pre_query = AddressNeighborhood.filter_all( + AddressNeighborhood.neighborhood_code.icontains("10"), + db=db_session, + ).query + elif cls.context_retriever.token.is_employee: + AddressNeighborhood.pre_query = AddressNeighborhood.filter_all( + AddressNeighborhood.neighborhood_code.icontains("9"), + db=db_session, + ).query + records = AddressNeighborhood.filter_all(*query_options.convert(), db=db_session) + return list_options_base.paginated_result( + records=records, response_model=getattr(cls.context_retriever, 'RESPONSE_VALIDATOR', None) + ) + + +class AddressCreateFunctions(BaseRouteModel): @classmethod def template_example_function_list(cls, data: Optional[Union[dict, ListOptions]]) -> PaginationResult: @@ -67,4 +62,67 @@ class TemplateFunctions(BaseRouteModel): records = AddressNeighborhood.filter_all(*query_options.convert(), db=db_session) return list_options_base.paginated_result( records=records, response_model=getattr(cls.context_retriever, 'RESPONSE_VALIDATOR', None) - ) \ No newline at end of file + ) + + +class AddressSearchFunctions(BaseRouteModel): + """Event methods for searching addresses. + + This class handles address search functionality including text search + and filtering. + """ + + + @classmethod + def template_example_function_list(cls, data: Optional[Union[dict, ListOptions]]) -> PaginationResult: + from ApiLayers.Schemas import AddressNeighborhood + list_options_base = ListOptionsBase( + table=AddressNeighborhood, list_options=data, model_query=None, + ) + db_session, query_options = list_options_base.init_list_options() + if cls.context_retriever.token.is_occupant: + AddressNeighborhood.pre_query = AddressNeighborhood.filter_all( + AddressNeighborhood.neighborhood_code.icontains("10"), + db=db_session, + ).query + elif cls.context_retriever.token.is_employee: + AddressNeighborhood.pre_query = AddressNeighborhood.filter_all( + AddressNeighborhood.neighborhood_code.icontains("9"), + db=db_session, + ).query + records = AddressNeighborhood.filter_all(*query_options.convert(), db=db_session) + return list_options_base.paginated_result( + records=records, response_model=getattr(cls.context_retriever, 'RESPONSE_VALIDATOR', None) + ) + + +class AddressUpdateFunctions(BaseRouteModel): + + @classmethod + def template_example_function_list(cls, data: Optional[Union[dict, ListOptions]]) -> PaginationResult: + from ApiLayers.Schemas import AddressNeighborhood + list_options_base = ListOptionsBase( + table=AddressNeighborhood, list_options=data, model_query=None, + ) + db_session, query_options = list_options_base.init_list_options() + if cls.context_retriever.token.is_occupant: + AddressNeighborhood.pre_query = AddressNeighborhood.filter_all( + AddressNeighborhood.neighborhood_code.icontains("10"), + db=db_session, + ).query + elif cls.context_retriever.token.is_employee: + AddressNeighborhood.pre_query = AddressNeighborhood.filter_all( + AddressNeighborhood.neighborhood_code.icontains("9"), + db=db_session, + ).query + records = AddressNeighborhood.filter_all(*query_options.convert(), db=db_session) + return list_options_base.paginated_result( + records=records, response_model=getattr(cls.context_retriever, 'RESPONSE_VALIDATOR', None) + ) + + +class AddressSuperUserFunctions: + AddressListFunctions = AddressListFunctions + AddressCreateFunctions = AddressCreateFunctions + AddressSearchFunctions = AddressSearchFunctions + AddressUpdateFunctions = AddressUpdateFunctions diff --git a/Events/AllEvents/events/address/info.py b/Events/AllEvents/events/address/info.py index 89d163f..ceb220a 100644 --- a/Events/AllEvents/events/address/info.py +++ b/Events/AllEvents/events/address/info.py @@ -1,7 +1,7 @@ from Events.Engine.abstract_class import PageInfo -template_page_info = PageInfo( +address_page_info = PageInfo( name="template", title={"en": "template"}, description={"en": "template"}, diff --git a/Events/AllEvents/validations/validation/api_events.py b/Events/AllEvents/validations/validation/api_events.py index 08ddeeb..ca8aee0 100644 --- a/Events/AllEvents/validations/validation/api_events.py +++ b/Events/AllEvents/validations/validation/api_events.py @@ -7,7 +7,7 @@ from .models import ValidationsPydantic from .function_handlers import RetrieveValidation -# Auth Login +# Validation Event validation_event = Event( name="validation_event", key="02b5a596-14ba-4361-90d7-c6755727c63f", @@ -23,3 +23,21 @@ def get_validation_by_event_function_code(request: Request, data: Any): validation_event.endpoint_callable = get_validation_by_event_function_code + + +# Menu Event +menu_event = Event( + name="menu_event", + key="a1613ca0-4843-498b-bfff-07ecea6777b2", + request_validator=ValidationsPydantic, + language_models=[], + statics=None, + description="Get Left Menu of the user", +) + + +def get_menu_by_event_function_code(request: Request, data: Any): + return RetrieveValidation.retrieve_validation(data=data) + + +menu_event.endpoint_callable = get_menu_by_event_function_code diff --git a/Events/AllEvents/validations/validation/cluster.py b/Events/AllEvents/validations/validation/cluster.py index ee4d7d6..2aeaf86 100644 --- a/Events/AllEvents/validations/validation/cluster.py +++ b/Events/AllEvents/validations/validation/cluster.py @@ -1,6 +1,9 @@ from Events.Engine.abstract_class import CategoryCluster -from .validation import ValidationEventMethods +from .validation import ( + ValidationEventMethods, + MenuEventMethods, +) ValidationsCluster = CategoryCluster( @@ -10,6 +13,7 @@ ValidationsCluster = CategoryCluster( description="Validations cluster", endpoints={ "ValidationEventMethods": ValidationEventMethods, + "MenuEventMethods": MenuEventMethods, }, include_in_schema=True, sub_category=[], diff --git a/Events/AllEvents/validations/validation/validation.py b/Events/AllEvents/validations/validation/validation.py index 1854206..709d1a7 100644 --- a/Events/AllEvents/validations/validation/validation.py +++ b/Events/AllEvents/validations/validation/validation.py @@ -1,5 +1,5 @@ """ -template related API endpoints. +Validation related API endpoints. """ from typing import Any, Dict @@ -9,7 +9,7 @@ from Events.Engine.abstract_class import MethodToEvent from Events.base_request_model import EndpointBaseRequestModel, ContextRetrievers from ApiLayers.Middleware.auth_middleware import MiddlewareModule -from .api_events import validation_event +from .api_events import validation_event, menu_event from .function_handlers import RetrieveValidation @@ -26,19 +26,41 @@ ValidationEventMethods = MethodToEvent( ) -def authentication_login_with_domain_and_creds_endpoint( +def validations_endpoint( request: Request, data: EndpointBaseRequestModel ) -> Dict[str, Any]: function = ValidationEventMethods.retrieve_event( event_function_code=f"{validation_event.key}" ) data = function.REQUEST_VALIDATOR(**data.data) - RetrieveValidation.context_retriever = ContextRetrievers( - func=authentication_login_with_domain_and_creds_endpoint - ) + RetrieveValidation.context_retriever = ContextRetrievers(func=validations_endpoint) return function.endpoint_callable(request=request, data=data) -ValidationEventMethods.endpoint_callable = ( - authentication_login_with_domain_and_creds_endpoint +ValidationEventMethods.endpoint_callable = validations_endpoint + + +MenuEventMethods = MethodToEvent( + name="MenuEventMethods", + events={menu_event.key: menu_event}, + headers=[], + errors=[], + url="/menu", + method="POST", + decorators_list=[MiddlewareModule.auth_required], + summary="Get Left Menu of the user", + description="Get Left Menu of the user", ) + + +def menu_endpoint( + request: Request, data: EndpointBaseRequestModel +) -> Dict[str, Any]: + function = ValidationEventMethods.retrieve_event( + event_function_code=f"{menu_event.key}" + ) + data = function.REQUEST_VALIDATOR(**data.data) + RetrieveValidation.context_retriever = ContextRetrievers(func=validations_endpoint) + return function.endpoint_callable(request=request, data=data) + +MenuEventMethods.endpoint_callable = menu_endpoint diff --git a/Events/Engine/abstract_class.py b/Events/Engine/abstract_class.py index 40ccb90..fefc73b 100644 --- a/Events/Engine/abstract_class.py +++ b/Events/Engine/abstract_class.py @@ -80,6 +80,10 @@ class Event: return static_response.first return None + @property + def static_key(self): + return self.STATICS + @property def description(self): return f"This is an event of {self.name}. Description: {self.DESCRIPTION}" @@ -186,6 +190,7 @@ class CategoryCluster: ENDPOINTS: dict[str, MethodToEvent] # {"MethodToEvent": MethodToEvent, ...} SUBCATEGORY: Optional[List["CategoryCluster"]] # [CategoryCluster, ...] INCLUDE_IN_SCHEMA: Optional[bool] = True + IS_CLIENT: Optional[bool] = False def __init__( self, @@ -197,6 +202,7 @@ class CategoryCluster: sub_category: list, pageinfo: Optional[PageInfo] = None, include_in_schema: Optional[bool] = True, + is_client: Optional[bool] = False, ): self.NAME = name self.TAGS = tags @@ -206,6 +212,11 @@ class CategoryCluster: self.ENDPOINTS = endpoints or {} self.SUBCATEGORY = sub_category or [] self.INCLUDE_IN_SCHEMA = include_in_schema + self.IS_CLIENT = is_client + + @property + def is_client(self): + return self.IS_CLIENT @property def name(self): diff --git a/Events/base_request_model.py b/Events/base_request_model.py index abe2cb4..7aebf94 100644 --- a/Events/base_request_model.py +++ b/Events/base_request_model.py @@ -36,7 +36,7 @@ class ContextRetrievers: key_: str = "" RESPONSE_VALIDATOR = None - def __init__(self, func): + def __init__(self, func, statics: Optional[str] = None): self.func = func if hasattr(self.func, "auth_context"): self.is_auth = True