wag-managment-api-service-version-2/api_events/events/identity/users.py

import typing

from fastapi import status
from fastapi.responses import JSONResponse

from api_configs import ApiStatic
from databases import MongoQueryIdentity, Users, Companies, People
from databases.no_sql_models.validations import DomainViaUser

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_services.email.service import send_email
from api_services.templates.password_templates import change_your_password_template
from api_validations.validations_request import (
    InsertUsers,
    UpdateUsers,
    PatchRecord,
    ListOptions,
    RegisterServices2Occupant,
)


class UserListEventMethods(MethodToEvent):

    event_type = "SELECT"
    __event_keys__ = {
        "1483a8a2-d244-4593-b9f8-f1b4bcbefcd5": "user_list",
    }

    @classmethod
    def user_list(
        cls,
        list_options: ListOptions,
        token_dict: typing.Union[EmployeeTokenObject, OccupantTokenObject],
    ):
        if isinstance(token_dict, OccupantTokenObject):
            raise Users.raise_http_exception(
                status_code="HTTP_403_FORBIDDEN",
                message="Occupant object can not list users",
                error_case="NOTAUTHORIZED",
                data={},
            )
        if "user_uu_id_list" in list_options.query:
            people_ids = list_options.query.pop("user_uu_id_list")
            people_id_list = (
                user.person_id
                for user in Users.filter_all(Users.uu_id.in_(people_ids)).data
            )
            Users.filter_attr = list_options
            records = Users.filter_all(
                Users.person_id.in_(people_id_list),
            )
            return AlchemyJsonResponse(
                completed=True,
                message="Users are listed successfully",
                result=records,
            )
        Users.filter_attr = list_options
        records = Users.filter_all()
        return AlchemyJsonResponse(
            completed=True,
            message="Users are listed successfully",
            result=records,
        )


class UserCreateEventMethods(MethodToEvent):

    event_type = "CREATE"
    __event_keys__ = {
        "8eb50c24-4bdc-4309-9836-f7048daee409": "user_create",
    }

    @classmethod
    def user_create(
        cls,
        data: InsertUsers,
        token_dict: typing.Union[EmployeeTokenObject, OccupantTokenObject],
    ):
        created_user = Users.create_action(create_user=data, token_dict=token_dict)
        domain_via_user = DomainViaUser(
            **{"user_uu_id": str(created_user.uu_id), "main_domain": "evyos.com.tr"}
        )
        created_user.save_and_confirm()
        mongo_query_identity = MongoQueryIdentity(
            company_uuid=created_user.related_company,
        )
        mongo_query_identity.create_domain_via_user(payload=domain_via_user)
        reset_password_token = created_user.reset_password_token(found_user=created_user)
        send_email_completed = send_email(
            subject=f"Dear {created_user.user_tag}, your password has been changed.",
            receivers=[str(created_user.email)],
            html=change_your_password_template(
                user_name=created_user.user_tag,
                forgot_link=ApiStatic.forgot_link(forgot_key=reset_password_token),
            ),
        )
        if not send_email_completed:
            raise created_user.raise_http_exception(
                status_code=400, message="Email can not be sent. Try again later"
            )
        return JSONResponse(
            content={
                "completed": True,
                "message": "Create User record",
                "data": created_user.get_dict(),
                "password_token": {
                    "password_token": created_user.password_token,
                    "password_expires_day": str(created_user.password_expires_day),
                    "password_expiry_begins": str(created_user.password_expiry_begins),
                    "hash_password": created_user.hash_password,
                    "related_company": created_user.related_company,
                },
            },
            status_code=status.HTTP_200_OK,
        )


class UserUpdateEventMethods(MethodToEvent):

    event_type = "UPDATE"
    __event_keys__ = {
        "d08a9470-1eb0-4890-a9e8-b6686239d7e9": "user_update",
    }

    @classmethod
    def user_update(
        cls,
        data: UpdateUsers,
        user_uu_id: str,
        token_dict: typing.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", 5)],
            filter_expr=[Companies.id == token_dict.get("")],
        )
        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,
            )
        return JSONResponse(
            content={"completed": True, "message": "Update User record", "data": {}},
            status_code=status.HTTP_200_OK,
        )


class UserPatchEventMethods(MethodToEvent):

    event_type = "PATCH"
    __event_keys__ = {
        "d26a1a3c-eaeb-4d01-b35b-a5ed714e29c0": "user_patch",
    }

    @classmethod
    def user_patch(cls, data: PatchRecord, user_uu_id: str, token_dict):
        find_one_user = Users.filter_one(
            Users.uu_id == user_uu_id,
        ).data
        access_authorized_company = Companies.select_action(
            duty_id_list=[getattr(token_dict, "duty_id", 5)],
            filter_expr=[Companies.id == find_one_user.id],
        )
        if access_authorized_company.count:
            action = data.excluded_dump()
            find_one_user.active = bool(action.get("active", find_one_user.active))
            find_one_user.is_confirmed = bool(
                action.get("confirm", find_one_user.is_confirmed)
            )
            find_one_user.deleted = bool(action.get("delete", find_one_user.deleted))
            find_one_user.save()
            return JSONResponse(
                content={
                    "completed": True,
                    "message": "Patch User record completed",
                    "data": find_one_user.get_dict(),
                },
                status_code=status.HTTP_200_OK,
            )
        return JSONResponse(
            content={
                "completed": False,
                "message": "Patch User record failed",
                "data": {},
            },
            status_code=status.HTTP_200_OK,
        )


UserListEventMethod = UserListEventMethods(action=ActionsSchema(endpoint="/user/list"))
UserCreateEventMethod = UserCreateEventMethods(
    action=ActionsSchema(endpoint="/user/create")
)
UserUpdateEventMethod = UserUpdateEventMethods(
    action=ActionsSchema(endpoint="/user/update")
)
UserPatchEventMethod = UserPatchEventMethods(
    action=ActionsSchema(endpoint="/user/patch")
)