Skip to content

auth_routes

routes.auth_routes

AuthLogIn

Bases: Resource

Class responsible for user log in

Source code in yaptide/routes/auth_routes.py 
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
class AuthLogIn(Resource):
    """Class responsible for user log in"""

    @staticmethod
    def post():
        """Method returning status of logging in (and token if it was successful)"""
        payload_dict: dict = request.get_json(force=True)
        if not payload_dict:
            return yaptide_response(message="No JSON in body", code=400)

        required_keys = {"username", "password"}

        if required_keys != required_keys.intersection(set(payload_dict.keys())):
            diff = required_keys.difference(set(payload_dict.keys()))
            return yaptide_response(message=f"Missing keys in JSON payload: {diff}", code=400)

        try:
            user: YaptideUserModel = fetch_yaptide_user_by_username(username=payload_dict['username'])
            if not user:
                return yaptide_response(message='Invalid login or password', code=401)

            if not user.check_password(password=payload_dict['password']):
                return yaptide_response(message='Invalid login or password', code=401)

            access_token, access_exp = encode_auth_token(user_id=user.id, is_refresh=False)
            refresh_token, refresh_exp = encode_auth_token(user_id=user.id, is_refresh=True)

            resp = yaptide_response(
                message='Successfully logged in',
                code=202,
                content={
                    'access_exp': int(access_exp.timestamp()*1000),
                    'refresh_exp': int(refresh_exp.timestamp()*1000),
                }
            )
            resp.set_cookie('access_token', access_token, httponly=True, samesite='Lax', expires=access_exp)
            resp.set_cookie('refresh_token', refresh_token, httponly=True, samesite='Lax', expires=refresh_exp)
            return resp
        except Exception as e:  # skipcq: PYL-W0703
            logging.error("%s", e)
            return error_internal_response()

post staticmethod 

post()

Method returning status of logging in (and token if it was successful)

Source code in yaptide/routes/auth_routes.py 
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
@staticmethod
def post():
    """Method returning status of logging in (and token if it was successful)"""
    payload_dict: dict = request.get_json(force=True)
    if not payload_dict:
        return yaptide_response(message="No JSON in body", code=400)

    required_keys = {"username", "password"}

    if required_keys != required_keys.intersection(set(payload_dict.keys())):
        diff = required_keys.difference(set(payload_dict.keys()))
        return yaptide_response(message=f"Missing keys in JSON payload: {diff}", code=400)

    try:
        user: YaptideUserModel = fetch_yaptide_user_by_username(username=payload_dict['username'])
        if not user:
            return yaptide_response(message='Invalid login or password', code=401)

        if not user.check_password(password=payload_dict['password']):
            return yaptide_response(message='Invalid login or password', code=401)

        access_token, access_exp = encode_auth_token(user_id=user.id, is_refresh=False)
        refresh_token, refresh_exp = encode_auth_token(user_id=user.id, is_refresh=True)

        resp = yaptide_response(
            message='Successfully logged in',
            code=202,
            content={
                'access_exp': int(access_exp.timestamp()*1000),
                'refresh_exp': int(refresh_exp.timestamp()*1000),
            }
        )
        resp.set_cookie('access_token', access_token, httponly=True, samesite='Lax', expires=access_exp)
        resp.set_cookie('refresh_token', refresh_token, httponly=True, samesite='Lax', expires=refresh_exp)
        return resp
    except Exception as e:  # skipcq: PYL-W0703
        logging.error("%s", e)
        return error_internal_response()

AuthLogOut

Bases: Resource

Class responsible for user log out

Source code in yaptide/routes/auth_routes.py 
125
126
127
128
129
130
131
132
133
134
class AuthLogOut(Resource):
    """Class responsible for user log out"""

    @staticmethod
    def delete():
        """Method logging the user out"""
        resp = yaptide_response(message='User logged out', code=200)
        resp.delete_cookie('access_token')
        resp.delete_cookie('refresh_token')
        return resp

delete staticmethod 

delete()

Method logging the user out

Source code in yaptide/routes/auth_routes.py 
128
129
130
131
132
133
134
@staticmethod
def delete():
    """Method logging the user out"""
    resp = yaptide_response(message='User logged out', code=200)
    resp.delete_cookie('access_token')
    resp.delete_cookie('refresh_token')
    return resp

AuthRefresh

Bases: Resource

Class responsible for refreshing user

Source code in yaptide/routes/auth_routes.py 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
class AuthRefresh(Resource):
    """Class responsible for refreshing user"""

    @staticmethod
    @requires_auth(is_refresh=True)
    def get(user: YaptideUserModel):
        """Method refreshing token"""
        access_token, access_exp = encode_auth_token(user_id=user.id, is_refresh=False)
        resp = yaptide_response(
            message='User refreshed',
            code=200,
            content={'access_exp': int(access_exp.timestamp()*1000)}
        )
        resp.set_cookie('access_token', access_token, httponly=True, samesite='Lax', expires=access_exp)
        return resp

get staticmethod 

get(user)

Method refreshing token

Source code in yaptide/routes/auth_routes.py 
 97
 98
 99
100
101
102
103
104
105
106
107
108
@staticmethod
@requires_auth(is_refresh=True)
def get(user: YaptideUserModel):
    """Method refreshing token"""
    access_token, access_exp = encode_auth_token(user_id=user.id, is_refresh=False)
    resp = yaptide_response(
        message='User refreshed',
        code=200,
        content={'access_exp': int(access_exp.timestamp()*1000)}
    )
    resp.set_cookie('access_token', access_token, httponly=True, samesite='Lax', expires=access_exp)
    return resp

AuthRegister

Bases: Resource

Class responsible for user registration

Source code in yaptide/routes/auth_routes.py 
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
class AuthRegister(Resource):
    """Class responsible for user registration"""

    class APIParametersSchema(Schema):
        """Class specifies API parameters"""

        username = fields.String()
        password = fields.String()

    @staticmethod
    def put():
        """Method returning status of registration"""
        try:
            json_data: dict = AuthRegister.APIParametersSchema().load(request.get_json(force=True))
        except ValidationError:
            return error_validation_response()

        user = fetch_yaptide_user_by_username(username=json_data.get('username'))

        if not user:
            try:
                user = YaptideUserModel(username=json_data.get('username'))
                user.set_password(json_data.get('password'))

                add_object_to_db(user)

                return yaptide_response(message='User created', code=201)

            except Exception as e:  # skipcq: PYL-W0703
                logging.error("%s", e)
                return error_internal_response()
        else:
            return yaptide_response(message='User existing', code=403)

APIParametersSchema

Bases: Schema

Class specifies API parameters

Source code in yaptide/routes/auth_routes.py 
19
20
21
22
23
class APIParametersSchema(Schema):
    """Class specifies API parameters"""

    username = fields.String()
    password = fields.String()
password class-attribute instance-attribute 
password = String()
username class-attribute instance-attribute 
username = String()

put staticmethod 

put()

Method returning status of registration

Source code in yaptide/routes/auth_routes.py 
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
@staticmethod
def put():
    """Method returning status of registration"""
    try:
        json_data: dict = AuthRegister.APIParametersSchema().load(request.get_json(force=True))
    except ValidationError:
        return error_validation_response()

    user = fetch_yaptide_user_by_username(username=json_data.get('username'))

    if not user:
        try:
            user = YaptideUserModel(username=json_data.get('username'))
            user.set_password(json_data.get('password'))

            add_object_to_db(user)

            return yaptide_response(message='User created', code=201)

        except Exception as e:  # skipcq: PYL-W0703
            logging.error("%s", e)
            return error_internal_response()
    else:
        return yaptide_response(message='User existing', code=403)

AuthStatus

Bases: Resource

Class responsible for returning user status

Source code in yaptide/routes/auth_routes.py 
111
112
113
114
115
116
117
118
119
120
121
122
class AuthStatus(Resource):
    """Class responsible for returning user status"""

    @staticmethod
    @requires_auth()
    def get(user: YaptideUserModel):
        """Method returning user's status"""
        return yaptide_response(
            message='User status',
            code=200,
            content={'username': user.username}
        )

get staticmethod 

get(user)

Method returning user's status

Source code in yaptide/routes/auth_routes.py 
114
115
116
117
118
119
120
121
122
@staticmethod
@requires_auth()
def get(user: YaptideUserModel):
    """Method returning user's status"""
    return yaptide_response(
        message='User status',
        code=200,
        content={'username': user.username}
    )