XYHLF / App /routers /users /routes.py
Mbonea's picture
removed pool erros
e558c69
from fastapi import APIRouter, HTTPException, Depends
from .models import User, Watchlist
from App.routers.portfolio.models import Portfolio
from .schemas import UserCreate, UserResponse, PortfolioItemSchema, WatchlistItemSchema, UserLogin
from App.schemas import ResponseModel
from tortoise.contrib.pydantic import pydantic_model_creator, pydantic_queryset_creator
from passlib.hash import bcrypt
from App.schemas import AppException
router = APIRouter(prefix="/users", tags=["Users"])
User_Pydantic = pydantic_model_creator(User, name="User")
Portfolio_Pydantic = pydantic_queryset_creator(Portfolio, name="Portfolio")
Portfolio_one_Pydantic = pydantic_model_creator(Portfolio, name="Portfolio")
@router.post("/login", response_model=ResponseModel)
async def login(user: UserLogin):
user_obj = await User.get_or_none(email=user.email)
if not user_obj:
raise AppException(status_code=400, detail=ResponseModel(success=False, message="Invalid email or password"))
if not bcrypt.verify(user.password, user_obj.hashed_password):
raise AppException(status_code=400, detail=ResponseModel(success=False, message="Invalid email or password"))
# Use the modified to_dict method
_user = await user_obj.to_dict()
# The _user object is now a Pydantic model, so we can pass it to UserResponse
_user_response = UserResponse.model_validate(_user.model_dump())
return ResponseModel(success=True, message="Login successful", data=_user_response)
@router.post("/register", response_model=ResponseModel)
async def register(user: UserCreate):
existing = await User.get_or_none(email=user.email)
if existing:
raise AppException(status_code=400, detail=ResponseModel(success=False, message="Email already registered"))
user_obj = await User.create(
username=user.username,
email=user.email,
hashed_password=bcrypt.hash(user.password)
)
await Portfolio.create(user=user_obj, name="Default Portfolio")
return ResponseModel(success=True, message="User created", data=await User_Pydantic.from_tortoise_orm(user_obj))
@router.get("/{user_id}/portfolio", response_model=ResponseModel)
async def get_portfolio(user_id: int):
portfolios = Portfolio.filter(user=user_id).all()
_portfolios = await Portfolio_Pydantic.from_queryset(portfolios)
return ResponseModel(success=True, message="Portfolio retrieved", data=_portfolios.model_dump())
@router.post("/{user_id}/portfolio", response_model=ResponseModel)
async def create_portfolio(user_id: int, data: PortfolioItemSchema):
# Check if user exists
user = await User.get_or_none(id=user_id)
if not user:
raise HTTPException(status_code=404, detail="User not found")
# Create a new portfolio
portfolio = await Portfolio.create(
user=user,
name=data.name
)
_portfolio=Portfolio_one_Pydantic.from_orm(portfolio)
return ResponseModel(success=True, message="Portfolio created", data=_portfolio.model_dump())
@router.post("/{user_id}/watchlist/add", response_model=ResponseModel)
async def add_to_watchlist(user_id: int, item: WatchlistItemSchema):
new_item = await Watchlist.create(
user_id=user_id,
stock_id=item.stock_id,
utt_id=item.utt_id
)
return ResponseModel(success=True, message="Added to watchlist", data=new_item)