81 lines
2.0 KiB
Python
81 lines
2.0 KiB
Python
from datetime import datetime
|
|
from fastapi import HTTPException
|
|
from passlib.context import CryptContext
|
|
|
|
from app.schemas import *
|
|
from database import SessionLocal
|
|
from database.models import *
|
|
|
|
|
|
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
|
|
|
|
|
|
def get_db():
|
|
db = SessionLocal()
|
|
try:
|
|
yield db
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
def instantiate_model(model, data):
|
|
table = eval(model)
|
|
instance = table(**data.dict())
|
|
return instance
|
|
|
|
|
|
def insert_data(model, data, db):
|
|
item = instantiate_model(model=model, data=data)
|
|
db.add(item)
|
|
db.commit()
|
|
db.refresh(item)
|
|
return item
|
|
|
|
|
|
# FIXME db.id has to be replaced with the table's UID
|
|
def delete_data(model, data, db):
|
|
item = instantiate_model(model=model, data=data)
|
|
result = db.query(item).filter(item.email == data.email).delete()
|
|
return result
|
|
|
|
|
|
def fetch_user_by_key(data, db):
|
|
return db.query(Users).filter(Users.access_key == data.access_key).first()
|
|
|
|
|
|
def fetch_user_by_email(data, db):
|
|
return db.query(Users).filter(Users.email == data.email).first()
|
|
|
|
|
|
def create_user(data, db):
|
|
data.password = pwd_context.hash(data.password)
|
|
user = insert_data(model="Users", data=data, db=db)
|
|
return user
|
|
|
|
|
|
def update_otp(data: OTPResend, db):
|
|
db.query(Users).filter(Users.email == data.email).update(
|
|
{Users.otp: data.otp, Users.otp_valid_time: data.otp_valid_time}
|
|
)
|
|
db.commit()
|
|
|
|
|
|
def activate_account(data: OTPVerify, db):
|
|
db.query(Users).filter(Users.access_key == data.access_key).update(
|
|
{Users.status: 1}
|
|
)
|
|
user = fetch_user_by_key(data=data, db=db)
|
|
return user
|
|
|
|
|
|
def verify_otp(data: OTPVerify, db):
|
|
user = fetch_user_by_key(data=data, db=db)
|
|
matching_otp = user.otp == data.otp
|
|
valid_time = datetime.now() < user.otp_valid_time
|
|
valid_otp = matching_otp and valid_time
|
|
if valid_otp:
|
|
result = activate_account(data=data, db=db)
|
|
return result
|
|
else:
|
|
raise HTTPException(status_code=400, detail="The OTP is not correct")
|