mekasu0124

Untitled

Jun 12th, 2024
107
0
Never
Not a member of Pastebin yet? Sign Up, it unlocks many cool features!
Python 4.28 KB | None | 0 0
  1. # root/backend/api/routers/routes.py
  2. import json
  3.  
  4. from fastapi import APIRouter, HTTPException, Depends, status
  5. from fastapi.security import OAuth2PasswordRequestForm, OAuth2PasswordBearer
  6. from sqlalchemy.orm import Session
  7. import bcrypt
  8.  
  9. from api.models.models import UserBase, User
  10. from api.validation.account_validation import (
  11.     validate_email,
  12.     validate_password,
  13.     is_name_not_in_password,
  14.     validate_universal_phone_number,
  15.     hash_password
  16. )
  17. from main import get_db, generate_session_token, oauth2_scheme
  18.  
  19. router = APIRouter()
  20.  
  21. oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
  22.  
  23. @router.get("/")
  24. async def root():
  25.     text = [
  26.         "Embark on a journey into the future of technology through our diverse range of mobile, desktop,",
  27.         "and web applications meticulously crafted to simplify and enrich your daily life. Immerse yourself",
  28.         "in the convenience of our mobile applications, offering innovative tools for on-the-go efficiency.",
  29.         "Elevate your productivity with our desktop applications, providing powerful features to streamline",
  30.         "your work seamlessly. Experience seamless online interactions with our web applications, designed",
  31.         "to bring ease and efficiency to your digital experiences. Our mission is to seamlessly integrate",
  32.         "technology into your lifestyle, making tasks easier, interactions more enjoyable, and connections",
  33.         "more meaningful. Step into a world where technology enhances every aspect of your life,",
  34.         "simplifying your tasks and enriching your digital journey."
  35.     ]
  36.     return { "message": "Welcome To Mek's Hub!", "description": " ".join(text) }
  37.  
  38. @router.get("/logged_in_links/")
  39. async def get_logged_in_links():
  40.     with open("./api/routers/route_paths.json", 'r', encoding="utf-8-sig") as f:
  41.         data = json.load(f)
  42.  
  43.     return { "navLinks": data[0] }
  44.  
  45. @router.get("/logged_out_links/")
  46. async def get_logged_out_links():
  47.     with open("./api/routers/route_paths.json", 'r', encoding="utf-8-sig") as f:
  48.         data = json.load(f)
  49.  
  50.     return { "navLinks": data[1] }
  51.  
  52. @router.post("/create_account/")
  53. async def create_new_user(user: UserBase, db: Session = Depends(get_db)):
  54.     if not validate_email(user.email_address):
  55.         raise HTTPException(status_code=400, detail="Not A Valid Email")
  56.    
  57.     if not validate_password(user.password):
  58.         raise HTTPException(status_code=400, detail="Not A Valid Password")
  59.    
  60.     if not is_name_not_in_password(user.first_name, user.last_name, user.password):
  61.         raise HTTPException(status_code=400, detail="Name Cannot Be In Password")
  62.    
  63.     if not validate_universal_phone_number(user.phone_number):
  64.         raise HTTPException(status_code=400, detail="Invalid Phone Number")
  65.    
  66.     query = db.query(User.id).filter(User.email_address == user.email_address)
  67.     user_exists = db.query(query.exists()).scalar()
  68.  
  69.     if user_exists:
  70.         raise HTTPException(status_code=409, detail="User Already Exists")
  71.    
  72.     user.password = hash_password(user.password)
  73.  
  74.     user = User(**user.model_dump())
  75.  
  76.     db.add(user)
  77.     db.commit()
  78.     db.refresh(user)
  79.     return user
  80.  
  81. @router.post("/login/")
  82. async def login(form_data: OAuth2PasswordRequestForm = Depends(), db: Session = Depends(get_db)):
  83.     user = db.query(User).filter(User.username == form_data.username).first()
  84.  
  85.     if not user:
  86.         raise HTTPException(status_code=400, detail="Incorrect Username/Password")
  87.    
  88.     if not bcrypt.checkpw(form_data.password.encode("utf-8"), user.password):
  89.         raise HTTPException(status_code=400, detail="Incorrect Username/Password")
  90.  
  91.     session_token = generate_session_token(user.id)
  92.  
  93.     return { "access_token": session_token, "token_type": "bearer" }
  94.  
  95. @router.get("/user/")
  96. async def get_user_data(db: Session = Depends(get_db), token: str = Depends(oauth2_scheme)):
  97.     session = db.query(Session).filter(Session.token == token).first()
  98.  
  99.     if not session:
  100.         raise HTTPException(status_code=401, detail="Invalid session token")
  101.    
  102.     user = db.query(User).filter(User.id == session.user_id).first()
  103.  
  104.     if not user:
  105.         raise HTTPException(status_code=404, detail="User not found")
  106.  
  107.     return {
  108.         "userName": user.username,
  109.         "loggedIn": True
  110.     }
Advertisement
Add Comment
Please, Sign In to add comment