users/mutations.py at master · Clubs-Council-IIITH/users · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
"""
Mutations for Users Microservice
"""

import os

import strawberry
from fastapi.encoders import jsonable_encoder

from db import db
from models import User

# import all models and types
from otypes import Info, PhoneInput, RoleInput, UserDataInput

inter_communication_secret = os.getenv("INTER_COMMUNICATION_SECRET")


# update role of user with uid
@strawberry.mutation
async def updateRole(roleInput: RoleInput, info: Info) -> bool:
    """
    This method is used to update the role of a user by CC.

    Args:
        roleInput (otypes.RoleInput): Contains the uid and role of the user.
        info (otypes.Info): Contains the user details.

    Returns:
        (bool): True if the role is updated successfully, False otherwise.

    Raises:
        Exception: Not logged in!
        Exception: Authentication Error! Only admins can assign roles!
        Exception: Authentication Error! Invalid secret!
    """

    user = info.context.user
    if not user:
        raise Exception("Not logged in!")

    roleInputData = jsonable_encoder(roleInput)

    # check if user is admin
    if user.get("role", None) not in ["cc"]:
        raise Exception("Authentication Error! Only admins can assign roles!")

    # check if the secret is correct
    if (
        roleInputData.get("inter_communication_secret", None)
        != inter_communication_secret
    ):
        raise Exception("Authentication Error! Invalid secret!")

    await db.users.update_one(
        {"uid": roleInputData["uid"]},
        {
            "$set": {"role": roleInputData["role"]},
            "$setOnInsert": {"uid": roleInputData["uid"]},
        },
        upsert=True,
    )

    return True


@strawberry.mutation
async def updateUserPhone(userDataInput: PhoneInput, info: Info) -> bool:
    """
    This method is used to update the phone number of a user by the cc and
    user.

    Args:
        userDataInput (otypes.PhoneInput): Contains the uid and
                                         phone number of the user.
        info (otypes.Info): Contains the user details.

    Returns:
        (bool): True if the phone number is updated successfully,
            False otherwise.

    Raises:
        Exception: Not logged in!
        Exception: Invalid phone number!
        Exception: You are not allowed to perform this action!
    """

    user = info.context.user
    if not user:
        raise Exception("Not logged in!")

    userData = jsonable_encoder(userDataInput)

    # Validate the data by putting in the model
    try:
        User(**userData)
    except Exception:
        raise Exception("Invalid phone number!")

    # check if user has access
    if not (
        user.get("role", None) in ["cc", "club"]
        or user.get("uid", None) == userData["uid"]
    ):
        raise Exception("You are not allowed to perform this action!")

    await db.users.update_one(
        {"uid": userData["uid"]},
        {"$set": {"phone": userData["phone"]}},
    )

    return True


@strawberry.mutation
async def updateUserData(userDataInput: UserDataInput, info: Info) -> bool:
    """
    Used to update the data of a user by CC and the User

    Args:
        userDataInput (otypes.UserDataInput): Contains the uid, image and phone
                                       number of the user.
        info (otypes.Info): Contains the user details.

    Returns:
        (bool): True if the data is updated successfully, False otherwise.

    Raises:
        Exception: Not logged in!
        Exception: You are not allowed to perform this action!
    """

    user = info.context.user
    if not user:
        raise Exception("Not logged in!")

    userData = jsonable_encoder(userDataInput)

    # check if user has access
    if (
        user.get("role", None) not in ["cc"]
        and user.get("uid", None) != userData["uid"]
    ):
        raise Exception("You are not allowed to perform this action!")

    # Validate the data by putting in the model
    User(**userData)

    await db.users.update_one(
        {"uid": userData["uid"]},
        {"$set": {"img": userData["img"], "phone": userData["phone"]}},
    )

    return True


# register all mutations
mutations = [
    updateRole,
    updateUserPhone,
    updateUserData,
]