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

migrator completed

This commit is contained in:
berkay 2024-11-14 18:11:53 +03:00
parent ac037ae54a
commit 1ae1264ace
23 changed files with 439 additions and 207 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
api_events/events/__init__.py
View File

@ -56,6 +56,9 @@ from api_events.events.building.building_build_sites import (
BuildSitesCreateEventMethod,
BuildSitesUpdateEventMethod,
)
from api_events.events.building.building_build_types import (
BuildTypesListEventMethod,
)
from api_events.events.company.company_company import (
CompanyListEventMethod,
CompanyCreateEventMethod,
@ -182,6 +185,7 @@ __all__ = [
"BuildSitesListEventMethod",
"BuildSitesCreateEventMethod",
"BuildSitesUpdateEventMethod",
"BuildTypesListEventMethod",
"DecisionBookListEventMethod",
"DecisionBookCreateEventMethod",
"DecisionBookUpdateEventMethod",

127
api_events/events/address/address.py
View File

@ -13,7 +13,9 @@ from databases import (
from api_validations.validations_request import (
ListOptions,
InsertAddress,
UpdateAddress,
InsertPostCode,
UpdatePostCode,
SearchAddress,
)
from api_validations.core_response import AlchemyJsonResponse
@ -117,16 +119,19 @@ class AddressCreateEventMethods(MethodToEvent):
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"]
data_dict["is_confirmed"] = True
address = Addresses.find_or_create(**data_dict)
RelationshipEmployee2PostCode.find_or_create(
relation_employee = 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()
address.save()
address.update(is_confirmed=True)
address.save()
relation_employee.update(is_confirmed=True)
relation_employee.save()
return JSONResponse(
content={
"completed": True,
@ -220,39 +225,36 @@ class AddressUpdateEventMethods(MethodToEvent):
def update_address(
cls,
address_uu_id: str,
data: InsertAddress,
data: UpdateAddress,
token_dict: Union[EmployeeTokenObject, OccupantTokenObject],
):
address = Addresses.filter_one(
Addresses.uu_id == address_uu_id,
*Addresses.valid_record_args(Addresses),
)
post_code = RelationshipEmployee2PostCode.filter_one(
RelationshipEmployee2PostCode.member_id == address.post_code_id,
*RelationshipEmployee2PostCode.valid_record_args(
RelationshipEmployee2PostCode
),
).data
if not post_code:
raise HTTPException(
status_code=404,
detail="Post code not found. User can not update address without post code.",
)
if isinstance(token_dict, EmployeeTokenObject):
address = Addresses.filter_one(
Addresses.uu_id == address_uu_id,
*Addresses.valid_record_args(Addresses),
).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()
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(
content={
"completed": True,
"message": "Update Address record",
"data": updated_address.get_dict(),
},
status_code=200,
)
data_dict = data.excluded_dump()
updated_address = address.update(**data_dict)
updated_address.save()
return JSONResponse(
content={
"completed": True,
"message": "Update Address record",
"data": updated_address.get_dict(),
},
status_code=200,
)
elif isinstance(token_dict, OccupantTokenObject):
raise HTTPException(
status_code=403,
detail="Occupant can not update address.",
)
class AddressPatchEventMethods(MethodToEvent):
@ -325,16 +327,19 @@ class AddressPostCodeCreateEventMethods(MethodToEvent):
)
data_dump["street_id"] = street.id
data_dump["postcode"] = data.post_code
del data_dump["street_uu_id"], data_dump["post_code"]
del data_dump["post_code"]
post_code = AddressPostcode.find_or_create(**data_dump)
AddressPostcode.__many__table__.find_or_create(
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,
is_confirmed=True,
)
AddressStreet.save()
post_code.save()
post_code.update(is_confirmed=True)
post_code.save()
relation_table.update(is_confirmed=True)
relation_table.save()
return JSONResponse(
content={
"completed": True,
@ -356,30 +361,46 @@ class AddressPostCodeUpdateEventMethods(MethodToEvent):
def update_post_code_address(
cls,
post_code_uu_id: str,
data: InsertPostCode,
data: UpdatePostCode,
token_dict: Union[EmployeeTokenObject, OccupantTokenObject],
):
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,
detail="Street not found. User can not update post code without street.",
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,
*AddressPostcode.valid_record_args(AddressPostcode),
).data
if not post_code:
raise HTTPException(
status_code=404,
detail="Street not found. User can not update post code without street.",
)
updated_post_code = post_code.update(**data.excluded_dump())
updated_post_code.save()
data_dict = data.excluded_dump()
updated_post_code = post_code.update(**data_dict)
updated_post_code.save()
return JSONResponse(
content={
"completed": True,
"message": "Update Post Code record",
"data": updated_post_code.get_dict(),
},
status_code=200,
)
elif isinstance(token_dict, OccupantTokenObject):
raise HTTPException(
status_code=403,
detail="Occupant can not update post code.",
)
return JSONResponse(
content={
"completed": True,
"message": "Update Post Code record",
"data": updated_post_code.get_dict(),
"data": {},
},
status_code=200,
status_code=404,
)

1
api_events/events/authentication.py
View File

@ -77,7 +77,6 @@ class AuthenticationLoginEventMethods(MethodToEvent):
status_code=status.HTTP_401_UNAUTHORIZED,
detail="This User has no active role registered. Please contact your administrator.",
)
return JSONResponse(
content={
"completed": True,

23
api_events/events/building/building_build.py
View File

@ -107,11 +107,20 @@ class BuildCreateEventMethods(MethodToEvent):
part_direction_uu_id=str(api_enum.uu_id),
part_code="MAN-ROOM",
human_livable=False,
is_confirmed=True,
)
man_build_part = BuildParts.find_or_create(**build_parts)
created_build.save()
created_build.update(management_room_id=man_build_part.id)
BuildParts.save()
created_build.save()
man_build_part.update(is_confirmed=True)
man_build_part.save()
# created_build_relation = RelationshipEmployee2Build.find_or_create(
# company_id=token_dict.selected_company.company_id,
# member_id=created_build.id,
# employee_id=token_dict.selected_company.employee_id,
# )
# created_build_relation.update(is_confirmed=True)
# created_build_relation.save()
return JSONResponse(
content={
"completed": True,
@ -147,12 +156,13 @@ class BuildCreateEventMethods(MethodToEvent):
)
created_build = Build.create_action(data=data, token=token_dict)
RelationshipEmployee2Build.find_or_create(
created_build_relation = RelationshipEmployee2Build.find_or_create(
company_id=token_dict.selected_company.company_id,
member_id=created_build.id,
employee_id=token_dict.selected_company.employee_id,
is_confirmed=True,
)
created_build.save()
return JSONResponse(
content={
"completed": True,
@ -207,7 +217,10 @@ class BuildPatchEventMethods(MethodToEvent):
@classmethod
def build_patch(cls, build_uu_id: str, data: PatchRecord, token_dict):
find_one_build = Build.find_one_or_abort(uu_id=build_uu_id)
find_one_build = Build.filter_one(
Build.uu_id==build_uu_id,
*Build.valid_record_args(Build)
)
access_authorized_build = Build.select_action(
employee_id=token_dict.selected_company.employee_id,
filter_expr=[Build.id == find_one_build.id],

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

@ -68,6 +68,8 @@ class BuildAreaCreateEventMethods(MethodToEvent):
data: InsertBuildArea,
token_dict: typing.Union[EmployeeTokenObject, OccupantTokenObject],
):
data_dict = data.excluded_dump()
selected_build = None
if isinstance(token_dict, OccupantTokenObject):
if not token_dict.selected_occupant.build_uuid == data.build_uu_id:
BuildArea.raise_http_exception(
@ -78,6 +80,7 @@ class BuildAreaCreateEventMethods(MethodToEvent):
"build_uu_id": data.build_uu_id,
},
)
selected_build = Build.filter_by_one(system=True, uu_id=data.build_uu_id).data
elif isinstance(token_dict, EmployeeTokenObject):
build_ids = Build.select_action(
employee_id=token_dict.selected_company.employee_id
@ -91,7 +94,10 @@ class BuildAreaCreateEventMethods(MethodToEvent):
"build_uu_id": data.build_uu_id,
},
)
data_dict = data.excluded_dump()
selected_build = Build.filter_by_one(system=True, uu_id=data.build_uu_id).data
data_dict["build_id"] = selected_build.id
data_dict["build_uu_id"] = str(selected_build.uu_id)
created_build_part = BuildArea.find_or_create(**data_dict)
created_build_part.save()
created_build_part.update(is_confirmed=True)
@ -99,7 +105,7 @@ class BuildAreaCreateEventMethods(MethodToEvent):
return AlchemyJsonResponse(
completed=True,
message="Created Build Area",
result=created_build_part,
result=created_build_part.get_dict(),
)

46
api_events/events/building/building_build_types.py
View File

@ -1,5 +1,49 @@
from typing import Union
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 (
ListOptions,
)
from databases.sql_models.building.build import BuildTypes
class BuildingBuildTypesEvents(MethodToEvent): ...
class BuildTypesListEventMethods(MethodToEvent):
event_type = "SELECT"
__event_keys__ = {
"5344d03c-fc47-43ec-8c44-6c2acd7e5d9f": "build_types_list",
}
@classmethod
def build_types_list(
cls,
list_options: ListOptions,
token_dict: Union[EmployeeTokenObject, OccupantTokenObject]
):
from fastapi.exceptions import HTTPException
if isinstance(token_dict, EmployeeTokenObject):
BuildTypes.filter_attr = list_options
results = BuildTypes.filter_all(
*BuildTypes.valid_record_args(BuildTypes)
)
return AlchemyJsonResponse(
completed=True,
result=results,
message="Build Types are listed successfully",
)
elif isinstance(token_dict, OccupantTokenObject):
raise HTTPException(
status_code=403,
detail="You are not authorized to access this endpoint"
)
else:
raise HTTPException(
status_code=403,
detail="You are not authorized to access this endpoint"
)
BuildTypesListEventMethod = BuildTypesListEventMethods(
action=ActionsSchema(endpoint="/building/types/list")
)

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

@ -131,7 +131,10 @@ class CompanyPatchEventMethods(MethodToEvent):
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)
find_one_company = Companies.filter_one(
Companies.uu_id==company_uu_id,
*Companies.valid_record_args(Companies)
).data
access_authorized_company = Companies.select_action(
duty_id_list=[
token_dict.selected_company.bulk_duties_id,

1
api_events/tasks2events/employee_tasks/super_user.py
View File

@ -89,6 +89,7 @@ class SuperUserEventBlock(AddEventFunctionality):
{"function_code": "f6900cb5-ac5b-478e-8e7c-fa87e65cd2e5"},
{"function_code": "76f11a08-5f4a-4e1f-961f-aaef21699acd"},
{"function_code": "41ea7f29-006a-4310-b5c4-b2a0e1a504bd"},
{"function_code": "5344d03c-fc47-43ec-8c44-6c2acd7e5d9f"},
]
def __new__(cls, *args, **kwargs):

10
api_validations/validations_request/__init__.py
View File

@ -5,7 +5,13 @@ from .core_request_validations import (
EndpointPydantic,
BaseModelRegular,
)
from .address import InsertAddress, InsertPostCode, SearchAddress
from .address import (
InsertAddress,
UpdateAddress,
UpdatePostCode,
InsertPostCode,
SearchAddress
)
from .application import (
SingleEnumUUID,
SingleEnumClassKey,
@ -138,6 +144,8 @@ __all__ = [
"EndpointPydantic",
"BaseModelRegular",
"InsertAddress",
"UpdateAddress",
"UpdatePostCode",
"InsertPostCode",
"SearchAddress",
"SingleEnumUUID",

30
api_validations/validations_request/address.py
View File

@ -6,11 +6,16 @@ from api_validations.validations_request import (
from typing import Optional
class InsertPostCode(PydanticBaseModel):
class InsertPostCode(BaseModelRegular):
street_uu_id: str
post_code: str
class UpdatePostCode(PydanticBaseModel):
street_uu_id: Optional[str] = None
post_code: Optional[str] = None
class SearchAddress(PydanticBaseModel):
search: str
list_options: ListOptions
@ -27,7 +32,7 @@ class InsertStreet(PydanticBaseModel):
address_geographic_uu_id: Optional[str] = None
class InsertAddress(PydanticBaseModel):
class InsertAddress(BaseModelRegular):
post_code_uu_id: str
comment_address: Optional[str] = None
@ -42,14 +47,13 @@ class InsertAddress(PydanticBaseModel):
longitude: Optional[float] = None
class _UpdateAddress(PydanticBaseModel):
...
# country_code: Optional[str] = None
# city: Optional[str] = None
# district: Optional[str] = None
# b_state: Optional[str] = None
# neighborhood: Optional[str] = None
# street: Optional[str] = None
# postcode: Optional[str] = None
# latitude: Optional[float] = None
# longitude: Optional[float] = None
class UpdateAddress(PydanticBaseModel):
country_code: Optional[str] = None
city: Optional[str] = None
district: Optional[str] = None
b_state: Optional[str] = None
neighborhood: Optional[str] = None
street: Optional[str] = None
postcode: Optional[str] = None
latitude: Optional[float] = None
longitude: Optional[float] = None

31
databases/sql_models/building/build.py
View File

@ -1,6 +1,7 @@
import typing
from operator import or_
from datetime import datetime, timedelta
from platform import system
from typing import List
from fastapi import HTTPException, status
@ -174,10 +175,10 @@ class Build(CrudCollection, SelectActionWithEmployee):
Integer, nullable=True, comment="Management Room ID"
)
site_id: Mapped[int] = mapped_column(ForeignKey("build_sites.id"))
site_uu_id: Mapped[str] = mapped_column(String, comment="Site UUID")
address_id: Mapped[int] = mapped_column(ForeignKey("addresses.id"))
address_uu_id: Mapped[str] = mapped_column(String, comment="Address UUID")
site_id: Mapped[int] = mapped_column(ForeignKey("build_sites.id"), nullable=True)
site_uu_id: Mapped[str] = mapped_column(String, comment="Site UUID", nullable=True)
address_id: Mapped[int] = mapped_column(ForeignKey("addresses.id"), nullable=False)
address_uu_id: Mapped[str] = mapped_column(String, comment="Address UUID", nullable=False)
build_types_id: Mapped[int] = mapped_column(
ForeignKey("build_types.id"), nullable=False, comment="Building Type"
)
@ -243,22 +244,24 @@ class Build(CrudCollection, SelectActionWithEmployee):
).data
data_dict["address_id"] = official_address.id
data_dict["build_no"] = str(official_address.build_number)
data_dict.pop("address_uu_id", None)
if not data_dict["address_id"]:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Address is not found in database. Re-enter address record then try again.",
)
build_type = BuildTypes.find_one(uu_id=str(data.build_types_uu_id))
build_type = BuildTypes.filter_by_one(system=True, uu_id=str(data.build_types_uu_id)).data
data_dict["build_types_id"] = build_type.id
del data_dict["build_types_uu_id"]
build_created = cls.find_or_create(**data_dict)
cls.__many__table__.find_or_create(
created_build_relation = cls.__many__table__.find_or_create(
company_id=token.selected_company.company_id,
employee_id=token.selected_company.employee_id,
member_id=build_created.id,
is_confirmed=True,
)
build_created.save()
build_created.update(is_confirmed=True)
build_created.save()
created_build_relation.update(is_confirmed=True)
created_build_relation.save()
return build_created
@classmethod
@ -316,7 +319,7 @@ class Build(CrudCollection, SelectActionWithEmployee):
building_types = None
for part in self.parts:
building_types = {}
build_type = BuildTypes.find_one(id=part.build_part_type_id)
build_type = BuildTypes.filter_by_one(system=True, id=part.build_part_type_id).data
if build_type.type_code in building_types:
building_types[build_type.type_code]["list"].append(part.part_no)
else:
@ -412,13 +415,14 @@ class BuildParts(CrudCollection):
building = build_from_duty.first()
if not building:
raise HTTPException(
status_code=status.HTTP_418_IM_A_TEAPOT,
status_code=status.HTTP_406_NOT_ACCEPTABLE,
detail="This Employee can not reach this building or building uu-id not found in database. "
"Check with your supervisor.",
)
if build_types := BuildTypes.filter_one(
BuildTypes.uu_id == data.build_part_type_uu_id
BuildTypes.uu_id == data.build_part_type_uu_id,
*BuildTypes.valid_record_args(BuildTypes)
).data:
part_direction = ApiEnumDropdown.get_by_uuid(
uuid=str(data.part_direction_uu_id)
@ -467,7 +471,7 @@ class BuildParts(CrudCollection):
@property
def part_name(self):
if build_type := BuildTypes.find_one(id=self.build_part_type_id):
if build_type := BuildTypes.filter_by_one(system=True, id=self.build_part_type_id).data:
return f"{str(build_type.type_name).upper()} : {str(self.part_no).upper()}"
return f"Undefined:{str(build_type.type_name).upper()}"
@ -594,6 +598,7 @@ class BuildArea(CrudCollection):
"""
__tablename__ = "build_area"
__exclude__fields__ = []
area_name: Mapped[str] = mapped_column(String, server_default="")
area_code: Mapped[str] = mapped_column(String, server_default="")

58
databases/sql_models/core_mixin.py
View File

@ -125,35 +125,34 @@ class CrudMixin(Base, SmartQueryMixin, SessionMixin, FilterAttributes):
row_attr = bool(getattr(getattr(cls, key), "foreign_keys", None))
if is_primary or row_attr and key_ == Mapped[int]:
return False, None
if key_:
elif val is None:
return True, None
elif str(key[-5:]).lower() == "uu_id":
return True, str(val)
elif key_:
if key_ == Mapped[int]:
return True, int(val) if val is not None else None
return True, int(val)
elif key_ == Mapped[bool]:
return True, bool(val) if val is not None else None
return True, bool(val)
elif key_ == Mapped[float] or key_ == Mapped[NUMERIC]:
return True, round(float(val), 3) if val is not None else None
return True, round(float(val), 3)
elif key_ == Mapped[int]:
return True, int(val) if val is not None else None
return True, int(val)
elif key_ == Mapped[TIMESTAMP]:
formatted_date = client_arrow.get(str(val)).format(
"DD-MM-YYYY HH:mm:ss"
)
return True, str(formatted_date) if val is not None else None
return True, str(client_arrow.get(str(val)).format("DD-MM-YYYY HH:mm:ss"))
elif key_ == Mapped[str]:
return True, str(val)
else:
if isinstance(val, datetime.datetime):
formatted_date = client_arrow.get(str(val)).format(
"DD-MM-YYYY HH:mm:ss"
)
return True, str(formatted_date) if val is not None else None
return True, str(client_arrow.get(str(val)).format("DD-MM-YYYY HH:mm:ss"))
elif isinstance(value_type, bool):
return True, bool(val) if val is not None else None
return True, bool(val)
elif isinstance(value_type, float) or isinstance(value_type, Decimal):
return True, round(float(val), 3) if val is not None else None
return True, round(float(val), 3)
elif isinstance(value_type, int):
return True, int(val) if val is not None else None
return True, int(val)
elif isinstance(value_type, str):
return True, str(val) if val is not None else None
return True, str(val)
elif isinstance(value_type, type(None)):
return True, None
return False, None
@ -242,10 +241,12 @@ class CrudMixin(Base, SmartQueryMixin, SessionMixin, FilterAttributes):
exclude_list = [
element
for element in self.__system_default_model__
if str(element)[-2:] == "id"
if str(element)[-2:] == "id" and str(element)[-5:].lower() == 'uu_id'
]
columns_include_list = list(set(include).difference(set(exclude_list)))
columns_include_list.extend(["uu_id", "active"])
# columns_include_list.extend([column for column in self.columns if str(column)[-5:].lower() == 'uu_id'])
columns_include_list.extend(['uu_id'])
for key in list(columns_include_list):
val = getattr(self, key)
correct, value_of_database = self.iterate_over_variables(val, key)
@ -262,7 +263,10 @@ class CrudMixin(Base, SmartQueryMixin, SessionMixin, FilterAttributes):
if str(element)[-2:] == "id"
]
)
columns_excluded_list = set(self.columns).difference(set(exclude))
columns_excluded_list = list(set(self.columns).difference(set(exclude)))
# columns_excluded_list.extend([column for column in self.columns if str(column)[-5:].lower() == 'uu_id'])
columns_excluded_list.extend(['uu_id', 'active'])
for key in list(columns_excluded_list):
val = getattr(self, key)
correct, value_of_database = self.iterate_over_variables(val, key)
@ -274,16 +278,17 @@ class CrudMixin(Base, SmartQueryMixin, SessionMixin, FilterAttributes):
)
columns_list = list(set(self.columns).difference(set(exclude_list)))
columns_list = [
columns
for columns in columns_list
if str(columns)[-2:] != "id" and "uu_id" not in str(columns)
columns for columns in columns_list if str(columns)[-2:] != "id"
]
columns_list.extend([column for column in self.columns if str(column)[-5:].lower() == 'uu_id'])
for remove_field in self.__system_default_model__:
if remove_field in columns_list:
columns_list.remove(remove_field)
for key in list(columns_list):
val = getattr(self, key)
correct, value_of_database = self.iterate_over_variables(val, key)
if correct:
return_dict[key] = value_of_database
# all_arguments = [
# record
# for record in self.__class__.__dict__
@ -305,7 +310,8 @@ class CrudMixin(Base, SmartQueryMixin, SessionMixin, FilterAttributes):
# return_dict[all_argument] = (
# populate_arg.get_dict() if populate_arg else []
# )
return dict(sorted(return_dict.items(), reverse=False))
# return dict(sorted(return_dict.items(), reverse=False))
return return_dict
class BaseMixin(CrudMixin, ReprMixin, SerializeMixin, FilterAttributes):

57
databases/sql_models/others/enums.py
View File

@ -9,10 +9,10 @@ from sqlalchemy.orm import (
Mapped,
mapped_column,
)
from databases.sql_models.core_mixin import BaseCollection
from databases.sql_models.core_mixin import CrudCollection
class ApiEnumDropdown(BaseCollection):
class ApiEnumDropdown(CrudCollection):
__tablename__ = "api_enum_dropdown"
__exclude__fields__ = ["enum_class"]
@ -31,35 +31,30 @@ class ApiEnumDropdown(BaseCollection):
@classmethod
def get_by_uuid(cls, uuid: str):
return cls.query.filter(cls.uu_id == uuid).first()
return cls.filter_by_one(system=True, uu_id=str(uuid)).data
@classmethod
def get_debit_search(cls, search_debit: str = None, search_uu_id: str = None):
if search_uu_id:
if search := cls.query.filter(
cls.enum_class.in_(["DebitTypes"]),
cls.uu_id == search_uu_id,
cls.active == True,
).first():
if search := cls.filter_one(
cls.enum_class.in_(["DebitTypes"]), cls.uu_id == search_uu_id, system=True
).data:
return search
elif search_debit:
if search := cls.query.filter(
cls.enum_class.in_(["DebitTypes"]),
cls.key == search_debit,
cls.active == True,
).first():
if search := cls.filter_one(
cls.enum_class.in_(["DebitTypes"]), cls.key == search_debit, system=True
).data:
return search
return cls.query.filter(
cls.enum_class.in_(["DebitTypes"]),
cls.active == True,
).all()
return cls.filter_all(
cls.enum_class.in_(["DebitTypes"]), system=True
).data
@classmethod
def get_due_types(cls):
if due_list := cls.filter_all(
cls.enum_class == "BuildDuesTypes",
cls.key.in_(["BDT-A", "BDT-D"]),
cls.active == True,
system=True
).data:
return [due.uu_id.__str__() for due in due_list]
raise HTTPException(
@ -70,23 +65,21 @@ class ApiEnumDropdown(BaseCollection):
@classmethod
def due_type_search(cls, search_management: str = None, search_uu_id: str = None):
if search_uu_id:
if search := cls.query.filter(
cls.enum_class.in_(["BuildDuesTypes"]),
cls.uu_id == search_uu_id,
cls.active == True,
).first():
if search := cls.filter_one(
cls.enum_class.in_(["BuildDuesTypes"]), cls.uu_id == search_uu_id,
system=True
).data:
return search
elif search_management:
if search := cls.query.filter(
cls.enum_class.in_(["BuildDuesTypes"]),
cls.key == search_management,
cls.active == True,
).first():
if search := cls.filter_one(
cls.enum_class.in_(["BuildDuesTypes"]), cls.key == search_management,
system=True
).data:
return search
return cls.query.filter(
return cls.filter_all(
cls.enum_class.in_(["BuildDuesTypes"]),
cls.active == True,
).all()
system=True
).data
def get_enum_dict(self):
return {
@ -101,7 +94,7 @@ class ApiEnumDropdown(BaseCollection):
def uuid_of_enum(cls, enum_class: str, key: str):
return str(
getattr(
cls.filter_one(cls.enum_class == enum_class, cls.key == key).data,
cls.filter_one(cls.enum_class == enum_class, cls.key == key, system=True).data,
"uu_id",
None,
)

3
service_app/routers/address/address/router.py
View File

@ -3,6 +3,7 @@ from fastapi.requests import Request
from api_validations.validations_request import (
InsertAddress,
UpdateAddress,
SearchAddress,
ListOptions,
PatchRecord,
@ -36,7 +37,7 @@ def address_search(request: Request, data: SearchAddress):
@address_router.post(
path="/update/{address_uu_id}", summary="Update Address with given auth levels"
)
def address_update(request: Request, address_uu_id: str, data: InsertAddress):
def address_update(request: Request, address_uu_id: str, data: UpdateAddress):
token_dict = parse_token_object_to_dict(request=request)
return token_dict.available_event(
data=data, address_uu_id=address_uu_id, token_dict=token_dict

3
service_app/routers/address/post_code/router.py
View File

@ -3,6 +3,7 @@ from fastapi.requests import Request
from api_validations.validations_request import (
InsertPostCode,
UpdatePostCode,
ListOptions,
PatchRecord,
)
@ -29,7 +30,7 @@ def post_code_create(request: Request, data: InsertPostCode):
@post_code_router.post(
path="/update/{post_code_uu_id}", summary="Update PostCode with given auth levels"
)
def post_code_update(request: Request, post_code_uu_id: str, data: InsertPostCode):
def post_code_update(request: Request, post_code_uu_id: str, data: UpdatePostCode):
token_dict = parse_token_object_to_dict(request=request)
return token_dict.available_event(
data=data, post_code_uu_id=post_code_uu_id, token_dict=token_dict

4
service_app/routers/building/buildtypes/router.py
View File

@ -21,9 +21,7 @@ def building_types_list(request: Request, list_options: ListOptions):
return token_dict.available_event(list_options=list_options, token_dict=token_dict)
@build_types_route.post(
path="/create", summary="Create BuildParts with given auth levels"
)
@build_types_route.post(path="/create", summary="Create BuildParts with given auth levels")
def building_types_create(request: Request, data: InsertBuildTypes):
token_dict = parse_token_object_to_dict(request=request)
return token_dict.available_event(data=data, token_dict=token_dict)

4
service_app_init/initialize_app/event_initator.py
View File

@ -63,7 +63,9 @@ def create_all_events_from_actions():
endpoint_uu_id=str(endpoint_restriction.uu_id),
**active_confirmed,
)
Events.save()
created_event.save()
created_event.update(is_confirmed=True)
created_event.save()
print(f"Event created: {created_event.uu_id}")
for item in an_empty_list:

12
service_app_init/initialize_app/initiator.py
View File

@ -69,9 +69,11 @@ def init_api_enums_build_types():
lang=insert_type["lang"],
type_code=str(insert_type["type_code"]).upper(),
type_name=insert_type["type_name"],
is_confirmed=True,
)
BuildTypes.find_or_create(**build_types.model_dump())
created_build_type = BuildTypes.find_or_create(**build_types.model_dump())
created_build_type.save()
created_build_type.update(is_confirmed=True)
created_build_type.save()
insert_enums = [
{"enum_class": "BuildDuesTypes", "type_code": "BDT-D", "type_name": "Debit"},
@ -230,13 +232,15 @@ def init_api_enums_build_types():
},
]
for insert_enum in insert_enums:
ApiEnumDropdown.find_or_create(
created_api_enum = ApiEnumDropdown.find_or_create(
enum_class=insert_enum["enum_class"],
value=insert_enum["type_name"],
key=str(insert_enum["type_code"]).upper(),
description=insert_enum["type_name"],
)
BuildTypes.save()
created_api_enum.save()
created_api_enum.update(is_confirmed=True)
created_api_enum.save()
# for insert_meeting_type in insert_meeting_types:
# ApiEnumDropdown.find_or_create(

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

@ -17,7 +17,7 @@ requester_dict_build_area = lambda data: {
"data": data,
}
requester_dict_build_part = lambda data: {
"endpoint": "/building/part/create",
"endpoint": "/building/parts/create",
"data": data,
}
requester_dict_build_iban = lambda data: {
@ -52,10 +52,12 @@ def get_build_area_from_json():
return with_pydantic
def get_build_part_from_json():
def get_build_part_from_json(build_part_type_uu_id: str, part_direction_uu_id: str):
read_files_json, with_pydantic = read_json_file(json_file="build_parts"), []
read_files = read_files_json.get("build_parts")
for row in read_files:
row["build_part_type_uu_id"] = build_part_type_uu_id
row["part_direction_uu_id"] = part_direction_uu_id
pydantic_row = InsertBuildParts(**row)
with_pydantic.append(pydantic_row.model_dump())
if not with_pydantic:
@ -100,7 +102,14 @@ def create_address(requester: BothAPIS, post_code):
)
print("post_code_response", post_code_response.text)
response_post_code = post_code_response.json()
return response_post_code
response_post_code_uu_id = response_post_code["data"]["uu_id"]
requester.local_api.post(
endpoint=f"/postcode/update/{response_post_code_uu_id}",
data={
"is_confirmed": True,
},
)
return response_post_code_uu_id
def search_street(search_text, requester):
@ -134,50 +143,24 @@ def create_addresses(address, requester):
data=address,
)
print("text", address_response.text)
print("json", address_response.json())
response_address = address_response.json()
print("json", response_address)
return response_address
def migrate_build(requester: BothAPIS):
street_uu_id = search_street("Reşat Nuri", requester=requester)
response_post_code = create_address(requester=requester, post_code=post_code_dict(uu_id_street=street_uu_id))
print("response_post_code", response_post_code)
created_address = create_addresses(
address=address_dict(post_code_uu_id=response_post_code["data"]["uu_id"]),
requester=requester
)
print('created_address', created_address)
created_address_uu_id = created_address["data"]["uu_id"]
for response_data in get_build_from_json():
print("response_data", response_data)
response_data["address_uu_id"] = created_address_uu_id
exit()
response = requester.local_api.post(**requester_dict_build(data=response_data))
print("response", response.text)
if response.ok:
build_uu_id = response.json()["data"]["uu_id"]
print("response build_uu_id", build_uu_id)
exit()
migrate_build_area(requester=requester, build_uu_id=build_uu_id)
migrate_build_part(requester=requester, build_uu_id=build_uu_id)
migrate_build_iban(requester=requester, build_uu_id=build_uu_id)
return
def grab_new_build_uu_id(requester: BothAPIS, build_uu_id: str):
response_wag = requester.wag_api.post(
endpoint="/building/build/list",
data={"page": 1, "size": 1, "query": {"uu_id": build_uu_id}},
)
build_uu_id = response_wag.json()["data"]["uu_id"]
if not build_uu_id:
raise Exception("Build UU ID not found")
return build_uu_id
# def grab_new_build_uu_id(requester: BothAPIS, build_uu_id: str):
# response_wag = requester.wag_api.post(
# endpoint="/building/build/list",
# data={"page": 1, "size": 1, "query": {"uu_id": build_uu_id}},
# )
# build_uu_id = response_wag.json()["data"]["uu_id"]
# if not build_uu_id:
# raise Exception("Build UU ID not found")
# return build_uu_id
def migrate_build_area(requester: BothAPIS, build_uu_id: str):
build_uu_id = grab_new_build_uu_id(requester=requester, build_uu_id=build_uu_id)
# build_uu_id = grab_new_build_uu_id(requester=requester, build_uu_id=build_uu_id)
response_datas = get_build_area_from_json()
for response_data in response_datas:
response_data["build_uu_id"] = build_uu_id
@ -189,10 +172,17 @@ def migrate_build_area(requester: BothAPIS, build_uu_id: str):
return
def migrate_build_part(requester: BothAPIS, build_uu_id: str):
build_uu_id = grab_new_build_uu_id(requester=requester, build_uu_id=build_uu_id)
response_datas = get_build_part_from_json()
def migrate_build_part(
requester: BothAPIS, build_uu_id: str, build_part_type_uu_id: str, part_direction_uu_id: str
):
# build_uu_id = grab_new_build_uu_id(requester=requester, build_uu_id=build_uu_id)
response_datas = get_build_part_from_json(
build_part_type_uu_id=build_part_type_uu_id, part_direction_uu_id=part_direction_uu_id
)
for response_data in response_datas:
if response_data.get('part_no') == 0:
continue
response_data["build_uu_id"] = build_uu_id
response = requester.local_api.post(
**requester_dict_build_part(data=response_data)
@ -202,7 +192,7 @@ def migrate_build_part(requester: BothAPIS, build_uu_id: str):
def migrate_build_iban(requester: BothAPIS, build_uu_id: str):
build_uu_id = grab_new_build_uu_id(requester=requester, build_uu_id=build_uu_id)
# build_uu_id = grab_new_build_uu_id(requester=requester, build_uu_id=build_uu_id)
response_datas = get_build_iban_from_json()
for response_data in response_datas:
response_data["build_uu_id"] = build_uu_id
@ -214,7 +204,7 @@ def migrate_build_iban(requester: BothAPIS, build_uu_id: str):
def migrate_build_living_space(requester: BothAPIS, build_uu_id: str):
build_uu_id = grab_new_build_uu_id(requester=requester, build_uu_id=build_uu_id)
# build_uu_id = grab_new_build_uu_id(requester=requester, build_uu_id=build_uu_id)
response_datas = get_build_iban_from_json()
for response_data in response_datas:
response_data["build_uu_id"] = build_uu_id
@ -223,3 +213,43 @@ def migrate_build_living_space(requester: BothAPIS, build_uu_id: str):
)
print("response", response.text)
return
def migrate_build(requester: BothAPIS):
from service_app_test.test_application.migrate_old_data.get_building_types import (
list_building_types,
)
from service_app_test.test_application.migrate_old_data.get_type_codes import (
get_type_codes_key_and_class
)
building_type_flat = list_building_types(requester=requester.local_api, type_code="daire").get("uu_id")
building_type_build = list_building_types(requester=requester.local_api, type_code="apt").get("uu_id")
api_enum_dropdown_nn = get_type_codes_key_and_class(
class_name="Directions", key_name="NN", requester=requester.local_api
)
api_enum_dropdown_nn_uuid = api_enum_dropdown_nn.get("data").get("uu_id")
street_uu_id = search_street("Reşat Nuri", requester=requester)
post_code_uuid = create_address(requester=requester, post_code=post_code_dict(uu_id_street=street_uu_id))
created_address = create_addresses(
address=address_dict(post_code_uu_id=post_code_uuid),
requester=requester
)
created_address_uu_id = created_address["data"]["uu_id"]
for response_data in get_build_from_json():
response_data["address_uu_id"] = created_address_uu_id
response_data["build_types_uu_id"] = building_type_build
response = requester.local_api.post(**requester_dict_build(data=response_data))
response_json = response.json()
build_uu_id = str(response_json.get("data").get("uu_id"))
migrate_build_area(requester=requester, build_uu_id=build_uu_id)
migrate_build_part(
requester=requester,
build_uu_id=build_uu_id,
build_part_type_uu_id=building_type_flat,
part_direction_uu_id=api_enum_dropdown_nn_uuid
)
migrate_build_iban(requester=requester, build_uu_id=build_uu_id)
return

29
service_app_test/test_application/migrate_old_data/get_building_types.py Normal file
View File

@ -0,0 +1,29 @@
def list_building_types(requester, lang: str = "TR", type_code: str = None):
from service_app_test.test_application.evyos.datas.company_employee_data import (
list_options,
)
options_smart_query = {
**list_options,
"query": {
"lang": lang,
},
}
if type_code:
options_smart_query["query"]["type_code__ilike"] = type_code
response = requester.post(
endpoint="/building/types/list",
data=options_smart_query,
)
if type_code:
for item in response.json().get("data"):
if str(item.get("type_code")).upper() == str(type_code).upper():
return item
print("text", response.text)
print("json", response.json())
return response.json()
# list_building_types(type_code="apt")

18
service_app_test/test_application/migrate_old_data/get_occupants_codes.py Normal file
View File

@ -0,0 +1,18 @@
def get_occupants_types(occupant_code, requester):
response = requester.post(
endpoint="/occupant_types/get/code",
data={"type_code": occupant_code},
)
print("text", response.text)
print("json", response.json())
return response.json()
def get_occupants_types_uu_id(occupant_uu_id, requester):
response = requester.post(
endpoint="/occupant_types/get/uu_id",
data={"uu_id": occupant_uu_id},
)
print("text", response.text)
print("json", response.json())
return response.json()

42
service_app_test/test_application/migrate_old_data/get_type_codes.py Normal file
View File

@ -0,0 +1,42 @@
def get_type_codes_key_and_class(class_name, key_name, requester):
response = requester.post(
endpoint="/enums/get/key",
data={"class_name": class_name, "key_name": key_name},
)
print("text", response.text)
print("json", response.json())
return response.json()
def get_type_codes_via_uuid(uu_id, requester):
response = requester.post(
endpoint="/enums/get/uu_id",
data={"uu_id": uu_id},
)
print("text", response.text)
print("json", response.json())
return response.json()
def get_all_type_codes(requester):
response = requester.post(
endpoint="/enums/list/all",
data={},
)
print("text", response.text)
print("json", response.json())
return response.json()
def get_all_type_codes_with_class(class_name, requester):
response = requester.post(
endpoint="/enums/list/class",
data={"class_name": class_name},
)
print("text", response.text)
print("json", response.json())
return response.json()
# get_type_codes(class_name="buildmanagementtype", key_name="BM")
# get_all_type_codes()

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

@ -7,13 +7,13 @@ from service_app_test.test_application.migrate_old_data.building import (
)
from service_app_test.test_application.migrate_old_data.company import migrate_company
password_token = "b-4liqvEJIX9wtmMScnJhDOkPTIvyra1I_1HQCgTKG3Mp6Oaj-Vh8dVqqpZFC-fGlQ-5bnuDpzrju4Jg6qoi48EB5brdNT4YQCLdQqMlO8uUrL8iuJmRPk4L9AOQl82NFXD0U4pbZ9fhZkp4vHl487S9HcO1Dz5qUYR1VOs5mt2p0d96c5qrWB4QcDkkbz2F"
login_data = {
"domain": "evyos.com.tr",
"access_key": "karatay.berkay.sup@evyos.com.tr",
"password": "string",
"remember_me": False,
"password_token": "Jx4t2JM3Xc_yNGCBe-hk_4c299ov51_s20kQueVUjm-NJU0KGWENm3alGfyGO6-y79esC5WtuhXHAsw35XJYFn-UOdUAvuW6kbHk-F2MTNbavXMp2f_UdkPrav0PppZEQ2TRcIJQ6DzFWO0ONOxJMhcGTUbqxO4DfidKiq_VjYUqfwcl6ZxSsJolctLzyfwf",
"password_token": password_token
}
login_data_wag = {
"domain": "evyos.com.tr",
@ -39,6 +39,6 @@ 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)