111 lines
3.7 KiB
Python
111 lines
3.7 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-tasks", response_model=UserPublic)
|
|
async def update_user_task(task: UpdateTask, session: SessionDep, current_user: User = Depends(get_current_user)):
|
|
user_db = session.get(User, current_user.id)
|
|
user_data = task.model_dump(exclude_unset=True)
|
|
|
|
new_task = user_data
|
|
|
|
# Filter out the task with matching title and keep all others
|
|
updated_tasks = [
|
|
task for task in user_db.tasks
|
|
if task['title'] != new_task['title']
|
|
]
|
|
|
|
# Add the new task to the filtered list
|
|
updated_tasks.append(new_task)
|
|
|
|
# Update user_db tasks with the new list
|
|
user_db.tasks = updated_tasks
|
|
|
|
# Update the session
|
|
session.add(user_db)
|
|
session.commit()
|
|
session.refresh(user_db)
|
|
|
|
return user_db
|
|
|
|
@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)
|
|
|
|
user_db.sqlmodel_update(user_data)
|
|
session.add(user_db)
|
|
session.commit()
|
|
session.refresh(user_db)
|
|
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 |