auths.py 4.4 KB
Newer Older
1
2
3
4
from pydantic import BaseModel
from typing import List, Union, Optional
import time
import uuid
5
import logging
Timothy J. Baek's avatar
Timothy J. Baek committed
6
from peewee import *
7
8

from apps.web.models.users import UserModel, Users
Tim Farrell's avatar
Tim Farrell committed
9
from utils.utils import verify_password
10

Timothy J. Baek's avatar
Timothy J. Baek committed
11
from apps.web.internal.db import DB
12

13
from config import SRC_LOG_LEVELS
Timothy J. Baek's avatar
Timothy J. Baek committed
14

15
16
17
log = logging.getLogger(__name__)
log.setLevel(SRC_LOG_LEVELS["MODELS"])

18
19
20
21
22
####################
# DB MODEL
####################


Timothy J. Baek's avatar
Timothy J. Baek committed
23
24
25
26
27
28
29
30
31
32
class Auth(Model):
    id = CharField(unique=True)
    email = CharField()
    password = CharField()
    active = BooleanField()

    class Meta:
        database = DB


33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
class AuthModel(BaseModel):
    id: str
    email: str
    password: str
    active: bool = True


####################
# Forms
####################


class Token(BaseModel):
    token: str
    token_type: str

Timothy J. Baek's avatar
fix  
Timothy J. Baek committed
49

liu.vaayne's avatar
liu.vaayne committed
50
51
class ApiKey(BaseModel):
    api_key: Optional[str] = None
52

Timothy J. Baek's avatar
fix  
Timothy J. Baek committed
53

54
55
56
57
58
class UserResponse(BaseModel):
    id: str
    email: str
    name: str
    role: str
Timothy J. Baek's avatar
Timothy J. Baek committed
59
    profile_image_url: str
60
61
62
63
64
65
66
67
68
69
70


class SigninResponse(Token, UserResponse):
    pass


class SigninForm(BaseModel):
    email: str
    password: str


71
72
73
74
class ProfileImageUrlForm(BaseModel):
    profile_image_url: str


75
76
77
78
79
class UpdateProfileForm(BaseModel):
    profile_image_url: str
    name: str


80
81
82
83
84
class UpdatePasswordForm(BaseModel):
    password: str
    new_password: str


85
86
87
88
class SignupForm(BaseModel):
    name: str
    email: str
    password: str
Timothy J. Baek's avatar
refac  
Timothy J. Baek committed
89
    profile_image_url: Optional[str] = "/user.png"
90
91
92
93
94


class AuthsTable:
    def __init__(self, db):
        self.db = db
Timothy J. Baek's avatar
Timothy J. Baek committed
95
        self.db.create_tables([Auth])
96

Timothy J. Baek's avatar
Timothy J. Baek committed
97
    def insert_new_auth(
Danny Liu's avatar
Danny Liu committed
98
99
100
101
        self,
        email: str,
        password: str,
        name: str,
Timothy J. Baek's avatar
refac  
Timothy J. Baek committed
102
        profile_image_url: str = "/user.png",
Danny Liu's avatar
Danny Liu committed
103
        role: str = "pending",
Timothy J. Baek's avatar
Timothy J. Baek committed
104
    ) -> Optional[UserModel]:
105
        log.info("insert_new_auth")
106
107
108

        id = str(uuid.uuid4())

Timothy J. Baek's avatar
Timothy J. Baek committed
109
110
111
        auth = AuthModel(
            **{"id": id, "email": email, "password": password, "active": True}
        )
Timothy J. Baek's avatar
Timothy J. Baek committed
112
113
        result = Auth.create(**auth.model_dump())

114
        user = Users.insert_new_user(id, name, email, profile_image_url, role)
115
116
117
118
119
120

        if result and user:
            return user
        else:
            return None

Timothy J. Baek's avatar
Timothy J. Baek committed
121
    def authenticate_user(self, email: str, password: str) -> Optional[UserModel]:
122
        log.info(f"authenticate_user: {email}")
Timothy J. Baek's avatar
Timothy J. Baek committed
123
124
125
126
127
128
129
130
        try:
            auth = Auth.get(Auth.email == email, Auth.active == True)
            if auth:
                if verify_password(password, auth.password):
                    user = Users.get_user_by_id(auth.id)
                    return user
                else:
                    return None
131
132
            else:
                return None
Timothy J. Baek's avatar
Timothy J. Baek committed
133
        except:
134
135
            return None

liu.vaayne's avatar
liu.vaayne committed
136
137
138
139
140
    def authenticate_user_by_api_key(self, api_key: str) -> Optional[UserModel]:
        log.info(f"authenticate_user_by_api_key: {api_key}")
        # if no api_key, return None
        if not api_key:
            return None
Timothy J. Baek's avatar
fix  
Timothy J. Baek committed
141

liu.vaayne's avatar
liu.vaayne committed
142
        try:
143
144
            user = Users.get_user_by_api_key(api_key)
            return user if user else None
Timothy J. Baek's avatar
fix  
Timothy J. Baek committed
145
146
147
        except:
            return False

Jun Siang Cheah's avatar
Jun Siang Cheah committed
148
    def authenticate_user_by_trusted_header(self, email: str) -> Optional[UserModel]:
149
150
151
152
153
154
        log.info(f"authenticate_user_by_trusted_header: {email}")
        try:
            auth = Auth.get(Auth.email == email, Auth.active == True)
            if auth:
                user = Users.get_user_by_id(auth.id)
                return user
liu.vaayne's avatar
liu.vaayne committed
155
156
157
        except:
            return None

Timothy J. Baek's avatar
Timothy J. Baek committed
158
    def update_user_password_by_id(self, id: str, new_password: str) -> bool:
159
        try:
Timothy J. Baek's avatar
Timothy J. Baek committed
160
161
            query = Auth.update(password=new_password).where(Auth.id == id)
            result = query.execute()
Timothy J. Baek's avatar
Timothy J. Baek committed
162
163

            return True if result == 1 else False
164
165
166
        except:
            return False

Timothy J. Baek's avatar
Timothy J. Baek committed
167
168
169
170
171
172
173
174
175
    def update_email_by_id(self, id: str, email: str) -> bool:
        try:
            query = Auth.update(email=email).where(Auth.id == id)
            result = query.execute()

            return True if result == 1 else False
        except:
            return False

176
    def delete_auth_by_id(self, id: str) -> bool:
Timothy J. Baek's avatar
Timothy J. Baek committed
177
178
179
180
181
182
183
        try:
            # Delete User
            result = Users.delete_user_by_id(id)

            if result:
                # Delete Auth
                query = Auth.delete().where(Auth.id == id)
Timothy J. Baek's avatar
Timothy J. Baek committed
184
                query.execute()  # Remove the rows, return number of rows removed.
Timothy J. Baek's avatar
Timothy J. Baek committed
185
186
187
188
189
190
191

                return True
            else:
                return False
        except:
            return False

192
193

Auths = AuthsTable(DB)