|
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")) |
|
|
|
|
|
_user = await user_obj.to_dict() |
|
|
|
|
|
_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): |
|
|
|
user = await User.get_or_none(id=user_id) |
|
if not user: |
|
raise HTTPException(status_code=404, detail="User not found") |
|
|
|
|
|
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) |