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

build area updated

This commit is contained in:
berkay 2024-11-12 13:57:14 +03:00
parent ffb85a62f6
commit 952d742999
5 changed files with 311 additions and 10 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -1,5 +1,153 @@
import typing
from databases import (
Build,
BuildArea,
)
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 BuildingBuildAreaEvents(MethodToEvent): ...
class BuildListEventMethods(MethodToEvent):
event_type = "SELECT"
__event_keys__ = {
"0bb51845-65a2-4340-8872-a3b5aad95468": "build_area_list",
}
@classmethod
def build_area_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]
BuildArea.pre_query = BuildArea.filter_all(
BuildArea.build_id.in_(build_id_list),
*BuildArea.valid_record_args(BuildArea),
).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]
BuildArea.pre_query = BuildArea.filter_all(
BuildArea.build_id.in_(build_id_list),
*BuildArea.valid_record_args(BuildArea),
).query
BuildArea.filter_attr = list_options
records = BuildArea.filter_all(
*BuildArea.get_smart_query(smart_query=list_options.query),
*BuildArea.valid_record_args(BuildArea),
)
return AlchemyJsonResponse(
completed=True,
message="Update Build record",
result=records
)
class BuildCreateEventMethods(MethodToEvent):
event_type = "CREATE"
__event_keys__ = {
"a10571fa-ac1d-4546-9272-cacb911d8004": "build_area_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:
BuildArea.raise_http_exception(
status_code="HTTP_403_FORBIDDEN",
error_case="UNAUTHORIZED",
message=f"Occupant can not create build area 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]:
BuildArea.raise_http_exception(
status_code="HTTP_403_FORBIDDEN",
error_case="UNAUTHORIZED",
message=f"Employee can not create build area for {data.build_uu_id}",
data={
"build_uu_id": data.build_uu_id,
},
)
data_dict = data.excluded_dump()
created_build_part = BuildArea.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 BuildUpdateEventMethods(MethodToEvent):
event_type = "UPDATE"
__event_keys__ = {
"58178738-7489-4f8f-954e-5c8f083c1845": "build_area_update",
}
@classmethod
def build_area_update(cls, build_uu_id: str, data: UpdateBuildArea, token_dict):
return AlchemyJsonResponse(
completed=False,
message="Update Build record",
result=None,
)
class BuildPatchEventMethods(MethodToEvent):
event_type = "PATCH"
__event_keys__ = {
"d6bd8a5f-fa76-49da-b82e-4a95f1bcce39": "build_area_patch",
}
@classmethod
def build_area_patch(cls, build_uu_id: str, data, token_dict):
return AlchemyJsonResponse(
completed=False,
message="Update Build record",
result=None,
)
BuildListEventMethod = BuildListEventMethods(
action=ActionsSchema(endpoint="/building/area/list")
)
BuildCreateEventMethod = BuildCreateEventMethods(
action=ActionsSchema(endpoint="/building/area/create")
)
BuildUpdateEventMethod = BuildUpdateEventMethods(
action=ActionsSchema(endpoint="/building/area/update")
)
BuildPatchEventMethod = BuildPatchEventMethods(
action=ActionsSchema(endpoint="/building/area/patch")
)

6
databases/sql_models/identity/identity.py
View File

@ -338,7 +338,7 @@ class People(CrudCollection, SelectAction):
@classmethod
def create_action(cls, data: InsertPerson, token):
from databases import Employees, Duties
from databases import Duties
token_duties_id, token_company_id = (
token.selected_company.duty_id,
@ -364,11 +364,13 @@ class People(CrudCollection, SelectAction):
create_dict["surname"] = str(create_dict["surname"]).upper()
create_dict["birth_place"] = str(create_dict["birth_place"]).upper()
created_people = cls.find_or_create(**create_dict)
RelationshipDutyPeople.find_or_create(
created_people.update(is_confirmed=True)
duty_people = RelationshipDutyPeople.find_or_create(
company_id=token.selected_company.company_id,
duties_id=bulk_duty.id,
member_id=created_people.id,
)
duty_people.update(is_confirmed=True)
return created_people

127
service_app_test/test_application/migrate_old_data/building.py
View File

@ -1,21 +1,144 @@
from service_app_test.api_configs import BothAPIS
from service_app_test.bases import FilterObject
def migrate_build(requester: BothAPIS):
# Migrate old data
filter_object = FilterObject(
page=1,
size=100,
)
response = requester.wag_api.post(
endpoint="/building/build/list",
data=filter_object.dump(),
)
response_json = response.json()
response_datas = response_json["data"]
for response_data in response_datas:
new_response_data = dict()
for key, value in dict(response_data).items():
if value is not None and not str(value) == "None":
new_response_data[key] = response_data[key]
new_response_data.pop("uu_id", None)
print("new_response_data", new_response_data)
exit()
response = requester.local_api.post(
endpoint="/building/build/create",
data=new_response_data,
)
print("response", response.text)
return
def migrate_build_area(requester: BothAPIS):
def migrate_build_area(requester: BothAPIS, build_uu_id: str):
# Migrate old data
filter_object = FilterObject(
page=1,
size=100,
)
response = requester.wag_api.post(
endpoint="/building/area/list",
data=filter_object.dump(),
)
print('response', response.text)
exit()
response_json = response.json()
response_datas = response_json["data"]
for response_data in response_datas:
new_response_data = dict()
for key, value in dict(response_data).items():
if value is not None and not str(value) == "None":
new_response_data[key] = response_data[key]
new_response_data.pop("uu_id", None)
new_response_data['build_uu_id'] = str(build_uu_id)
exit()
print("new_response_data", new_response_data)
response = requester.local_api.post(
endpoint="/building/area/create",
data=new_response_data,
)
print("response", response.text)
return
def migrate_build_part(requester: BothAPIS):
def migrate_build_part(requester: BothAPIS, build_uu_id: str):
# Migrate old data
filter_object = FilterObject(
page=1,
size=100,
)
response = requester.wag_api.post(
endpoint="/build/list",
data=filter_object.dump(),
)
response_json = response.json()
response_datas = response_json["data"]
for response_data in response_datas:
new_response_data = dict()
for key, value in dict(response_data).items():
if value is not None and not str(value) == "None":
new_response_data[key] = response_data[key]
new_response_data.pop("uu_id", None)
print("new_response_data", new_response_data)
new_response_data['build_uu_id'] = str(build_uu_id)
exit()
response = requester.local_api.post(
endpoint="/build/create",
data=new_response_data,
)
print("response", response.text)
return
def migrate_build_iban(requester: BothAPIS):
# Migrate old data
filter_object = FilterObject(
page=1,
size=100,
)
response = requester.wag_api.post(
endpoint="/build/list",
data=filter_object.dump(),
)
response_json = response.json()
response_datas = response_json["data"]
for response_data in response_datas:
new_response_data = dict()
for key, value in dict(response_data).items():
if value is not None and not str(value) == "None":
new_response_data[key] = response_data[key]
new_response_data.pop("uu_id", None)
print("new_response_data", new_response_data)
response = requester.local_api.post(
endpoint="/build/create",
data=new_response_data,
)
print("response", response.text)
return
def migrate_build_living_space(requester: BothAPIS):
# Migrate old data
filter_object = FilterObject(
page=1,
size=100,
)
response = requester.wag_api.post(
endpoint="/build/list",
data=filter_object.dump(),
)
response_json = response.json()
response_datas = response_json["data"]
for response_data in response_datas:
new_response_data = dict()
for key, value in dict(response_data).items():
if value is not None and not str(value) == "None":
new_response_data[key] = response_data[key]
new_response_data.pop("uu_id", None)
print("new_response_data", new_response_data)
response = requester.local_api.post(
endpoint="/build/create",
data=new_response_data,
)
print("response", response.text)
return

29
service_app_test/test_application/migrate_old_data/people.py
View File

@ -1,6 +1,33 @@
from service_app_test.api_configs import BothAPIS
from service_app_test.bases import FilterObject
def migrate_people(requester: BothAPIS):
# Migrate old data
pass
filter_object = FilterObject(
page=1,
size=100,
)
response = requester.wag_api.post(
endpoint="/people/list",
data=filter_object.dump(),
)
response_json = response.json()
response_datas = response_json["data"]
counter = 0
for response_data in response_datas:
new_response_data = dict()
for key, value in dict(response_data).items():
if value is not None and not str(value) == "None":
new_response_data[key] = response_data[key]
new_response_data.pop("uu_id", None)
print("new_response_data", new_response_data)
new_response_data["national_identity_id"] = f"000000000{str(counter).zfill(2)}"
response = requester.local_api.post(
endpoint="/people/create",
data=new_response_data,
)
print("response", response.text)
if response.ok:
counter += 1
return

9
service_app_test/test_application/migrate_old_data/runner.py
View File

@ -43,11 +43,12 @@ both_apis = BothAPIS()
both_apis.wag_api = wag_api
both_apis.local_api = local_api
migrate_company(requester=both_apis)
migrate_people(requester=both_apis)
# migrate_company(requester=both_apis)
# migrate_people(requester=both_apis)
# migrate_build(requester=both_apis)
migrate_build_area(requester=both_apis, build_uu_id="string")
exit()
migrate_build(requester=both_apis)
migrate_build_area(requester=both_apis)
migrate_build_part(requester=both_apis)
migrate_build_iban(requester=both_apis)
migrate_build_living_space(requester=both_apis)