542 lines
17 KiB
Python
542 lines
17 KiB
Python
from typing import Union, Dict, Any, List
|
|
|
|
from fastapi import status
|
|
from fastapi.responses import JSONResponse
|
|
|
|
from api_validations.validations_response import (
|
|
ListAddressResponse,
|
|
AddressPostCodeResponse,
|
|
)
|
|
from databases import (
|
|
AddressDistrict,
|
|
AddressCity,
|
|
AddressCountry,
|
|
AddressStreet,
|
|
AddressNeighborhood,
|
|
AddressPostcode,
|
|
# AddressCountryCode,
|
|
RelationshipEmployee2PostCode,
|
|
Addresses,
|
|
)
|
|
|
|
from api_validations.validations_request import (
|
|
ListOptions,
|
|
InsertAddress,
|
|
UpdateAddress,
|
|
InsertPostCode,
|
|
UpdatePostCode,
|
|
SearchAddress,
|
|
)
|
|
from ApiServices.api_handlers import AlchemyJsonResponse
|
|
from api_events.events.abstract_class import MethodToEvent, ActionsSchema
|
|
from api_objects.auth.token_objects import EmployeeTokenObject, OccupantTokenObject
|
|
|
|
|
|
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],
|
|
):
|
|
post_code_list = RelationshipEmployee2PostCode.filter_all(
|
|
RelationshipEmployee2PostCode.company_id
|
|
== token_dict.selected_company.company_id,
|
|
).data
|
|
post_code_id_list = [post_code.member_id for post_code in post_code_list]
|
|
if not post_code_id_list:
|
|
raise HTTPException(
|
|
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 HTTPException(
|
|
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 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 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 HTTPException(
|
|
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:
|
|
HTTPException: 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 HTTPException(
|
|
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 HTTPException 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 HTTPException(
|
|
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 HTTPException(
|
|
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 HTTPException(
|
|
status_code=403,
|
|
detail="Occupant can not update address.",
|
|
)
|
|
|
|
|
|
class AddressPatchEventMethods(MethodToEvent):
|
|
|
|
event_type = "PATCH"
|
|
event_description = ""
|
|
event_category = ""
|
|
|
|
__event_keys__ = {
|
|
"b0e55a7e-af81-468c-b46c-a6b3a6b68d5d": "patch_address",
|
|
}
|
|
__event_validation__ = {
|
|
"b0e55a7e-af81-468c-b46c-a6b3a6b68d5d": None,
|
|
}
|
|
|
|
@classmethod
|
|
def patch_address(
|
|
cls,
|
|
address_uu_id: str,
|
|
data: InsertAddress,
|
|
token_dict: Union[EmployeeTokenObject, OccupantTokenObject],
|
|
):
|
|
address = Addresses.filter_one(
|
|
Addresses.uu_id == address_uu_id,
|
|
).data
|
|
post_code = RelationshipEmployee2PostCode.filter_one(
|
|
RelationshipEmployee2PostCode.member_id == address.post_code_id,
|
|
)
|
|
if not post_code:
|
|
raise HTTPException(
|
|
status_code=404,
|
|
detail="Post code not found. User can not patch address without post code.",
|
|
)
|
|
|
|
data_dict = data.excluded_dump()
|
|
data_dict["post_code_id"] = post_code.id
|
|
del data_dict["post_code_uu_id"]
|
|
|
|
patched_address = address.patch(**data_dict)
|
|
return AlchemyJsonResponse(
|
|
completed=True,
|
|
message="Address patched successfully",
|
|
result=patched_address.get_dict(),
|
|
)
|
|
|
|
|
|
class AddressPostCodeCreateEventMethods(MethodToEvent):
|
|
|
|
event_type = "CREATE"
|
|
event_description = ""
|
|
event_category = ""
|
|
|
|
__event_keys__ = {
|
|
"6f1406ac-577d-4f2c-8077-71fff2252c5f": "create_post_code_address",
|
|
}
|
|
__event_validation__ = {
|
|
"6f1406ac-577d-4f2c-8077-71fff2252c5f": InsertPostCode,
|
|
}
|
|
|
|
@classmethod
|
|
def create_post_code_address(
|
|
cls,
|
|
data: InsertPostCode,
|
|
token_dict: Union[EmployeeTokenObject, OccupantTokenObject],
|
|
):
|
|
data_dump = data.excluded_dump()
|
|
street = AddressStreet.filter_one(
|
|
AddressStreet.uu_id == data.street_uu_id,
|
|
).data
|
|
if not street:
|
|
raise HTTPException(
|
|
status_code=404,
|
|
detail="Street not found. User can not create post code without street.",
|
|
)
|
|
data_dump["street_id"] = street.id
|
|
data_dump["postcode"] = data.post_code
|
|
del data_dump["post_code"]
|
|
|
|
post_code = AddressPostcode.find_or_create(**data_dump)
|
|
relation_table = AddressPostcode.__many__table__.find_or_create(
|
|
member_id=post_code.id,
|
|
employee_id=token_dict.selected_company.employee_id,
|
|
company_id=token_dict.selected_company.company_id,
|
|
)
|
|
post_code.save()
|
|
post_code.update(is_confirmed=True)
|
|
post_code.save()
|
|
relation_table.update(is_confirmed=True)
|
|
relation_table.save()
|
|
return AlchemyJsonResponse(
|
|
completed=True,
|
|
message="Post code created successfully",
|
|
result=post_code.get_dict(),
|
|
)
|
|
|
|
|
|
class AddressPostCodeUpdateEventMethods(MethodToEvent):
|
|
|
|
event_type = "UPDATE"
|
|
event_description = ""
|
|
event_category = ""
|
|
|
|
__event_keys__ = {
|
|
"df18e489-a63c-477f-984c-aa52d30640ad": "update_post_code_address",
|
|
}
|
|
__event_validation__ = {
|
|
"df18e489-a63c-477f-984c-aa52d30640ad": UpdatePostCode,
|
|
}
|
|
|
|
@classmethod
|
|
def update_post_code_address(
|
|
cls,
|
|
post_code_uu_id: str,
|
|
data: UpdatePostCode,
|
|
token_dict: Union[EmployeeTokenObject, OccupantTokenObject],
|
|
):
|
|
if isinstance(token_dict, EmployeeTokenObject):
|
|
AddressPostcode.pre_query = AddressPostcode.select_action(
|
|
employee_id=token_dict.selected_company.employee_id,
|
|
)
|
|
post_code = AddressPostcode.filter_one(
|
|
AddressPostcode.uu_id == post_code_uu_id,
|
|
).data
|
|
if not post_code:
|
|
raise HTTPException(
|
|
status_code=404,
|
|
detail="Street not found. User can not update post code without street.",
|
|
)
|
|
|
|
data_dict = data.excluded_dump()
|
|
updated_post_code = post_code.update(**data_dict)
|
|
updated_post_code.save()
|
|
return AlchemyJsonResponse(
|
|
completed=True,
|
|
message="Post code updated successfully",
|
|
result=updated_post_code.get_dict(),
|
|
)
|
|
elif isinstance(token_dict, OccupantTokenObject):
|
|
raise HTTPException(
|
|
status_code=403,
|
|
detail="Occupant can not update post code.",
|
|
)
|
|
return AlchemyJsonResponse(
|
|
completed=True, message="Update Post Code record", result={}
|
|
)
|
|
|
|
|
|
class AddressPostCodeListEventMethods(MethodToEvent):
|
|
|
|
event_type = "SELECT"
|
|
event_description = ""
|
|
event_category = ""
|
|
|
|
__event_keys__ = {
|
|
"88d37b78-1ac4-4513-9d25-090ac3a24f31": "list_post_code_address",
|
|
}
|
|
__event_validation__ = {
|
|
"88d37b78-1ac4-4513-9d25-090ac3a24f31": AddressPostCodeResponse,
|
|
}
|
|
|
|
@classmethod
|
|
def list_post_code_address(
|
|
cls,
|
|
list_options: ListOptions,
|
|
token_dict: Union[EmployeeTokenObject, OccupantTokenObject],
|
|
):
|
|
post_code_list = AddressPostcode.__many__table__.filter_all(
|
|
AddressPostcode.__many__table__.company_id
|
|
== token_dict.selected_company.company_id,
|
|
).data
|
|
if not post_code_list:
|
|
raise HTTPException(
|
|
status_code=404,
|
|
detail="User has no post code registered or not yet any post code created.",
|
|
)
|
|
|
|
AddressPostcode.filter_attr = list_options
|
|
post_codes = AddressPostcode.filter_all()
|
|
return AlchemyJsonResponse(
|
|
completed=True, message="List Post code records", result=post_codes
|
|
)
|
|
|
|
|
|
AddressListEventMethod = AddressListEventMethods(
|
|
action=ActionsSchema(endpoint="/address/list")
|
|
)
|
|
AddressCreateEventMethod = AddressCreateEventMethods(
|
|
action=ActionsSchema(endpoint="/address/create")
|
|
)
|
|
AddressUpdateEventMethod = AddressUpdateEventMethods(
|
|
action=ActionsSchema(endpoint="/address/update")
|
|
)
|
|
AddressPatchEventMethod = AddressPatchEventMethods(
|
|
action=ActionsSchema(endpoint="/address/patch")
|
|
)
|
|
AddressPostCodeCreateEventMethod = AddressPostCodeCreateEventMethods(
|
|
action=ActionsSchema(endpoint="/postcode/create")
|
|
)
|
|
AddressPostCodeUpdateEventMethod = AddressPostCodeUpdateEventMethods(
|
|
action=ActionsSchema(endpoint="/postcode/update")
|
|
)
|
|
AddressPostCodeListEventMethod = AddressPostCodeListEventMethods(
|
|
action=ActionsSchema(endpoint="/postcode/list")
|
|
)
|
|
AddressSearchEventMethod = AddressSearchEventMethods(
|
|
action=ActionsSchema(endpoint="/address/search")
|
|
)
|