alcehmy and event functions updated

api_events/events/address/address.py

@@ -1,3 +1,5 @@
+from typing import Union
+
 from fastapi.exceptions import HTTPException
 from fastapi.responses import JSONResponse
 
@@ -31,7 +33,11 @@ class AddressListEventMethods(MethodToEvent):
     }
 
     @classmethod
-    def address_list_super_user(cls, list_options: ListOptions, token_dict):
+    def address_list_super_user(
+        cls,
+        list_options: ListOptions,
+        token_dict: Union[EmployeeTokenObject, OccupantTokenObject],
+    ):
         from sqlalchemy import select
 
         post_code_list = RelationshipEmployee2PostCode.filter_all(
@@ -45,17 +51,29 @@ class AddressListEventMethods(MethodToEvent):
                 detail="User has no post code registered. User can not list addresses.",
             )
 
-        get_street_ids = AddressPostcode.session.execute(
-            select(AddressPostcode.street_id).where(
-                AddressPostcode.id.in_(post_code_id_list)
+        get_street_ids = [
+            street_id[0]
+            for street_id in AddressPostcode.select_only(
+                AddressPostcode.id.in_(post_code_id_list),
+                AddressPostcode.active == True,
+                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.",
             )
-        ).all()
-        Addresses.pre_query = Addresses.filter_active(
-            Addresses.street_id.in_(*get_street_ids) if get_street_ids else None
+
+        Addresses.pre_query = Addresses.filter_all(
+            Addresses.street_id.in_(*get_street_ids),
+            Addresses.active == True,
         ).query
         Addresses.filter_attr = list_options
-        records = Addresses.filter_active(
-            *Addresses.get_smart_query(list_options.query)
+        records = Addresses.filter_all(
+            *Addresses.get_smart_query(list_options.query),
+            Addresses.active == True,
         )
         return AlchemyJsonResponse(
             completed=True,
@@ -64,7 +82,11 @@ class AddressListEventMethods(MethodToEvent):
         )
 
     @classmethod
-    def address_list_employee(cls, list_options: ListOptions, token_dict):
+    def address_list_employee(
+        cls,
+        list_options: ListOptions,
+        token_dict: Union[EmployeeTokenObject, OccupantTokenObject],
+    ):
         Addresses.filter_attr = list_options
         records = Addresses.list_via_employee(
             token_dict=token_dict,
@@ -76,6 +98,7 @@ class AddressListEventMethods(MethodToEvent):
             result=records,
         )
 
+
 class AddressCreateEventMethods(MethodToEvent):
 
     event_type = "CREATE"
@@ -84,8 +107,15 @@ class AddressCreateEventMethods(MethodToEvent):
     }
 
     @classmethod
-    def create_address(cls, data: InsertAddress, token_dict):
-        post_code = AddressPostcode.find_one(uu_id=data.post_code_uu_id)
+    def create_address(
+        cls,
+        data: InsertAddress,
+        token_dict: Union[EmployeeTokenObject, OccupantTokenObject],
+    ):
+        post_code = AddressPostcode.filter_one(
+            AddressPostcode.uu_id==data.post_code_uu_id,
+            AddressPostcode.active == True,
+        ).data
         if not post_code:
             raise HTTPException(
                 status_code=404,
@@ -95,15 +125,14 @@ class AddressCreateEventMethods(MethodToEvent):
         data_dict = data.excluded_dump()
         data_dict["street_id"] = post_code.street_id
         del data_dict["post_code_uu_id"]
-
+        data_dict["is_confirmed"] = True
         address = Addresses.find_or_create(**data_dict)
-        if not address.is_found:
-            RelationshipEmployee2PostCode.find_or_create(
-                employee_id=token_dict.selected_company.employee_id,
-                member_id=post_code.id,
-                company_id=token_dict.selected_company.company_id,
-                is_confirmed=True,
-            )
+        RelationshipEmployee2PostCode.find_or_create(
+            employee_id=token_dict.selected_company.employee_id,
+            member_id=post_code.id,
+            company_id=token_dict.selected_company.company_id,
+            is_confirmed=True,
+        )
         Addresses.save()
         return JSONResponse(
             content={
@@ -123,7 +152,11 @@ class AddressSearchEventMethods(MethodToEvent):
     }
 
     @classmethod
-    def search_address(cls, data: SearchAddress, token_dict):
+    def search_address(
+        cls,
+        data: SearchAddress,
+        token_dict: Union[EmployeeTokenObject, OccupantTokenObject],
+    ):
         import databases as database_sql_models
         from time import perf_counter
 
@@ -191,11 +224,16 @@ class AddressUpdateEventMethods(MethodToEvent):
     }
 
     @classmethod
-    def update_address(cls, address_uu_id: str, data: InsertAddress, token_dict):
+    def update_address(
+        cls,
+        address_uu_id: str,
+        data: InsertAddress,
+        token_dict: Union[EmployeeTokenObject, OccupantTokenObject],
+    ):
         address = Addresses.find_one_or_abort(uu_id=address_uu_id)
-        post_code = RelationshipEmployee2PostCode.postcode.find_one(
-            member_id=address.post_code_id
-        )
+        post_code = RelationshipEmployee2PostCode.filter_one(
+            RelationshipEmployee2PostCode.member_id==address.post_code_id
+        ).data
         if not post_code:
             raise HTTPException(
                 status_code=404,
@@ -205,7 +243,7 @@ class AddressUpdateEventMethods(MethodToEvent):
         data_dict = data.excluded_dump()
         data_dict["post_code_id"] = post_code.id
         del data_dict["post_code_uu_id"]
-
+        data_dict["is_confirmed"] = True
         updated_address = address.update(**data_dict)
         updated_address.save()
         return JSONResponse(
@@ -226,7 +264,12 @@ class AddressPatchEventMethods(MethodToEvent):
     }
 
     @classmethod
-    def patch_address(cls, address_uu_id: str, data: InsertAddress, token_dict):
+    def patch_address(
+        cls,
+        address_uu_id: str,
+        data: InsertAddress,
+        token_dict: Union[EmployeeTokenObject, OccupantTokenObject],
+    ):
         address = Addresses.find_one_or_abort(uu_id=address_uu_id)
         post_code = RelationshipEmployee2PostCode.filter_one(
             RelationshipEmployee2PostCode.member_id == address.post_code_id
@@ -260,9 +303,13 @@ class AddressPostCodeCreateEventMethods(MethodToEvent):
     }
 
     @classmethod
-    def create_post_code_address(cls, data: InsertPostCode, token_dict):
+    def create_post_code_address(
+        cls,
+        data: InsertPostCode,
+        token_dict: Union[EmployeeTokenObject, OccupantTokenObject],
+    ):
         data_dump = data.excluded_dump()
-        street = AddressStreet.find_one(uu_id=data.street_uu_id)
+        street = AddressStreet.filter_one(AddressStreet.uu_id==data.street_uu_id).data
         if not street:
             raise HTTPException(
                 status_code=404,
@@ -273,13 +320,12 @@ class AddressPostCodeCreateEventMethods(MethodToEvent):
         del data_dump["street_uu_id"], data_dump["post_code"]
 
         post_code = AddressPostcode.find_or_create(**data_dump)
-        if not post_code.is_found:
-            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,
-                is_confirmed=True,
-            )
+        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,
+            is_confirmed=True,
+        )
         AddressStreet.save()
         return JSONResponse(
             content={
@@ -300,10 +346,13 @@ class AddressPostCodeUpdateEventMethods(MethodToEvent):
 
     @classmethod
     def update_post_code_address(
-        cls, post_code_uu_id: str, data: InsertPostCode, token_dict
+        cls,
+        post_code_uu_id: str,
+        data: InsertPostCode,
+        token_dict: Union[EmployeeTokenObject, OccupantTokenObject],
     ):
-        post_code = AddressPostcode.find_one_or_abort(uu_id=post_code_uu_id)
-        street = AddressStreet.find_one(uu_id=data.street_uu_id)
+        post_code = AddressPostcode.filter_one(AddressPostcode.uu_id==post_code_uu_id).data
+        street = AddressStreet.filter_one(AddressPostcode.uu_id==data.street_uu_id).data
         if not street:
             raise HTTPException(
                 status_code=404,
@@ -330,8 +379,12 @@ class AddressPostCodeListEventMethods(MethodToEvent):
     }
 
     @classmethod
-    def list_post_code_address(cls, list_options: ListOptions, token_dict):
-        post_code_list = AddressPostcode.__many__table__.filter_active(
+    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
@@ -341,13 +394,14 @@ class AddressPostCodeListEventMethods(MethodToEvent):
                 detail="User has no post code registered or not yet any post code created.",
             )
 
-        AddressPostcode.pre_query = AddressPostcode.filter_active(
+        AddressPostcode.pre_query = AddressPostcode.filter_all(
             AddressPostcode.id.in_(
                 [post_code.member_id for post_code in post_code_list]
-            )
+            ),
+            AddressPostcode.active == True,
         ).query
         AddressPostcode.filter_attr = list_options
-        records = AddressPostcode.filter_active(
+        records = AddressPostcode.filter_all(
             *Addresses.get_smart_query(list_options.query)
         )
         return AlchemyJsonResponse(