UniqAI/open-webui
1
0
Fork 0
mirror of https://github.com/open-webui/open-webui synced 2025-03-24 22:49:22 +00:00
Code Issues Actions 4 Packages Projects Releases Wiki Activity

Update auths.py

Browse Source
This commit is contained in:
Timothy J. Baek 2024-07-04 00:25:45 -07:00
parent 8fe2a7bb75
commit 8b13755d56
1 changed files with 51 additions and 38 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

89
backend/apps/webui/models/auths.py
View File

@ -102,40 +102,44 @@ class AuthsTable:
role: str = "pending",
oauth_sub: Optional[str] = None,
) -> Optional[UserModel]:
log.info("insert_new_auth")
with get_db() as db:
id = str(uuid.uuid4())
log.info("insert_new_auth")
auth = AuthModel(
**{"id": id, "email": email, "password": password, "active": True}
)
result = Auth(**auth.model_dump())
db.add(result)
id = str(uuid.uuid4())
user = Users.insert_new_user(
id, name, email, profile_image_url, role, oauth_sub
)
auth = AuthModel(
**{"id": id, "email": email, "password": password, "active": True}
)
result = Auth(**auth.model_dump())
db.add(result)
db.commit()
db.refresh(result)
user = Users.insert_new_user(
id, name, email, profile_image_url, role, oauth_sub
)
if result and user:
return user
else:
return None
db.commit()
db.refresh(result)
if result and user:
return user
else:
return None
def authenticate_user(self, email: str, password: str) -> Optional[UserModel]:
log.info(f"authenticate_user: {email}")
try:
auth = db.query(Auth).filter_by(email=email, active=True).first()
if auth:
if verify_password(password, auth.password):
user = Users.get_user_by_id(auth.id)
return user
with get_db() as db:
auth = db.query(Auth).filter_by(email=email, active=True).first()
if auth:
if verify_password(password, auth.password):
user = Users.get_user_by_id(auth.id)
return user
else:
return None
else:
return None
else:
return None
except:
return None
@ -154,38 +158,47 @@ class AuthsTable:
def authenticate_user_by_trusted_header(self, email: str) -> Optional[UserModel]:
log.info(f"authenticate_user_by_trusted_header: {email}")
try:
auth = db.query(Auth).filter(email=email, active=True).first()
if auth:
user = Users.get_user_by_id(auth.id)
return user
with get_db() as db:
auth = db.query(Auth).filter(email=email, active=True).first()
if auth:
user = Users.get_user_by_id(auth.id)
return user
except:
return None
def update_user_password_by_id(self, id: str, new_password: str) -> bool:
try:
result = db.query(Auth).filter_by(id=id).update({"password": new_password})
return True if result == 1 else False
with get_db() as db:
result = (
db.query(Auth).filter_by(id=id).update({"password": new_password})
)
return True if result == 1 else False
except:
return False
def update_email_by_id(self, id: str, email: str) -> bool:
try:
result = db.query(Auth).filter_by(id=id).update({"email": email})
return True if result == 1 else False
with get_db() as db:
result = db.query(Auth).filter_by(id=id).update({"email": email})
return True if result == 1 else False
except:
return False
def delete_auth_by_id(self, id: str) -> bool:
try:
# Delete User
result = Users.delete_user_by_id(id)
with get_db() as db:
if result:
db.query(Auth).filter_by(id=id).delete()
# Delete User
result = Users.delete_user_by_id(id)
return True
else:
return False
if result:
db.query(Auth).filter_by(id=id).delete()
return True
else:
return False
except:
return False