open-webui/backend/apps/web/models/auths.py

151 lines
3.1 KiB
Python
Raw Normal View History

2023-11-19 01:47:12 +01:00
from pydantic import BaseModel
from typing import List, Union, Optional
import time
import uuid
2023-12-26 06:44:28 +01:00
from peewee import *
2023-11-19 01:47:12 +01:00
from apps.web.models.users import UserModel, Users
2023-11-19 06:41:43 +01:00
from utils.utils import (
2023-11-19 01:47:12 +01:00
verify_password,
get_password_hash,
bearer_scheme,
create_token,
)
2023-12-26 06:44:28 +01:00
from apps.web.internal.db import DB
2023-11-19 01:47:12 +01:00
####################
# DB MODEL
####################
2023-12-26 06:44:28 +01:00
class Auth(Model):
id = CharField(unique=True)
email = CharField()
password = CharField()
active = BooleanField()
class Meta:
database = DB
2023-11-19 01:47:12 +01:00
class AuthModel(BaseModel):
id: str
email: str
password: str
active: bool = True
####################
# Forms
####################
class Token(BaseModel):
token: str
token_type: str
class UserResponse(BaseModel):
id: str
email: str
name: str
role: str
2023-11-19 06:41:43 +01:00
profile_image_url: str
2023-11-19 01:47:12 +01:00
class SigninResponse(Token, UserResponse):
pass
class SigninForm(BaseModel):
email: str
password: str
2023-12-29 09:12:30 +01:00
class UpdatePasswordForm(BaseModel):
password: str
new_password: str
2023-11-19 01:47:12 +01:00
class SignupForm(BaseModel):
name: str
email: str
password: str
class AuthsTable:
2023-11-19 01:47:12 +01:00
def __init__(self, db):
self.db = db
2023-12-26 06:44:28 +01:00
self.db.create_tables([Auth])
2023-11-19 01:47:12 +01:00
def insert_new_auth(self,
email: str,
password: str,
name: str,
role: str = "pending") -> Optional[UserModel]:
2023-11-19 01:47:12 +01:00
print("insert_new_auth")
id = str(uuid.uuid4())
auth = AuthModel(**{
"id": id,
"email": email,
"password": password,
"active": True
})
2023-12-26 06:44:28 +01:00
result = Auth.create(**auth.model_dump())
2023-11-19 01:47:12 +01:00
user = Users.insert_new_user(id, name, email, role)
if result and user:
return user
else:
return None
def authenticate_user(self, email: str,
password: str) -> Optional[UserModel]:
2023-12-26 06:44:28 +01:00
print("authenticate_user", email)
2023-12-26 08:43:21 +01:00
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
2023-11-19 01:47:12 +01:00
else:
return None
2023-12-26 08:43:21 +01:00
except:
2023-11-19 01:47:12 +01:00
return None
2023-12-29 09:29:18 +01:00
def update_user_password_by_id(self, id: str, new_password: str) -> bool:
2023-12-29 09:12:30 +01:00
try:
2023-12-29 09:29:18 +01:00
query = Auth.update(password=new_password).where(Auth.id == id)
result = query.execute()
2023-12-29 09:31:23 +01:00
return True if result == 1 else False
2023-12-29 09:12:30 +01:00
except:
return False
def delete_auth_by_id(self, id: str) -> bool:
2023-12-29 08:24:51 +01:00
try:
# Delete User
result = Users.delete_user_by_id(id)
if result:
# Delete Auth
query = Auth.delete().where(Auth.id == id)
query.execute(
) # Remove the rows, return number of rows removed.
2023-12-29 08:24:51 +01:00
return True
else:
return False
except:
return False
2023-11-19 01:47:12 +01:00
Auths = AuthsTable(DB)