74 lines
2.0 KiB
Python
74 lines
2.0 KiB
Python
from datetime import datetime
|
|
from pydoc import locate
|
|
|
|
from app import app
|
|
from app.schemas import *
|
|
from database import database
|
|
|
|
|
|
@app.on_event("startup")
|
|
async def startup():
|
|
await database.connect()
|
|
|
|
|
|
@app.on_event("shutdown")
|
|
async def shutdown():
|
|
await database.disconnect()
|
|
|
|
|
|
async def instantiate_model(schema, data):
|
|
model = locate("database.models." + schema)
|
|
instance = model(**data)
|
|
return instance
|
|
|
|
|
|
async def insert_data(schema, data):
|
|
instance = instantiate_model(schema, data)
|
|
query = instance.insert()
|
|
await database.execute(query=query, values=data)
|
|
|
|
|
|
# FIXME instance.id has to be replaced with the table's UID
|
|
async def delete_data(schema, data):
|
|
instance = instantiate_model(schema, data)
|
|
query = instance.delete().where(instance.id == data.id)
|
|
await database.execute(query=query)
|
|
|
|
|
|
async def fetch_user(data):
|
|
instance = instantiate_model(schema="Users", data=data)
|
|
query = instance.select().where(instance.email == data.email)
|
|
return await database.fetch_one(query=query)
|
|
|
|
|
|
async def fetch_otp(data: OTPSchema):
|
|
instance = instantiate_model(schema="Users", data=data)
|
|
query = instance.select().where(instance.mobile == data.mobile)
|
|
return await database.fetch_one(query=query)
|
|
|
|
|
|
async def save_otp(data: OTPSchema):
|
|
instance = instantiate_model(schema="Users", data=data)
|
|
query = instance.update().where(instance.mobile == data.mobile).values(otp=data.otp)
|
|
await database.execute(query=query)
|
|
|
|
|
|
async def activate_account(data: OTPSchema):
|
|
timestamp = datetime.now()
|
|
instance = instantiate_model(schema="Users", data=data)
|
|
query = (
|
|
instance.update()
|
|
.where(instance.mobile == data.mobile)
|
|
.values(mobile=instance.data, otp_valid_time=timestamp, status=1)
|
|
)
|
|
await database.execute(query=query)
|
|
|
|
|
|
async def verify_otp(data: OTPSchema):
|
|
user = fetch_otp(data=data)
|
|
stored.otp = user.otp
|
|
if stored_otp == otp:
|
|
activate_account(mobile=mobile)
|
|
return True
|
|
return False
|