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:59:57 +03:00
parent 077d264b28
commit 72bf3b8ddf
18 changed files with 190 additions and 100 deletions
Download Patch File Download Diff File Expand all files Collapse all files

78
api_events/events/account/account_records.py
View File

@@ -1,9 +1,5 @@
import typing
from databases import (
AccountRecords,
)
from api_validations.validations_request import (
InsertAccountRecord,
UpdateAccountRecord,
@@ -13,6 +9,10 @@ from api_validations.validations_request import (
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
from databases import (
AccountRecords,
BuildIbans,
)
class AccountRecordsListEventMethods(MethodToEvent):
@@ -29,14 +29,22 @@ class AccountRecordsListEventMethods(MethodToEvent):
token_dict: typing.Union[EmployeeTokenObject, OccupantTokenObject],
):
if isinstance(token_dict, OccupantTokenObject):
pass
AccountRecords.pre_query = AccountRecords.filter_all(
AccountRecords.company_id
== token_dict.selected_occupant.responsible_company_id,
*AccountRecords.valid_record_args(AccountRecords),
)
elif isinstance(token_dict, EmployeeTokenObject):
pass
AccountRecords.pre_query = AccountRecords.filter_all(
AccountRecords.company_id == token_dict.selected_company.company_id,
*AccountRecords.valid_record_args(AccountRecords),
)
records = AccountRecords.filter_all(
*AccountRecords.smart_query(list_options.query),
*AccountRecords.valid_record_args(AccountRecords),
)
return AlchemyJsonResponse(
completed=True,
message="Update Build record",
result=None
completed=True, message="Update Build record", result=records
)
@@ -53,16 +61,48 @@ class AccountRecordsCreateEventMethods(MethodToEvent):
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,
)
if isinstance(token_dict, OccupantTokenObject):
build_iban = BuildIbans.filter_one(
BuildIbans.iban == data.iban,
BuildIbans.company_id
== token_dict.selected_occupant.responsible_company_id,
).data
if not build_iban:
BuildIbans.raise_http_exception(
status_code="HTTP_404_NOT_FOUND",
error_case="UNAUTHORIZED",
message=f"{data.iban} is not found in company related to your organization",
data={
"iban": data.iban,
},
)
account_record = AccountRecords.find_or_create(**data.excluded_dump())
return AlchemyJsonResponse(
completed=True,
message="Update Build record",
result=account_record.get_dict(),
)
elif isinstance(token_dict, EmployeeTokenObject):
build_iban = BuildIbans.filter_one(
BuildIbans.iban == data.iban,
BuildIbans.company_id == token_dict.selected_company.company_id,
).data
if not build_iban:
BuildIbans.raise_http_exception(
status_code="HTTP_404_NOT_FOUND",
error_case="UNAUTHORIZED",
message=f"{data.iban} is not found in company related to your organization",
data={
"iban": data.iban,
},
)
account_record = AccountRecords.find_or_create(**data.excluded_dump())
return AlchemyJsonResponse(
completed=True,
message="Update Build record",
result=account_record.get_dict(),
)
class AccountRecordsUpdateEventMethods(MethodToEvent):