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

bank updated

Browse Source
This commit is contained in:
berkay
2024-12-02 11:40:51 +03:00
parent a4fd52c28a
commit 665d961be8
26 changed files with 335 additions and 253 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
api_events/events/__init__.py
View File

@@ -163,6 +163,11 @@ from api_events.events.decision_book.decision_book_invitations import (
BuildDecisionBookInvitationsCreateEventMethod,
BuildDecisionBookInvitationsUpdateEventMethod,
)
from api_events.events.events.events_events import (
EventsBindEventToEmployeeMethod,
EventsBindEventToOccupantMethod,
EventsListEventMethod,
)
__all__ = [
@@ -279,4 +284,7 @@ __all__ = [
"BuildDecisionBookInvitationsUpdateEventMethod",
"DecisionBookPersonAttendEventMethod",
"DecisionBookPersonAssignOccupantEventMethod",
"EventsBindEventToEmployeeMethod",
"EventsBindEventToOccupantMethod",
"EventsListEventMethod",
]

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

@@ -53,6 +53,7 @@ class AccountRecordsListEventMethods(MethodToEvent):
"""
return
class AccountRecordsCreateEventMethods(MethodToEvent):
event_type = "CREATE"

18
api_events/events/authentication.py
View File

@@ -331,7 +331,9 @@ class AuthenticationChangePasswordEventMethods(MethodToEvent):
Users.id == token_dict.user_id,
).data:
if found_user.check_password(data.old_password):
found_user.create_password(found_user=found_user, password=data.new_password)
found_user.create_password(
found_user=found_user, password=data.new_password
)
return JSONResponse(
content={
"completed": True,
@@ -408,10 +410,7 @@ class AuthenticationResetPasswordEventMethods(MethodToEvent):
}
@classmethod
def authentication_reset_password(
cls,
data: Forgot
):
def authentication_reset_password(cls, data: Forgot):
from sqlalchemy import or_
found_user = Users.query.filter(
@@ -426,7 +425,7 @@ class AuthenticationResetPasswordEventMethods(MethodToEvent):
detail="Given access key or domain is not matching with the any user record.",
)
reset_password_token = found_user.reset_password_token()
reset_password_token = found_user.reset_password_token(found_user=found_user)
send_email_completed = send_email(
subject=f"Dear {found_user.user_tag}, a password reset request has been received.",
receivers=[str(found_user.email)],
@@ -437,13 +436,12 @@ class AuthenticationResetPasswordEventMethods(MethodToEvent):
)
if not send_email_completed:
raise found_user.raise_http_exception(
status_code=400,
message="Email can not be sent. Try again later"
status_code=400, message="Email can not be sent. Try again later"
)
return JSONResponse(
content={
"completed": True,
"message": "Password is created successfully",
"message": "Password change link is sent to your email or phone",
"data": found_user.get_dict(),
},
status_code=status.HTTP_200_OK,
@@ -537,7 +535,7 @@ class AuthenticationLogoutEventMethods(MethodToEvent):
)
token_users = get_object_via_user_uu_id(token_dict.user_uu_id)
for token, token_user in token_users.items():
if token_user['domain'] == data.domain:
if token_user["domain"] == data.domain:
selected_user = Users.filter_one(
Users.uu_id == token_dict.user_uu_id,
).data

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

@@ -1,4 +1,5 @@
import typing
from typing import Union
from fastapi import status, HTTPException
from fastapi.responses import JSONResponse
@@ -28,6 +29,7 @@ from api_objects.auth.token_objects import EmployeeTokenObject, OccupantTokenObj
class BuildListEventMethods(MethodToEvent):
event_type = "SELECT"
__event_keys__ = {
"68b3b5ed-b74c-4a27-820f-3959214e94e9": "build_list",
}
@@ -78,7 +80,6 @@ class BuildCreateEventMethods(MethodToEvent):
)
created_build = Build.create_action(data=data, token=token_dict)
build_type = BuildTypes.filter_by_one(
**BuildTypes.valid_record_dict, type_code="APT_YNT"
).data
@@ -178,7 +179,12 @@ class BuildUpdateEventMethods(MethodToEvent):
}
@classmethod
def build_update(cls, build_uu_id: str, data: UpdateBuild, token_dict):
def build_update(
cls,
build_uu_id: str,
data: UpdateBuild,
token_dict: Union[EmployeeTokenObject, OccupantTokenObject]
):
Build.pre_query = Build.select_action(
employee_id=token_dict.selected_company.employee_id
)

227
api_events/events/events/events_events.py
View File

@@ -2,6 +2,7 @@ from typing import Union
from fastapi.exceptions import HTTPException
from api_events.events.events.events_services import ServicesEvents
from databases import (
Events,
Employees,
@@ -36,100 +37,40 @@ class EventsListEventMethods(MethodToEvent):
list_options: ListOptions,
token_dict: Union[EmployeeTokenObject, OccupantTokenObject],
):
list_options.page = 1
list_options.size = 10000
Events.filter_attr = list_options
records = Events.filter_all()
if isinstance(token_dict, OccupantTokenObject):
occupant_events = Event2Occupant.filter_all(
Event2Occupant.build_living_space_id == token_dict.selected_occupant.living_space_id
).data
records = Events.filter_all(
Events.id.in_([event.event_id for event in occupant_events])
)
return AlchemyJsonResponse(
completed=True,
message="DecisionBook are listed successfully",
result=records,
)
elif isinstance(token_dict, EmployeeTokenObject):
employee_events = Event2Employee.filter_all(
Event2Employee.employee_id == token_dict.selected_company.employee_id
).data
records = Events.filter_all(
Events.id.in_([event.event_id for event in employee_events])
)
return AlchemyJsonResponse(
completed=True,
message="DecisionBook are listed successfully",
result=records,
)
return AlchemyJsonResponse(
completed=True,
message="DecisionBook are listed successfully",
result=records,
completed=False,
message="DecisionBook are NOT listed successfully",
result=[],
)
class EventsCreateEventMethods(MethodToEvent):
event_type = "CREATE"
__event_keys__ = {
"514a9f8f-e5e5-4e10-9d0b-2de8f461fc1b": "events_create",
}
@classmethod
def events_create(cls, data: CreateEvents, token_dict):
event = Events.find_or_create(
**token_dict.user_creds,
event_name=data.event_name,
event_description=data.event_description,
event_date=data.event_date,
event_location=data.event_location,
active=True,
deleted=False,
)
Events.save()
return {
"status": "success",
"message": "Event created successfully.",
"event": event.uu_id,
}
class EventsUpdateEventMethods(MethodToEvent):
event_type = "UPDATE"
__event_keys__ = {
"f94e7b79-2369-4840-bf2b-244934ca3136": "events_update",
}
@classmethod
def events_update(cls, data: CreateEvents, token_dict):
event = Events.filter_by_one(uu_id=data.uu_id, **Events.valid_record_dict).data
if not event:
raise HTTPException(
status_code=404,
detail="No event found. Please contact your responsible company.",
)
event.update(
**token_dict.user_creds,
event_name=data.event_name,
event_description=data.event_description,
event_date=data.event_date,
event_location=data.event_location,
)
Events.save()
return {
"status": "success",
"message": "Event updated successfully.",
"event": event.uu_id,
}
class EventsPatchEventMethods(MethodToEvent):
event_type = "PATCH"
__event_keys__ = {
"41944c63-22d3-4866-affd-34bcd49da58b": "events_patch",
}
@classmethod
def events_patch(cls, data: CreateEvents, token_dict):
event = Events.filter_by_one(uu_id=data.uu_id, **Events.valid_record_dict).data
if not event:
raise HTTPException(
status_code=404,
detail="No event found. Please contact your responsible company.",
)
event.update(
**token_dict.user_creds,
event_name=data.event_name,
event_description=data.event_description,
event_date=data.event_date,
event_location=data.event_location,
)
return {
"status": "success",
"message": "Event patched successfully.",
"event": event.uu_id,
}
class EventsBindEventToOccupantMethods(MethodToEvent):
event_type = "UPDATE"
@@ -218,21 +159,105 @@ class EventsBindEventToEmployeeMethods(MethodToEvent):
}
EventsListEventMethod = EventsListEventMethods(
action=ActionsSchema(endpoint="/event/list")
)
EventsCreateEventMethod = EventsCreateEventMethods(
action=ActionsSchema(endpoint="/event/create")
)
EventsUpdateEventMethod = EventsUpdateEventMethods(
action=ActionsSchema(endpoint="/event/update")
)
EventsPatchEventMethod = EventsPatchEventMethods(
action=ActionsSchema(endpoint="/event/patch")
)
EventsBindEventToOccupantMethod = EventsBindEventToOccupantMethods(
action=ActionsSchema(endpoint="/bind/events/occupant")
)
EventsBindEventToEmployeeMethod = EventsBindEventToEmployeeMethods(
action=ActionsSchema(endpoint="/bind/events/employee")
)
EventsListEventMethod = EventsListEventMethods(
action=ActionsSchema(endpoint="/event/list")
)
# EventsCreateEventMethod = EventsCreateEventMethods(
# action=ActionsSchema(endpoint="/event/create")
# )
# EventsUpdateEventMethod = EventsUpdateEventMethods(
# action=ActionsSchema(endpoint="/event/update")
# )
# EventsPatchEventMethod = EventsPatchEventMethods(
# action=ActionsSchema(endpoint="/event/patch")
# )
#
# class EventsCreateEventMethods(MethodToEvent):
#
# event_type = "CREATE"
# __event_keys__ = {
# "514a9f8f-e5e5-4e10-9d0b-2de8f461fc1b": "events_create",
# }
#
# @classmethod
# def events_create(cls, data: CreateEvents, token_dict):
# event = Events.find_or_create(
# **token_dict.user_creds,
# event_name=data.event_name,
# event_description=data.event_description,
# event_date=data.event_date,
# event_location=data.event_location,
# active=True,
# deleted=False,
# )
# Events.save()
# return {
# "status": "success",
# "message": "Event created successfully.",
# "event": event.uu_id,
# }
# class EventsUpdateEventMethods(MethodToEvent):
#
# event_type = "UPDATE"
# __event_keys__ = {
# "f94e7b79-2369-4840-bf2b-244934ca3136": "events_update",
# }
#
# @classmethod
# def events_update(cls, data: CreateEvents, token_dict):
# event = Events.filter_by_one(uu_id=data.uu_id, **Events.valid_record_dict).data
# if not event:
# raise HTTPException(
# status_code=404,
# detail="No event found. Please contact your responsible company.",
# )
# event.update(
# **token_dict.user_creds,
# event_name=data.event_name,
# event_description=data.event_description,
# event_date=data.event_date,
# event_location=data.event_location,
# )
# Events.save()
# return {
# "status": "success",
# "message": "Event updated successfully.",
# "event": event.uu_id,
# }
#
#
# class EventsPatchEventMethods(MethodToEvent):
#
# event_type = "PATCH"
# __event_keys__ = {
# "41944c63-22d3-4866-affd-34bcd49da58b": "events_patch",
# }
#
# @classmethod
# def events_patch(cls, data: CreateEvents, token_dict):
# event = Events.filter_by_one(uu_id=data.uu_id, **Events.valid_record_dict).data
# if not event:
# raise HTTPException(
# status_code=404,
# detail="No event found. Please contact your responsible company.",
# )
# event.update(
# **token_dict.user_creds,
# event_name=data.event_name,
# event_description=data.event_description,
# event_date=data.event_date,
# event_location=data.event_location,
# )
# return {
# "status": "success",
# "message": "Event patched successfully.",
# "event": event.uu_id,
# }

3
api_events/events/identity/users.py
View File

@@ -98,8 +98,7 @@ class UserCreateEventMethods(MethodToEvent):
)
if not send_email_completed:
raise created_user.raise_http_exception(
status_code=400,
message="Email can not be sent. Try again later"
status_code=400, message="Email can not be sent. Try again later"
)
return JSONResponse(
content={