81 lines
2.3 KiB
Python
81 lines
2.3 KiB
Python
# Imports
|
|
from contextlib import asynccontextmanager
|
|
from fastapi import FastAPI, HTTPException
|
|
from functions import *
|
|
from sql import *
|
|
|
|
import random
|
|
|
|
# Create DB on startup
|
|
# noinspection PyUnusedLocal
|
|
@asynccontextmanager
|
|
async def lifespan(app: FastAPI):
|
|
create_db_and_tables()
|
|
yield # Code before the yield will run on startup, code after yield won't run until the program is over
|
|
|
|
# Initialize the FastAPI App
|
|
app = FastAPI(lifespan=lifespan)
|
|
|
|
@app.get("/")
|
|
def hello_world():
|
|
return {"message": "Hello World!"}
|
|
|
|
@app.get("/hash/{password}")
|
|
def hash(password: str):
|
|
salt = random.randint(00000, 99999)
|
|
hashed = hash_password(password, salt)
|
|
return {"hash": hashed, "salt": salt}
|
|
|
|
@app.get("/verify/{password}/{hash}/{salt}")
|
|
def verify(password: str, hash: str, salt: int):
|
|
if salt != 0:
|
|
hashed = hash_password(password, salt)
|
|
else:
|
|
hashed = hash_password(password, salt)
|
|
|
|
if hashed == hash:
|
|
return {"message": "Password is correct", "correct": True}
|
|
|
|
return {"message": "Password is incorrect", "correct": False}
|
|
|
|
@app.get("/hash/no-salt/{password}")
|
|
def no_salt(password: str):
|
|
hashed = hash_password(password)
|
|
return {"hash": hashed, "salt": 0}
|
|
|
|
@app.post('/users/create')
|
|
async def create_user(user: User, session: SessionDep) -> User | dict[str, str]:
|
|
get_user = get_user_by_email(user.email, session)
|
|
if get_user is None:
|
|
user.password = hash_password(user.password, salt=random.randint(00000, 99999))
|
|
session.add(user)
|
|
session.commit()
|
|
session.refresh(user)
|
|
return user
|
|
|
|
return {"message": "User already created"}
|
|
|
|
@app.get("/users/{type}")
|
|
async def get_user(type: str, session: SessionDep) -> User:
|
|
user = get_user_by_id(type, session)
|
|
if user is None:
|
|
user = get_user_by_email(type, session)
|
|
|
|
if not user:
|
|
raise HTTPException(status_code=404, detail="User not found")
|
|
|
|
return user
|
|
|
|
@app.delete("/users/{type}")
|
|
async def delete_user(type: str, session: SessionDep) -> User | dict[str, str | bool]:
|
|
user = get_user_by_id(type, session)
|
|
if user is None:
|
|
user = get_user_by_email(type, session)
|
|
|
|
if not user:
|
|
raise HTTPException(status_code=404, detail="User not found")
|
|
|
|
session.delete(user)
|
|
session.commit()
|
|
return {"message": "User deleted", "completed": True}
|