110 lines
4.2 KiB
Python
110 lines
4.2 KiB
Python
from auth import *
|
|
from fastapi import FastAPI
|
|
from fastapi.middleware.cors import CORSMiddleware
|
|
from fastapi.security import OAuth2PasswordRequestForm
|
|
from functions import *
|
|
import random
|
|
from sql import *
|
|
from tables import *
|
|
|
|
# fastapi init
|
|
app = FastAPI(lifespan=lifespan)
|
|
origins = [
|
|
"http://localhost:3000",
|
|
"https://task.fstropii.com",
|
|
"https://www.task.fstropii.com",
|
|
"https://task.fstropii.com/",
|
|
"http://localhost"
|
|
]
|
|
app.add_middleware(
|
|
CORSMiddleware,
|
|
allow_origins=origins,
|
|
allow_credentials=True,
|
|
allow_methods=["*"],
|
|
allow_headers=["*"],
|
|
)
|
|
|
|
# routes
|
|
@app.get("/")
|
|
async def root():
|
|
return {"message": "Hello World"}
|
|
|
|
@app.post("/user/create", response_model=UserPublic)
|
|
def create_user(user: UserCreate, session: SessionDep):
|
|
salt = random.randint(00000, 99999)
|
|
existing_user = session.query(User).filter(User.email == user.email).first()
|
|
if existing_user:
|
|
raise HTTPException(status_code=400, detail="Email already registered")
|
|
|
|
user.password = hash_password(user.password, salt)
|
|
user.salt = salt
|
|
|
|
db_user = User.model_validate(user)
|
|
session.add(db_user)
|
|
session.commit()
|
|
session.refresh(db_user)
|
|
return db_user
|
|
|
|
@app.patch("/user/update", response_model=UserPublic)
|
|
async def update_user(user: UserUpdate, session: SessionDep, current_user: User = Depends(get_current_user)):
|
|
user_db = session.get(User, current_user.id)
|
|
user_data = user.model_dump(exclude_unset=True)
|
|
print(f"l52_db: {user_db.tasks}")
|
|
if 'tasks' in user_data:
|
|
if user_db.tasks is None:
|
|
user_db.tasks = []
|
|
|
|
for i in range(len(user_db.tasks)): # loop through all tasks
|
|
if user_db.tasks[i]['title'] == user_data['tasks'][i]['title']: # check if there is a task with the same title
|
|
# if there is, check if the is_completed status has changed
|
|
# if it has, update the task in the db
|
|
# if it hasn't, do nothing
|
|
if user_db.tasks[i]['is_completed'] != user_data['tasks'][i]['is_completed']:
|
|
# update the task in the db
|
|
updated_task = user_db.tasks[i]
|
|
updated_task['is_completed'] = user_data['tasks'][i]['is_completed']
|
|
break
|
|
else:
|
|
current_tasks = user_db.tasks
|
|
new_task = user_data['tasks']
|
|
user_data.tasks = current_tasks + new_task
|
|
print(f"l71_update_db: {user_db.tasks}")
|
|
print(f"l72_update_user_data: {user_data}")
|
|
user_db.sqlmodel_update(user_data)
|
|
print(f"l74_user_data {user_data}")
|
|
print(f"l75_user_db: {user_db.tasks}")
|
|
session.add(user_db)
|
|
print("l##_db: ", user_db.tasks)
|
|
session.commit()
|
|
print("l76_not_refresh_db: ", user_db.tasks) # randomly sets is_completed to false even though its true every other time
|
|
session.refresh(user_db)
|
|
print(f"l77_db: {user_db.tasks}")
|
|
return user_db
|
|
|
|
@app.get('/user/login')
|
|
async def verify_user(user: VerifyUser, session: SessionDep):
|
|
existing_user = session.query(User).filter(User.email == user.email).first()
|
|
if not existing_user:
|
|
raise HTTPException(status_code=400, detail="Email not registered")
|
|
|
|
is_password_valid = validate_password(user.password, existing_user.password, existing_user.salt)
|
|
|
|
if not is_password_valid:
|
|
raise HTTPException(status_code=400, detail="Password not valid")
|
|
|
|
return {"is_password_valid": is_password_valid, "email": existing_user.email, "id": existing_user.id}
|
|
|
|
|
|
@app.post('/token', response_model=Token)
|
|
async def login_for_access_token(session: SessionDep, form_data: OAuth2PasswordRequestForm = Depends()):
|
|
user = authenticate_user(form_data.username, form_data.password, session)
|
|
if not user:
|
|
raise HTTPException(status_code=401, detail="Incorrect username or password", headers={"WWW-Authenticate": "Bearer"})
|
|
|
|
access_token_expires = timedelta(minutes=int(os.getenv("ACCESS_TOKEN_EXPIRES_MINUTES")))
|
|
access_token = create_access_token(data={"sub": user.username}, expires_delta=access_token_expires)
|
|
return {"access_token": access_token, "token_type": "bearer"}
|
|
|
|
@app.get("/users/me/", response_model=UserPublic)
|
|
async def read_users_me(session: SessionDep, current_user: User = Depends(get_current_user)):
|
|
return current_user |