Berkay
/
wag-managment-api-service-version-5
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

events updated

This commit is contained in:
berkay 2025-01-31 16:29:39 +03:00
parent df82daea86
commit f0613e96e8
6 changed files with 466 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

21
Events/AllEvents/events/building/building/api_events.py Normal file
View File

@ -0,0 +1,21 @@
from Events.Engine.abstract_class import Event
from ApiLayers.LanguageModels.Request import (
LoginRequestLanguageModel,
)
# from models import TemplateResponseModels, TemplateRequestModels
from function_handlers import TemplateFunctions
# Auth Login
template_event = Event(
name="authentication_login_super_user_event",
key="a5d2d0d1-3e9b-4b0f-8c7d-6d4a4b4c4d4e",
# request_validator=TemplateRequestModels.TemplateRequestModelX,
language_models=[LoginRequestLanguageModel],
statics="LOGIN_SUCCESS",
description="Login super user",
)
template_event.endpoint_callable = TemplateFunctions.template_example_function_list

19
Events/AllEvents/events/building/building/building.py Normal file
View File

@ -0,0 +1,19 @@
"""
request models.
"""
from typing import TYPE_CHECKING, Dict, Any, Literal, Optional, TypedDict, Union
from pydantic import BaseModel, Field, model_validator, RootModel, ConfigDict
from ApiEvents.base_request_model import BaseRequestModel, DictRequestModel
from ApiValidations.Custom.token_objects import EmployeeTokenObject, OccupantTokenObject
from ApiValidations.Request.base_validations import ListOptions
from ErrorHandlers.Exceptions.api_exc import HTTPExceptionApi
from Schemas.identity.identity import (
AddressPostcode,
Addresses,
RelationshipEmployee2PostCode,
)
if TYPE_CHECKING:
from fastapi import Request

14
Events/AllEvents/events/building/building/cluster.py Normal file
View File

@ -0,0 +1,14 @@
from Events.Engine.abstract_class import CategoryCluster
from .info import building_page_info
BuildingCluster = CategoryCluster(
name="BuildingCluster",
tags=["Building"],
prefix="/building",
description="Building Cluster",
pageinfo=building_page_info,
endpoints={},
include_in_schema=True,
sub_category=[],
)

70
Events/AllEvents/events/building/building/function_handlers.py Normal file
View File

@ -0,0 +1,70 @@
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
class Handlers:
"""Class for handling authentication functions"""
@classmethod # Requires no auth context
def handle_function(cls, **kwargs):
"""Handle function with kwargs"""
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"
},
...
]
}
"""
@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)
)

17
Events/AllEvents/events/building/building/info.py Normal file
View File

@ -0,0 +1,17 @@
from Events.Engine.abstract_class import PageInfo
building_page_info = PageInfo(
name="Building",
title={
"en": "Building Actions",
"tr": "Bina İşlemleri"
},
description={
"en": "Building Actions to manage build updates",
"tr": "Bina güncellemelerini yönetmek için Bina İşlemleri"
},
icon="",
parent="",
url="",
)

325
Events/AllEvents/events/building/building/models.py Normal file
View File

@ -0,0 +1,325 @@
"""
request models.
"""
from typing import TYPE_CHECKING, Dict, Any, Literal, Optional, TypedDict, Union
from pydantic import BaseModel, Field, model_validator, RootModel, ConfigDict
from ApiEvents.base_request_model import BaseRequestModel, DictRequestModel
from ApiValidations.Custom.token_objects import EmployeeTokenObject, OccupantTokenObject
from ApiValidations.Request.base_validations import ListOptions
from ErrorHandlers.Exceptions.api_exc import HTTPExceptionApi
from Schemas.identity.identity import (
AddressPostcode,
Addresses,
RelationshipEmployee2PostCode,
)
if TYPE_CHECKING:
from fastapi import Request
class AddressListEventMethods(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,
"52afe375-dd95-4f4b-aaa2-4ec61bc6de52": ListAddressResponse,
}
@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(),
).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),
).query
Addresses.filter_attr = list_options
records = Addresses.filter_all().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
# )
class AddressCreateEventMethods(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,
}
@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(),
)
class AddressSearchEventMethods(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,
}
@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],
) -> JSONResponse:
"""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
class AddressUpdateEventMethods(MethodToEvent):
event_type = "UPDATE"
event_description = ""
event_category = ""
__event_keys__ = {
"1f9c3a9c-e5bd-4dcd-9b9a-3742d7e03a27": "update_address",
}
__event_validation__ = {
"1f9c3a9c-e5bd-4dcd-9b9a-3742d7e03a27": UpdateAddress,
}
@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.",
)