evyos-center-server/wag-managment-api-service-version-2
2
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

events updated

Browse Source
This commit is contained in:
berkay
2024-11-13 10:38:00 +03:00
parent 952d742999
commit 077d264b28
48 changed files with 277300 additions and 913 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
api_events/events/account/account.py
View File

@@ -1,5 +0,0 @@
from api_events.events.abstract_class import MethodToEvent, ActionsSchema
from api_objects.auth.token_objects import EmployeeTokenObject, OccupantTokenObject
class ApiEvents(MethodToEvent): ...

126
api_events/events/account/account_records.py Normal file
View File

@@ -0,0 +1,126 @@
import typing
from databases import (
AccountRecords,
)
from api_validations.validations_request import (
InsertAccountRecord,
UpdateAccountRecord,
ListOptions,
)
from api_validations.core_response import AlchemyJsonResponse
from api_events.events.abstract_class import MethodToEvent, ActionsSchema
from api_objects.auth.token_objects import EmployeeTokenObject, OccupantTokenObject
class AccountRecordsListEventMethods(MethodToEvent):
event_type = "SELECT"
__event_keys__ = {
"": "account_records_list",
}
@classmethod
def account_records_list(
cls,
list_options: ListOptions,
token_dict: typing.Union[EmployeeTokenObject, OccupantTokenObject],
):
if isinstance(token_dict, OccupantTokenObject):
pass
elif isinstance(token_dict, EmployeeTokenObject):
pass
return AlchemyJsonResponse(
completed=True,
message="Update Build record",
result=None
)
class AccountRecordsCreateEventMethods(MethodToEvent):
event_type = "CREATE"
__event_keys__ = {
"": "account_records_create",
}
@classmethod
def account_records_create(
cls,
data: InsertAccountRecord,
token_dict: typing.Union[EmployeeTokenObject, OccupantTokenObject],
):
if isinstance(token_dict, OccupantTokenObject):
pass
elif isinstance(token_dict, EmployeeTokenObject):
pass
return AlchemyJsonResponse(
completed=True,
message="Update Build record",
result=None,
)
class AccountRecordsUpdateEventMethods(MethodToEvent):
event_type = "UPDATE"
__event_keys__ = {
"": "account_records_update",
}
@classmethod
def build_area_update(
cls,
build_uu_id: str,
data: UpdateAccountRecord,
token_dict: typing.Union[EmployeeTokenObject, OccupantTokenObject],
):
if isinstance(token_dict, OccupantTokenObject):
pass
elif isinstance(token_dict, EmployeeTokenObject):
pass
return AlchemyJsonResponse(
completed=False,
message="Update Build record",
result=None,
)
class AccountRecordsPatchEventMethods(MethodToEvent):
event_type = "PATCH"
__event_keys__ = {
"": "account_records_patch",
}
@classmethod
def build_area_patch(
cls,
build_uu_id: str,
data,
token_dict: typing.Union[EmployeeTokenObject, OccupantTokenObject],
):
return AlchemyJsonResponse(
completed=False,
message="Patch Build record",
result=None,
)
AccountRecordsListEventMethod = AccountRecordsListEventMethods(
action=ActionsSchema(endpoint="/account/records/list")
)
AccountRecordsCreateEventMethod = AccountRecordsCreateEventMethods(
action=ActionsSchema(endpoint="/account/records/create")
)
AccountRecordsUpdateEventMethod = AccountRecordsUpdateEventMethods(
action=ActionsSchema(endpoint="/account/records/update")
)
AccountRecordsPatchEventMethod = AccountRecordsPatchEventMethods(
action=ActionsSchema(endpoint="/account/records/patch")
)

54
api_events/events/authentication.py
View File

@@ -63,7 +63,7 @@ class AuthenticationLoginEventMethods(MethodToEvent):
def authentication_login_with_domain_and_creds(
cls,
data: Login,
request,
request: Request,
):
access_dict = Users.login_user_with_credentials(data=data, request=request)
found_user = access_dict.get("user", None)
@@ -75,7 +75,7 @@ class AuthenticationLoginEventMethods(MethodToEvent):
if not access_object:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="This User has no active role registered. Please contact your administrator."
detail="This User has no active role registered. Please contact your administrator.",
)
return JSONResponse(
@@ -103,7 +103,6 @@ class AuthenticationSelectEventMethods(MethodToEvent):
cls,
request: Request,
data,
token_dict: typing.Union[EmployeeSelection, OccupantSelection],
):
from api_objects.auth.token_objects import OccupantToken, CompanyToken
@@ -118,13 +117,13 @@ class AuthenticationSelectEventMethods(MethodToEvent):
status_code=status.HTTP_401_UNAUTHORIZED,
)
if selected_company := Companies.filter_one(
Companies.uu_id==data.company_uu_id,
*Companies.valid_record_args(Companies),
Companies.uu_id == data.company_uu_id,
*Companies.valid_record_args(Companies),
).data:
department_ids = [
department.id
for department in Departments.filter_all(
Departments.company_id==selected_company.id,
Departments.company_id == selected_company.id,
*Departments.valid_record_args(Departments),
).data
]
@@ -197,13 +196,17 @@ class AuthenticationSelectEventMethods(MethodToEvent):
status_code=status.HTTP_200_OK,
)
elif token_user.user_type == 2:
occupant_type = OccupantTypes.filter_by_one(system=True, uu_id=data.occupant_uu_id).data
occupant_type = OccupantTypes.filter_by_one(
system=True, uu_id=data.occupant_uu_id
).data
if not occupant_type:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Occupant Type is not found",
)
build_part = BuildParts.filter_by_one(system=True, uu_id=data.build_part_uu_id).data
build_part = BuildParts.filter_by_one(
system=True, uu_id=data.build_part_uu_id
).data
if not build_part:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
@@ -280,7 +283,8 @@ class AuthenticationCheckTokenEventMethods(MethodToEvent):
@classmethod
def authentication_login_with_domain_and_creds(
cls, request, token_dict: dict = None
cls,
request,
):
if get_object_via_access_key(request=request):
return JSONResponse(
@@ -301,8 +305,10 @@ class AuthenticationRefreshEventMethods(MethodToEvent):
}
@classmethod
def authentication_refresh_user_info(cls, request, token_dict: dict = None):
def authentication_refresh_user_info(
cls,
request,
):
access_token = str(request.headers.get(Auth.ACCESS_TOKEN_TAG))
if token_user := get_object_via_access_key(request=request):
if found_user := Users.filter_one(
@@ -344,7 +350,7 @@ class AuthenticationChangePasswordEventMethods(MethodToEvent):
def authentication_change_password(
cls,
data: ChangePassword,
token_dict: Union[EmployeeTokenObject, OccupantTokenObject],
token_dict: typing.Union[EmployeeSelection, OccupantSelection]
):
if token_dict.user_type == 1:
if found_user := Users.filter_one(
@@ -381,7 +387,10 @@ class AuthenticationCreatePasswordEventMethods(MethodToEvent):
}
@classmethod
def authentication_create_password(cls, data: CreatePassword):
def authentication_create_password(
cls,
data: CreatePassword,
):
if not data.re_password == data.password:
raise HTTPException(
status_code=status.HTTP_406_NOT_ACCEPTABLE, detail="Password must match"
@@ -451,7 +460,7 @@ class AuthenticationDisconnectUserEventMethods(MethodToEvent):
token_user = json.loads(redis_cli.get(key) or {})
redis_cli.delete(key)
selected_user = Users.filter_one(
Users.uu_id==token_user.get("uu_id"),
Users.uu_id == token_user.get("uu_id"),
*Users.valid_record_args(Users),
)
selected_user.remove_refresher_token(
@@ -512,7 +521,7 @@ class AuthenticationLogoutEventMethods(MethodToEvent):
if token_dict.domain == data.domain:
redis_cli.delete(token_user)
selected_user = Users.filter_one(
Users.uu_id==token_user.get("uu_id"),
Users.uu_id == token_user.get("uu_id"),
*Users.valid_record_args(Users),
)
selected_user.remove_refresher_token(domain=data.domain)
@@ -550,7 +559,10 @@ class AuthenticationRefreshTokenEventMethods(MethodToEvent):
@classmethod
def authentication_refresher_token(
cls, request: Request, data: Remember, token_dict: dict = None
cls,
request: Request,
data: Remember,
token_dict: typing.Union[EmployeeSelection, OccupantSelection],
):
token_refresher = UsersTokens.filter_by_one(
system=True, token=data.refresh_token, domain=data.domain
@@ -618,7 +630,10 @@ class AuthenticationForgotPasswordEventMethods(MethodToEvent):
@classmethod
def authentication_forgot_password(
cls, request: Request, data: Forgot, token_dict: dict = None
cls,
request: Request,
data: Forgot,
token_dict: typing.Union[EmployeeSelection, OccupantSelection],
):
found_user: Users = Users.check_user_exits(
access_key=data.access_key, domain=data.domain
@@ -677,7 +692,10 @@ class AuthenticationDownloadAvatarEventMethods(MethodToEvent):
@classmethod
def authentication_download_avatar(
cls, request: Request, data: Forgot, token_dict: dict = None
cls,
request: Request,
data: Forgot,
token_dict: typing.Union[EmployeeSelection, OccupantSelection],
):
found_user = Users.check_user_exits(
access_key=data.access_key, domain=data.domain

48
api_events/events/building/building_build_area.py
View File

@@ -13,10 +13,11 @@ from api_validations.validations_request import (
from api_validations.core_response import AlchemyJsonResponse
from api_events.events.abstract_class import MethodToEvent, ActionsSchema
from api_events.events.building.building_build import BuildPatchEventMethods
from api_objects.auth.token_objects import EmployeeTokenObject, OccupantTokenObject
class BuildListEventMethods(MethodToEvent):
class BuildAreaListEventMethods(MethodToEvent):
event_type = "SELECT"
__event_keys__ = {
@@ -25,9 +26,9 @@ class BuildListEventMethods(MethodToEvent):
@classmethod
def build_area_list(
cls,
list_options: ListOptions,
token_dict: typing.Union[EmployeeTokenObject, OccupantTokenObject],
cls,
list_options: ListOptions,
token_dict: typing.Union[EmployeeTokenObject, OccupantTokenObject],
):
if isinstance(token_dict, OccupantTokenObject):
build_ids = Build.filter_all(
@@ -54,13 +55,11 @@ class BuildListEventMethods(MethodToEvent):
*BuildArea.valid_record_args(BuildArea),
)
return AlchemyJsonResponse(
completed=True,
message="Update Build record",
result=records
completed=True, message="Update Build record", result=records
)
class BuildCreateEventMethods(MethodToEvent):
class BuildAreaCreateEventMethods(MethodToEvent):
event_type = "CREATE"
__event_keys__ = {
@@ -69,9 +68,9 @@ class BuildCreateEventMethods(MethodToEvent):
@classmethod
def build_area_create(
cls,
data: InsertBuildArea,
token_dict: typing.Union[EmployeeTokenObject, OccupantTokenObject],
cls,
data: InsertBuildArea,
token_dict: typing.Union[EmployeeTokenObject, OccupantTokenObject],
):
if isinstance(token_dict, OccupantTokenObject):
if not token_dict.selected_occupant.build_uuid == data.build_uu_id:
@@ -107,7 +106,8 @@ class BuildCreateEventMethods(MethodToEvent):
result=created_build_part,
)
class BuildUpdateEventMethods(MethodToEvent):
class BuildAreaUpdateEventMethods(MethodToEvent):
event_type = "UPDATE"
__event_keys__ = {
@@ -115,7 +115,12 @@ class BuildUpdateEventMethods(MethodToEvent):
}
@classmethod
def build_area_update(cls, build_uu_id: str, data: UpdateBuildArea, token_dict):
def build_area_update(
cls,
build_uu_id: str,
data: UpdateBuildArea,
token_dict: typing.Union[EmployeeTokenObject, OccupantTokenObject],
):
return AlchemyJsonResponse(
completed=False,
message="Update Build record",
@@ -123,7 +128,7 @@ class BuildUpdateEventMethods(MethodToEvent):
)
class BuildPatchEventMethods(MethodToEvent):
class BuildAreaPatchEventMethods(MethodToEvent):
event_type = "PATCH"
__event_keys__ = {
@@ -131,21 +136,26 @@ class BuildPatchEventMethods(MethodToEvent):
}
@classmethod
def build_area_patch(cls, build_uu_id: str, data, token_dict):
def build_area_patch(
cls,
build_uu_id: str,
data,
token_dict: typing.Union[EmployeeTokenObject, OccupantTokenObject],
):
return AlchemyJsonResponse(
completed=False,
message="Update Build record",
message="Patch Build record",
result=None,
)
BuildListEventMethod = BuildListEventMethods(
BuildAreaListEventMethod = BuildAreaListEventMethods(
action=ActionsSchema(endpoint="/building/area/list")
)
BuildCreateEventMethod = BuildCreateEventMethods(
BuildAreaCreateEventMethod = BuildAreaCreateEventMethods(
action=ActionsSchema(endpoint="/building/area/create")
)
BuildUpdateEventMethod = BuildUpdateEventMethods(
BuildUpdateEventMethod = BuildAreaUpdateEventMethods(
action=ActionsSchema(endpoint="/building/area/update")
)
BuildPatchEventMethod = BuildPatchEventMethods(

4
api_events/events/building/building_build_parts.py
View File

@@ -34,10 +34,10 @@ class BuildingBuildPartsListEventMethods(MethodToEvent):
employee_id=token_dict.selected_company.employee_id,
)
build_list_ids = [build.id for build in build_list_query.all()]
BuildParts.pre_query = BuildParts.query.filter(
BuildParts.pre_query = BuildParts.filter_all(
BuildParts.build_id.in_(build_list_ids),
*BuildParts.valid_record_args(BuildParts),
)
).query
records = BuildParts.filter_active(
*BuildParts.get_smart_query(list_options.query),
*BuildParts.valid_record_args(BuildParts),

159
api_events/events/building/building_build_sites.py
View File

@@ -1,5 +1,162 @@
import typing
from databases import (
Build,
BuildSites,
)
from api_validations.validations_request import (
InsertBuildArea,
UpdateBuildArea,
ListOptions,
)
from api_validations.core_response import AlchemyJsonResponse
from api_events.events.abstract_class import MethodToEvent, ActionsSchema
from api_objects.auth.token_objects import EmployeeTokenObject, OccupantTokenObject
class BuildingBuildSitesEvents(MethodToEvent): ...
class BuildSitesListEventMethods(MethodToEvent):
event_type = "SELECT"
__event_keys__ = {
"": "build_sites_list",
}
@classmethod
def build_sites_list(
cls,
list_options: ListOptions,
token_dict: typing.Union[EmployeeTokenObject, OccupantTokenObject],
):
if isinstance(token_dict, OccupantTokenObject):
build_ids = Build.filter_all(
Build.id == token_dict.selected_occupant.build_id,
*Build.valid_record_args(Build),
).data
build_id_list = [build.id for build in build_ids]
BuildSites.pre_query = BuildSites.filter_all(
BuildSites.build_id.in_(build_id_list),
*BuildSites.valid_record_args(BuildSites),
).query
elif isinstance(token_dict, EmployeeTokenObject):
build_ids = Build.select_action(
employee_id=token_dict.selected_company.employee_id
)
build_id_list = [build.id for build in build_ids]
BuildSites.pre_query = BuildSites.filter_all(
BuildSites.build_id.in_(build_id_list),
*BuildSites.valid_record_args(BuildSites),
).query
BuildSites.filter_attr = list_options
records = BuildSites.filter_all(
*BuildSites.get_smart_query(smart_query=list_options.query),
*BuildSites.valid_record_args(BuildSites),
)
return AlchemyJsonResponse(
completed=True, message="Update Build record", result=records
)
class BuildSitesCreateEventMethods(MethodToEvent):
event_type = "CREATE"
__event_keys__ = {
"": "build_sites_create",
}
@classmethod
def build_area_create(
cls,
data: InsertBuildArea,
token_dict: typing.Union[EmployeeTokenObject, OccupantTokenObject],
):
if isinstance(token_dict, OccupantTokenObject):
if not token_dict.selected_occupant.build_uuid == data.build_uu_id:
BuildSites.raise_http_exception(
status_code="HTTP_403_FORBIDDEN",
error_case="UNAUTHORIZED",
message=f"Occupant can not create build sites for {data.build_uu_id}",
data={
"build_uu_id": data.build_uu_id,
},
)
elif isinstance(token_dict, EmployeeTokenObject):
build_ids = Build.select_action(
employee_id=token_dict.selected_company.employee_id
).all()
if not str(data.build_uu_id) in [str(build.uu_id) for build in build_ids]:
BuildSites.raise_http_exception(
status_code="HTTP_403_FORBIDDEN",
error_case="UNAUTHORIZED",
message=f"Employee can not create build sites for {data.build_uu_id}",
data={
"build_uu_id": data.build_uu_id,
},
)
data_dict = data.excluded_dump()
created_build_part = BuildSites.find_or_create(**data_dict)
created_build_part.save()
created_build_part.update(is_confirmed=True)
created_build_part.save()
return AlchemyJsonResponse(
completed=True,
message="Update Build record",
result=created_build_part,
)
class BuildSitesUpdateEventMethods(MethodToEvent):
event_type = "UPDATE"
__event_keys__ = {
"": "build_sites_update",
}
@classmethod
def build_area_update(
cls,
build_uu_id: str,
data: UpdateBuildArea,
token_dict: typing.Union[EmployeeTokenObject, OccupantTokenObject],
):
return AlchemyJsonResponse(
completed=False,
message="Update Build record",
result=None,
)
class BuildSitesPatchEventMethods(MethodToEvent):
event_type = "PATCH"
__event_keys__ = {
"": "build_sites_patch",
}
@classmethod
def build_area_patch(
cls,
build_uu_id: str,
data,
token_dict: typing.Union[EmployeeTokenObject, OccupantTokenObject],
):
return AlchemyJsonResponse(
completed=False,
message="Patch Build record",
result=None,
)
BuildSitesListEventMethod = BuildSitesListEventMethods(
action=ActionsSchema(endpoint="/building/sites/list")
)
BuildSitesCreateEventMethod = BuildSitesCreateEventMethods(
action=ActionsSchema(endpoint="/building/sites/create")
)
BuildSitesUpdateEventMethod = BuildSitesUpdateEventMethods(
action=ActionsSchema(endpoint="/building/sites/update")
)
BuildSitesPatchEventMethod = BuildSitesPatchEventMethods(
action=ActionsSchema(endpoint="/building/sites/patch")
)

289
api_events/events/building/building_living_spaces.py
View File

@@ -1,9 +1,8 @@
import typing
from fastapi import status, HTTPException
from typing import Union
from api_events.events.events.events_bind_modules import ModulesBindOccupantEventMethods
from databases import (
Modules,
BuildParts,
Build,
BuildLivingSpace,
@@ -12,70 +11,97 @@ from databases import (
)
from api_events.events.abstract_class import MethodToEvent, ActionsSchema
from api_objects.auth.token_objects import EmployeeTokenObject, OccupantTokenObject
from api_validations.core_response import AlchemyJsonResponse
from api_validations.validations_request import (
InsertBuildLivingSpace,
UpdateBuildLivingSpace,
ListOptions,
)
from databases.sql_models.event.event import Modules
class BuildingLivingSpacesPartsListEventMethods(MethodToEvent):
event_type = "SELECT"
__event_keys__ = {
"8fd04d94-68fb-4a07-9549-8b47aee3a870": "building_build_parts_list",
"2f3041a9-6184-44c2-ac38-8ea934297ed1": "building_build_parts_list_with_expired",
"36961d8a-cefa-46cc-9f7c-9d841d6351b6": "building_live_space_list",
}
@classmethod
def building_build_parts_list(cls, list_options: ListOptions, token_dict):
build_id_list_query = Build.select_action(
employee_id=token_dict.selected_company.employee_id
)
Build.filter_attr = list_options
build_part_id_list_query = BuildParts.filter_all(
BuildParts.build_id.in_([build.id for build in build_id_list_query.all()]),
*BuildParts.valid_record_args(BuildParts),
).data
list_options.query.pop("expiry_starts", None)
list_options.query.pop("expiry_ends", None)
records = BuildLivingSpace.filter_all(
BuildLivingSpace.build_parts_id.in_(
[build_part.id for build_part in build_part_id_list_query]
),
*BuildLivingSpace.get_smart_query(smart_query=list_options.query),
*BuildLivingSpace.valid_record_args(BuildLivingSpace),
).data
return AlchemyJsonResponse(
completed=True,
message="Building Living Spaces are listed successfully",
result=records,
)
@classmethod
def building_build_parts_list_with_expired(
cls, list_options: ListOptions, token_dict
def building_build_parts_list(
cls,
list_options: ListOptions,
token_dict: Union[EmployeeTokenObject, OccupantTokenObject],
):
build_id_list_query = Build.select_action(
employee_id=token_dict.selected_company.employee_id
)
Build.filter_attr = list_options
build_part_id_list_query = BuildParts.filter_all(
BuildParts.build_id.in_([build.id for build in build_id_list_query.all()]),
*BuildLivingSpace.valid_record_args(BuildLivingSpace),
).data
records = BuildLivingSpace.filter_all(
BuildLivingSpace.build_parts_id.in_(
[build_part.id for build_part in build_part_id_list_query]
),
*BuildLivingSpace.get_smart_query(smart_query=list_options.query),
*BuildLivingSpace.valid_record_args(BuildLivingSpace),
).data
records, Build.filter_attr = [], None
if isinstance(token_dict, OccupantTokenObject):
occupants_build_id = Build.filter_one(
Build.id == token_dict.selected_occupant.build_id,
*Build.valid_record_args(Build),
).data
if not occupants_build_id:
Build.raise_http_exception(
status_code="HTTP_403_FORBIDDEN",
error_case="UNAUTHORIZED",
message=f"Occupant has no build registered in the system. Contact with your company.",
data={},
)
occupants_build_parts = BuildParts.filter_all(
BuildParts.build_id.in_(occupants_build_id.id),
*BuildParts.valid_record_args(BuildParts),
).data
if not occupants_build_parts:
Build.raise_http_exception(
status_code="HTTP_403_FORBIDDEN",
error_case="UNAUTHORIZED",
message=f"Occupant has no build parts registered in the system. Contact with your company.",
data={},
)
BuildLivingSpace.pre_query = BuildLivingSpace.filter_all(
BuildLivingSpace.build_parts_id.in_(
[build_part.id for build_part in occupants_build_parts]
),
*BuildLivingSpace.valid_record_args(BuildLivingSpace),
).query
BuildLivingSpace.filter_attr = list_options
records = BuildLivingSpace.filter_all(
*BuildLivingSpace.get_smart_query(smart_query=list_options.query),
*BuildLivingSpace.valid_record_args(BuildLivingSpace),
)
elif isinstance(token_dict, EmployeeTokenObject):
build_id_list_query = Build.select_action(
employee_id=token_dict.selected_company.employee_id
)
if not build_id_list_query:
Build.raise_http_exception(
status_code="HTTP_403_FORBIDDEN",
error_case="UNAUTHORIZED",
message=f"Employee has no build registered in the system. Contact with your supervisor.",
data={},
)
build_part_id_list_query = BuildParts.filter_all(
BuildParts.build_id.in_(
[build.id for build in build_id_list_query.all()]
),
*BuildLivingSpace.valid_record_args(BuildLivingSpace),
).data
if not build_part_id_list_query:
Build.raise_http_exception(
status_code="HTTP_403_FORBIDDEN",
error_case="UNAUTHORIZED",
message=f"Employee has no build parts registered in the system. Contact with your supervisor.",
data={},
)
BuildLivingSpace.pre_query = BuildLivingSpace.filter_all(
BuildLivingSpace.build_parts_id.in_(
[build_part.id for build_part in build_part_id_list_query]
),
*BuildLivingSpace.valid_record_args(BuildLivingSpace),
).query
Build.filter_attr = list_options
records = BuildLivingSpace.filter_all(
*BuildLivingSpace.get_smart_query(smart_query=list_options.query),
*BuildLivingSpace.valid_record_args(BuildLivingSpace),
)
return AlchemyJsonResponse(
completed=True,
message="Building Living Spaces are listed successfully",
@@ -86,15 +112,13 @@ class BuildingLivingSpacesPartsListEventMethods(MethodToEvent):
class BuildingLivingSpacesPartsCreateEventMethods(MethodToEvent):
event_type = "CREATE"
__event_keys__ = {
"b78ca45c-b9f4-41f6-9ddb-2c6f2faa2570": "building_live_space_create"
}
__event_keys__ = {"46d90119-3b23-4784-8053-fe11da4a3584": "building_live_space_create"}
@classmethod
def building_live_space_create(
cls,
data: InsertBuildLivingSpace,
token_dict: typing.Union[EmployeeTokenObject, OccupantTokenObject],
token_dict: Union[EmployeeTokenObject, OccupantTokenObject],
):
from api_library.date_time_actions.date_functions import system_arrow
@@ -108,28 +132,36 @@ class BuildingLivingSpacesPartsCreateEventMethods(MethodToEvent):
*BuildLivingSpace.valid_record_args(BuildLivingSpace),
).data
if not build_part:
raise HTTPException(
status_code=status.HTTP_418_IM_A_TEAPOT,
detail=f"{data.build_parts_uu_id} - Build Part is not found in database. Check build part uu_id",
BuildLivingSpace.raise_http_exception(
status_code="HTTP_404_NOT_FOUND",
error_case="UNAUTHORIZED",
message=f"{data.build_parts_uu_id} - Build Part is not found in database. Check build part uu_id",
data={
"life_person_uu_id": data.life_person_uu_id,
},
)
life_person = People.filter_one(
People.uu_id == data.person_uu_id or "",
*People.valid_record_args(People),
).data
if not life_person:
raise HTTPException(
status_code=status.HTTP_418_IM_A_TEAPOT,
detail=f"{data.life_person_uu_id} - Living Person is not found in database. "
f"Check build live person uu_id",
BuildLivingSpace.raise_http_exception(
status_code="HTTP_404_NOT_FOUND",
error_case="UNAUTHORIZED",
message=f"{data.life_person_uu_id} - Living Person is not found in database.",
data={
"life_person_uu_id": data.life_person_uu_id,
},
)
occupant_type = OccupantTypes.filter_by_one(uu_id=data.occupant_type_uu_id).data
if not occupant_type:
raise HTTPException(
status_code=status.HTTP_418_IM_A_TEAPOT,
detail=f"{data.occupant_type_uu_id} - Occupant Type is not found in database. "
f"Check occupant type uu_id",
BuildLivingSpace.raise_http_exception(
status_code="HTTP_404_NOT_FOUND",
error_case="UNAUTHORIZED",
message=f"{data.occupant_type_uu_id} - Occupant Type is not found in database. Check occupant type uu_id",
data={
"occupant_type_uu_id": data.occupant_type_uu_id,
},
)
data_dict["occupant_type"] = occupant_type.id
@@ -144,8 +176,6 @@ class BuildingLivingSpacesPartsCreateEventMethods(MethodToEvent):
BuildLivingSpace.person_id == life_person.id,
BuildLivingSpace.occupant_type == occupant_type.id,
*BuildLivingSpace.valid_record_args(BuildLivingSpace),
# str(system_arrow.now()) < BuildLivingSpace.expiry_ends,
# str(system_arrow.now()) >= BuildLivingSpace.expiry_starts,
select_args=[BuildLivingSpace.id],
order_by=BuildLivingSpace.expiry_starts.desc(),
limit=1,
@@ -181,69 +211,82 @@ class BuildingLivingSpacesPartsUpdateEventMethods(MethodToEvent):
event_type = "UPDATE"
__event_keys__ = {
"70b4666f-4ceb-46ec-b89e-24be8712f0e7": "building_live_space_update",
"c786e15c-c03e-4e8f-936c-7e5e5ec9bbcc": "building_live_space_update",
}
@classmethod
def building_live_space_update(cls, build_uu_id: str, data, token_dict):
def building_live_space_update(
cls,
build_uu_id: str,
data: UpdateBuildLivingSpace,
token_dict: Union[EmployeeTokenObject, OccupantTokenObject],
):
from api_library.date_time_actions.date_functions import system_arrow
data_dict = data.dump()
build_id_list_query = Build.select_action(
employee_id=token_dict.selected_company.employee_id
)
build_part = BuildParts.filter_one(
BuildParts.uu_id == data.build_parts_uu_id,
BuildParts.build_id.in_([build.id for build in build_id_list_query.all()]),
*BuildParts.valid_record_args(BuildParts),
).data
if not build_part:
raise HTTPException(
status_code=status.HTTP_418_IM_A_TEAPOT,
detail=f"{data.build_parts_uu_id} - Build Part is not found in database. Check build part uu_id",
if isinstance(token_dict, OccupantTokenObject):
data_dict = data.dump()
build_id_list_query = Build.select_action(
employee_id=token_dict.selected_company.employee_id
)
build_part = BuildParts.filter_one(
BuildParts.uu_id == data.build_parts_uu_id,
BuildParts.build_id.in_(
[build.id for build in build_id_list_query.all()]
),
*BuildParts.valid_record_args(BuildParts),
).data
if not build_part:
BuildLivingSpace.raise_http_exception(
status_code="HTTP_403_FORBIDDEN",
error_case="UNAUTHORIZED",
message=f"{data.life_person_uu_id} - Living Person is not found in database.",
data={
"person_uu_id": data.life_person_uu_id,
},
)
life_person = People.filter_one(
People.uu_id == data.life_person_uu_id or ""
).data
if not life_person:
BuildLivingSpace.raise_http_exception(
status_code="HTTP_403_FORBIDDEN",
error_case="UNAUTHORIZED",
message=f"{data.life_person_uu_id} - Living Person is not found in database.",
data={
"person_uu_id": data.life_person_uu_id,
},
)
life_person = People.filter_one(
People.uu_id == data.life_person_uu_id or ""
).data
if not life_person:
raise HTTPException(
status_code=status.HTTP_418_IM_A_TEAPOT,
detail=f"{data.life_person_uu_id} - Living Person is not found in database. "
f"Check build live person uu_id",
)
living_space_id = BuildLivingSpace.select_only(
*BuildLivingSpace.valid_record_args(BuildLivingSpace),
select_args=[BuildLivingSpace.id],
order_by=BuildLivingSpace.expiry_starts.desc(),
limit=1,
).get(1)
living_space_id = BuildLivingSpace.select_only(
*BuildLivingSpace.valid_record_args(BuildLivingSpace),
select_args=[BuildLivingSpace.id],
order_by=BuildLivingSpace.expiry_starts.desc(),
limit=1,
).get(1)
last_living_space = BuildLivingSpace.filter_one(
BuildLivingSpace.id == living_space_id if living_space_id else None,
*BuildLivingSpace.valid_record_args(BuildLivingSpace),
).data
last_living_space = BuildLivingSpace.filter_one(
BuildLivingSpace.id == living_space_id if living_space_id else None,
*BuildLivingSpace.valid_record_args(BuildLivingSpace),
).data
data_dict["expiry_starts"] = str(system_arrow.now())
data_dict["is_tenant_live"] = bool(data.is_tenant_live)
data_dict["build_parts_id"] = build_part.id
if data_dict["is_tenant_live"]:
owner_person = getattr(last_living_space, "owner_person_id", None)
if not owner_person:
raise HTTPException(
status_code=status.HTTP_418_IM_A_TEAPOT,
detail=dict(
data_dict["expiry_starts"] = str(system_arrow.now())
data_dict["is_tenant_live"] = bool(data.is_tenant_live)
data_dict["build_parts_id"] = build_part.id
if data_dict["is_tenant_live"]:
owner_person = getattr(last_living_space, "owner_person_id", None)
if not owner_person:
BuildLivingSpace.raise_http_exception(
status_code="HTTP_403_FORBIDDEN",
error_case="UNAUTHORIZED",
message="Owner person of build part is not defined. Please register owner of part first.",
data=build_part.get_dict(),
),
)
data_dict["life_person_id"] = life_person.id
data_dict["owner_person_id"] = owner_person
else:
data_dict["life_person_id"] = life_person.id
data_dict["owner_person_id"] = life_person.id
del data_dict["build_parts_uu_id"], data_dict["life_person_uu_id"]
)
data_dict["life_person_id"] = life_person.id
data_dict["owner_person_id"] = owner_person
else:
data_dict["life_person_id"] = life_person.id
data_dict["owner_person_id"] = life_person.id
del data_dict["build_parts_uu_id"], data_dict["life_person_uu_id"]
BuildingLivingSpacesPartsListEventMethod = BuildingLivingSpacesPartsListEventMethods(

58
api_events/events/company/company_company.py
View File

@@ -61,18 +61,18 @@ class CompanyCreateEventMethods(MethodToEvent):
@classmethod
def company_create(
cls, data: InsertCompany, token_dict: typing.Union[EmployeeTokenObject, OccupantTokenObject]
cls,
data: InsertCompany,
token_dict: EmployeeTokenObject,
):
created_company = Companies.create_action(
data=data, token=token_dict
)
created_company = Companies.create_action(data=data, token=token_dict)
created_company.related_company = token_dict.selected_company.company_uu_id
created_company.save()
return JSONResponse(
content={
"completed": True,
"message": "Create Company record",
"data": created_company.get_dict()
"data": created_company.get_dict(),
},
status_code=status.HTTP_200_OK,
)
@@ -86,26 +86,39 @@ class CompanyUpdateEventMethods(MethodToEvent):
}
@classmethod
def company_update(cls, company_uu_id: str, data: UpdateCompany, token_dict):
find_one_company = Companies.filter_one(Companies.uu_id == company_uu_id)
access_authorized_company = Companies.select_action(
duty_id=getattr(token_dict, "duty_id", 5),
filter_expr=[Companies.id == token_dict.get("")],
def company_update(
cls,
company_uu_id: str,
data: UpdateCompany, token_dict: EmployeeTokenObject
):
Companies.pre_query = Companies.select_action(
duty_id_list=[
token_dict.selected_company.bulk_duties_id,
token_dict.selected_company.duty_id,
],
)
if access_authorized_company.count:
updated_company = find_one_company.data.update(**data.excluded_dump())
Companies.save()
find_one_company = Companies.filter_one(
Companies.uu_id == company_uu_id,
Companies.valid_record_args(Companies),
).data
if not find_one_company:
return JSONResponse(
content={
"completed": True,
"message": "Update Company record",
"data": updated_company,
"data": {}
},
status_code=status.HTTP_200_OK,
status_code=200,
)
updated_company = find_one_company.update(**data.excluded_dump())
Companies.save()
return JSONResponse(
content={"completed": True, "message": "Update Company record", "data": {}},
status_code=status.HTTP_200_OK,
content={
"completed": True,
"message": "Update Company record",
"data": updated_company,
},
status_code=200,
)
@@ -117,11 +130,15 @@ class CompanyPatchEventMethods(MethodToEvent):
}
@classmethod
def company_patch(cls, company_uu_id: str, data: PatchRecord, token_dict):
def company_patch(
cls, company_uu_id: str, data: PatchRecord, token_dict: EmployeeTokenObject
):
find_one_company = Companies.find_one_or_abort(uu_id=company_uu_id)
access_authorized_company = Companies.select_action(
duty_id=getattr(token_dict, "duty_id", 5),
filter_expr=[Companies.id == find_one_company.id],
duty_id_list=[
token_dict.selected_company.bulk_duties_id,
token_dict.selected_company.duty_id,
],
)
if access_authorized_company.count:
action = data.excluded_dump()
@@ -152,7 +169,6 @@ class CompanyPatchEventMethods(MethodToEvent):
status_code=status.HTTP_200_OK,
)
CompanyPatchEventMethod = CompanyListEventMethods(
action=ActionsSchema(endpoint="/company/list")
)

16
api_events/events/company/company_department.py
View File

@@ -1,4 +1,4 @@
from typing import Optional, Union
from typing import Optional
from fastapi import status
from fastapi.responses import JSONResponse
@@ -12,7 +12,7 @@ from api_validations.validations_request import (
from databases import Departments
from api_events.events.abstract_class import MethodToEvent, ActionsSchema
from api_objects.auth.token_objects import EmployeeTokenObject, OccupantTokenObject
from api_objects.auth.token_objects import EmployeeTokenObject
from api_validations.core_response import AlchemyJsonResponse
@@ -27,7 +27,7 @@ class DepartmentListEventMethods(MethodToEvent):
def department_list(
cls,
list_options: ListOptions,
token_dict: Union[EmployeeTokenObject, OccupantTokenObject],
token_dict: EmployeeTokenObject,
):
Departments.filter_attr = list_options
records = Departments.filter_active(
@@ -58,7 +58,9 @@ class DepartmentCreateEventMethods(MethodToEvent):
}
@classmethod
def super_user_department_create(cls, data: DepartmentsPydantic, token_dict):
def super_user_department_create(
cls, data: DepartmentsPydantic, token_dict: EmployeeTokenObject,
):
data_dict = data.excluded_dump()
data_dict["company_id"] = token_dict.selected_company.company_id
data_dict["company_uu_id"] = token_dict.selected_company.company_uu_id
@@ -83,7 +85,7 @@ class DepartmentUpdateEventMethods(MethodToEvent):
@classmethod
def department_update(
cls, company_uu_id: str, data: DepartmentsPydantic, token_dict
cls, company_uu_id: str, data: DepartmentsPydantic, token_dict: EmployeeTokenObject,
):
find_one_company = Departments.filter_one(Departments.uu_id == company_uu_id)
access_authorized_company = Departments.select_action(
@@ -116,7 +118,9 @@ class DepartmentPatchEventMethods(MethodToEvent):
}
@classmethod
def department_patch(cls, company_uu_id: str, data: PatchRecord, token_dict):
def department_patch(
cls, company_uu_id: str, data: PatchRecord, token_dict: EmployeeTokenObject
):
find_one_company = Departments.find_one_or_abort(uu_id=company_uu_id)
access_authorized_company = Departments.select_action(
duty_id=getattr(token_dict, "duty_id", 5),

21
api_events/events/company/company_duties.py
View File

@@ -1,5 +1,3 @@
from typing import Union
from fastapi import status
from fastapi.responses import JSONResponse
@@ -14,7 +12,7 @@ from api_validations.validations_request import (
from databases import Departments, Duty, Duties
from api_events.events.abstract_class import MethodToEvent, ActionsSchema
from api_objects.auth.token_objects import EmployeeTokenObject, OccupantTokenObject
from api_objects.auth.token_objects import EmployeeTokenObject
class DutiesListEventMethods(MethodToEvent):
@@ -28,7 +26,7 @@ class DutiesListEventMethods(MethodToEvent):
def duties_list(
cls,
list_options: ListOptions,
token_dict: Union[EmployeeTokenObject, OccupantTokenObject],
token_dict: EmployeeTokenObject,
):
Duties.filter_attr = list_options
records = Duties.filter_all(
@@ -53,7 +51,7 @@ class DutiesGetByUUIDEventMethods(MethodToEvent):
def duties_get_by_uuid(
cls,
data: SelectDuties,
token_dict: Union[EmployeeTokenObject, OccupantTokenObject],
token_dict: EmployeeTokenObject,
):
duty = Duty.filter_one(Duty.uu_id == data.duty_uu_id).data
@@ -98,7 +96,7 @@ class DutiesCreateEventMethods(MethodToEvent):
def duties_create(
cls,
data: InsertDuties,
token_dict: Union[EmployeeTokenObject, OccupantTokenObject],
token_dict: EmployeeTokenObject,
):
duty = Duty.filter_one(Duty.uu_id == data.duties_uu_id).data
department = Departments.filter_one(Duty.uu_id == data.department_uu_id).data
@@ -144,7 +142,12 @@ class DutiesUpdateEventMethods(MethodToEvent):
}
@classmethod
def duties_update(cls, duties_uu_id: str, data: UpdateDuties, token_dict):
def duties_update(
cls,
duties_uu_id: str,
data: UpdateDuties,
token_dict: EmployeeTokenObject,
):
find_one_duties = Duties.find_one_or_abort(uu_id=duties_uu_id)
access_authorized_duties = Duties.select_action(
duty_id=getattr(token_dict, "duty_id", 5),
@@ -174,7 +177,9 @@ class DutiesPatchEventMethods(MethodToEvent):
}
@classmethod
def duties_patch(cls, duties_uu_id: str, data: PatchRecord, token_dict):
def duties_patch(
cls, duties_uu_id: str, data: PatchRecord, token_dict: EmployeeTokenObject,
):
find_one_duties = Duties.find_one_or_abort(uu_id=duties_uu_id)
access_authorized_duties = Duties.select_action(
duty_id=getattr(token_dict, "duty_id", 5),

12
api_events/events/company/company_duty.py
View File

@@ -1,5 +1,3 @@
from typing import Union
from fastapi import status
from fastapi.responses import JSONResponse
@@ -27,7 +25,7 @@ class DutyListEventMethods(MethodToEvent):
def duty_list(
cls,
list_options: ListOptions,
token_dict: Union[EmployeeTokenObject, OccupantTokenObject],
token_dict: EmployeeTokenObject,
):
records = Duty.filter_active(
*Duty.get_smart_query(list_options.query),
@@ -50,7 +48,7 @@ class DutyCreateEventMethods(MethodToEvent):
def duty_create(
cls,
data: InsertCompanyDuty,
token_dict: Union[EmployeeTokenObject, OccupantTokenObject],
token_dict: EmployeeTokenObject,
):
created_duty = Duty.find_or_create(**data.excluded_dump())
Duty.save()
@@ -76,7 +74,7 @@ class DutyUpdateEventMethods(MethodToEvent):
cls,
company_uu_id: str,
data,
token_dict: Union[EmployeeTokenObject, OccupantTokenObject],
token_dict: EmployeeTokenObject,
):
find_one_company = Duty.find_one_or_abort(uu_id=company_uu_id)
access_authorized_company = Duty.select_action(
@@ -109,7 +107,9 @@ class DutyPatchEventMethods(MethodToEvent):
}
@classmethod
def duty_patch(cls, company_uu_id: str, data: PatchRecord, token_dict):
def duty_patch(
cls, company_uu_id: str, data: PatchRecord, token_dict: EmployeeTokenObject,
):
find_one_company = Duty.find_one_or_abort(uu_id=company_uu_id)
access_authorized_company = Duty.select_action(
duty_id=getattr(token_dict, "duty_id", 5),

21
api_events/events/company/company_employee.py
View File

@@ -1,6 +1,3 @@
from datetime import datetime
from typing import Union
from fastapi import status, HTTPException
from fastapi.responses import JSONResponse
@@ -30,7 +27,7 @@ class EmployeeListEventMethods(MethodToEvent):
def employee_list(
cls,
list_options: ListOptions,
token_dict: Union[EmployeeTokenObject, OccupantTokenObject],
token_dict: EmployeeTokenObject,
):
Employees.filter_attr = list_options
records = Employees.filter_active(
@@ -55,7 +52,7 @@ class EmployeeCreateEventMethods(MethodToEvent):
def employee_create(
cls,
data: InsertEmployees,
token_dict: Union[EmployeeTokenObject, OccupantTokenObject],
token_dict: EmployeeTokenObject,
):
person = People.filter_one(
People.uu_id == data.people_uu_id,
@@ -99,7 +96,9 @@ class EmployeeUpdateEventMethods(MethodToEvent):
}
@classmethod
def employee_update(cls, employee_uu_id: str, data: PatchRecord, token_dict):
def employee_update(
cls, employee_uu_id: str, data: PatchRecord, token_dict: EmployeeTokenObject,
):
find_one_employee = Employees.filter_one(
Employees.uu_id == employee_uu_id,
*Employees.valid_record_args(Employees),
@@ -142,7 +141,7 @@ class EmployeePatchEventMethods(MethodToEvent):
cls,
employee_uu_id: str,
data: PatchRecord,
token_dict: Union[EmployeeTokenObject, OccupantTokenObject],
token_dict: EmployeeTokenObject,
):
find_one_employee = Employees.find_one_or_abort(uu_id=employee_uu_id)
access_authorized_employee = Employees.select_action(
@@ -186,7 +185,9 @@ class Employee2PeopleEmployEventMethods(MethodToEvent):
}
@classmethod
def company_employee_employ(cls, data: BindEmployees2People, token_dict):
def company_employee_employ(
cls, data: BindEmployees2People, token_dict: EmployeeTokenObject,
):
selected_staff = Staff.filter_one(
Staff.uu_id == data.staff_uu_id,
*Staff.valid_record_args(Staff),
@@ -248,7 +249,9 @@ class Employee2PeopleFireEventMethods(MethodToEvent):
}
@classmethod
def company_employee_fire(cls, data: BindEmployees2People, token_dict):
def company_employee_fire(
cls, data: BindEmployees2People, token_dict: EmployeeTokenObject,
):
selected_people = People.filter_one(
People.uu_id == data.people_uu_id,
*People.valid_record_args(People),

22
api_events/events/company/company_staff.py
View File

@@ -12,7 +12,7 @@ from api_validations.validations_request import (
from databases import Staff, Duties
from api_events.events.abstract_class import MethodToEvent, ActionsSchema
from api_objects.auth.token_objects import EmployeeTokenObject, OccupantTokenObject
from api_objects.auth.token_objects import EmployeeTokenObject
from api_validations.core_response import AlchemyJsonResponse
@@ -23,7 +23,9 @@ class StaffListEventMethods(MethodToEvent):
}
@classmethod
def staff_list(cls, list_options: ListOptions, token_dict):
def staff_list(
cls, list_options: ListOptions, token_dict: EmployeeTokenObject
):
Staff.filter_attr = list_options
records = Staff.filter_active(
*Staff.get_smart_query(smart_query=list_options.query),
@@ -42,7 +44,9 @@ class StaffCreateEventMethods(MethodToEvent):
}
@classmethod
def staff_create(cls, data: InsertStaff, token_dict: EmployeeTokenObject):
def staff_create(
cls, data: InsertStaff, token_dict: EmployeeTokenObject
):
data_dict = data.excluded_dump()
duties = Duties.filter_one(
Duties.uu_id == data.duties_uu_id,
@@ -74,7 +78,9 @@ class StaffGetByUUIDEventMethods(MethodToEvent):
}
@classmethod
def staff_get_by_uu_id(cls, data: SelectStaff, token_dict):
def staff_get_by_uu_id(
cls, data: SelectStaff, token_dict: EmployeeTokenObject
):
if data.duties_uu_id:
duties_id = Duties.filter_one(
Duties.uu_id == data.duties_uu_id, *Duties.valid_record_args(Duties)
@@ -111,7 +117,9 @@ class StaffUpdateEventMethods(MethodToEvent):
}
@classmethod
def staff_update(cls, staff_uu_id: str, data, token_dict):
def staff_update(
cls, staff_uu_id: str, data, token_dict: EmployeeTokenObject
):
return JSONResponse(
content={"completed": True, "message": "Update Staff record", "data": {}},
status_code=status.HTTP_200_OK,
@@ -126,7 +134,9 @@ class StaffPatchEventMethods(MethodToEvent):
}
@classmethod
def staff_patch(cls, staff_uu_id: str, data: PatchRecord, token_dict):
def staff_patch(
cls, staff_uu_id: str, data: PatchRecord, token_dict: EmployeeTokenObject
):
return JSONResponse(
content={
"completed": False,

117
api_events/events/identity/people.py
View File

@@ -4,13 +4,13 @@ from fastapi import status
from fastapi.responses import JSONResponse
from databases import (
Build,
People,
Users,
Companies,
)
from api_validations.validations_request import InsertPerson, UpdateUsers
from api_events.events.abstract_class import MethodToEvent, ActionsSchema
from api_objects.auth.token_objects import EmployeeTokenObject, OccupantTokenObject
from api_validations.core_response import AlchemyJsonResponse
@@ -22,18 +22,40 @@ class PeopleListEventMethods(MethodToEvent):
__event_keys__ = {
"0a05f03c-6ed8-4230-a4ff-6e7cf886909b": "super_users_people_list",
"b5612538-0445-4a4a-ab13-d2a06037f7a5": "sales_users_people_list",
"c81c2cec-d32c-4cf2-9727-d4493e11ee1f": "human_resources_users_people_list",
"d1b1b1b1-1b1b-1b1b-1b1b-1b1b1b1b1b1b": "people_list_only_occupant_tenant_or_owner",
"25cbbaf8-117a-470f-a844-2cfc70f71dde": "human_resources_users_people_list",
"cdf62f06-ec50-40de-b19e-adb3dd34bb95": "people_list_only_occupant_tenant_or_owner",
}
@classmethod
def super_users_people_list(
cls, list_options, token_dict: Union[EmployeeTokenObject, OccupantTokenObject]
cls,
list_options,
token_dict: Union[EmployeeTokenObject, OccupantTokenObject]
):
records = People.filter_all(
*People.get_smart_query(smart_query=list_options.query),
*People.valid_record_args(People),
).data
records = []
if isinstance(token_dict, EmployeeTokenObject):
People.pre_query = People.select_action(
duty_id_list=[
token_dict.selected_company.duty_id,
token_dict.selected_company.bulk_duties_id,
],
)
records = People.filter_all(
*People.get_smart_query(smart_query=list_options.query),
*People.valid_record_args(People),
).data
elif isinstance(token_dict, OccupantTokenObject):
related_users =Users.filter_all(
Users.related_company == token_dict.selected_occupant.responsible_company_id,
).data
People.pre_query = People.filter_all(
People.id.in_([user.person_id for user in related_users]),
*People.valid_record_args(People),
).query
records = People.filter_all(
*People.get_smart_query(smart_query=list_options.query),
*People.valid_record_args(People),
).data
return AlchemyJsonResponse(
completed=True,
message="People are listed successfully",
@@ -42,7 +64,9 @@ class PeopleListEventMethods(MethodToEvent):
@classmethod
def sales_users_people_list(
cls, list_options, token_dict: Union[EmployeeTokenObject, OccupantTokenObject]
cls,
list_options,
token_dict: EmployeeTokenObject,
):
records = People.filter_all(
*People.get_smart_query(smart_query=list_options.query),
@@ -56,12 +80,15 @@ class PeopleListEventMethods(MethodToEvent):
@classmethod
def human_resources_users_people_list(
cls, list_options, token_dict: Union[EmployeeTokenObject, OccupantTokenObject]
cls,
list_options,
token_dict: EmployeeTokenObject,
):
records = People.filter_all(
*People.get_smart_query(smart_query=list_options.query),
*People.valid_record_args(People),
).data
if isinstance(token_dict, EmployeeTokenObject):
records = People.filter_all(
*People.get_smart_query(smart_query=list_options.query),
*People.valid_record_args(People),
).data
return AlchemyJsonResponse(
completed=True,
message="People are listed successfully",
@@ -82,8 +109,12 @@ class PeopleCreateEventMethods(MethodToEvent):
data: InsertPerson,
token_dict: Union[EmployeeTokenObject, OccupantTokenObject],
):
created_user = People.create_action(data=data, token=token_dict)
People.save()
if isinstance(token_dict, EmployeeTokenObject):
created_user = People.create_action(data=data, token=token_dict)
People.save()
elif isinstance(token_dict, OccupantTokenObject):
created_user = People.create_action(data=data, token=token_dict)
People.save()
return JSONResponse(
content={
"completed": True,
@@ -108,26 +139,42 @@ class PeopleUpdateEventMethods(MethodToEvent):
user_uu_id: str,
token_dict: Union[EmployeeTokenObject, OccupantTokenObject],
):
find_one_user = Users.filter_one(
Users.uu_id == user_uu_id,
*Users.valid_record_args(Users),
).data
access_authorized_company = Companies.select_action(
duty_id_list=[getattr(token_dict, "duty_id")],
filter_expr=[Companies.id == find_one_user.id],
)
if access_authorized_company.count:
data_dict = data.excluded_dump()
updated_user = find_one_user.update(**data_dict)
Users.save()
return JSONResponse(
content={
"completed": True,
"message": "Update User record",
"data": updated_user,
},
status_code=status.HTTP_200_OK,
data_dict = data.excluded_dump()
if isinstance(token_dict, EmployeeTokenObject):
find_one_user = Users.filter_one(
Users.uu_id == user_uu_id,
*Users.valid_record_args(Users),
).data
access_authorized_company = Companies.select_action(
duty_id_list=[
token_dict.selected_company.duty_id,
token_dict.selected_company.bulk_duties_id
],
)
if access_authorized_company.count:
updated_user = find_one_user.update(**data_dict)
Users.save()
return JSONResponse(
content={
"completed": True,
"message": "Update User record",
"data": updated_user,
},
status_code=status.HTTP_200_OK,
)
elif isinstance(token_dict, OccupantTokenObject):
find_one_user = People.filter_one(
People.uu_id == user_uu_id,
*People.valid_record_args(People),
).data
access_authorized_company = Companies.select_action(
duty_id_list=[getattr(token_dict, "duty_id")],
filter_expr=[Companies.id == find_one_user.id],
)
if access_authorized_company.count:
data_dict = data.excluded_dump()
updated_user = find_one_user.update(**data_dict)
People.save()
return JSONResponse(
content={"completed": True, "message": "Update User record", "data": {}},
status_code=status.HTTP_200_OK,