self-constructing_graph/agent/router/user_router.py

import sys,os
import uuid
current_path = os.getcwd()
sys.path.append(current_path)

from config.site import SiteConfig
from fastapi import APIRouter, Depends, Query
from db.database import get_db
from sqlalchemy.orm import Session
from agent.models.web.response import StandardResponse,FAILED,SUCCESS
from agent.models.web.request import BasicRequest
from agent.libs.user import UserBusiness,SessionBusiness
import logging

router = APIRouter(prefix="/user", tags=["agent job interface"])
logger = logging.getLogger(__name__)
config = SiteConfig()

@router.post("/session", response_model=StandardResponse)
def register(request: BasicRequest, db: Session = Depends(get_db)):
    if request.action == 'register':
        biz = UserBusiness(db)
        request_username = request.get_param("username", "")
        request_password = request.get_param("password", "")
        user = biz.get_user_by_username(request_username)
        if user is not None:
            return StandardResponse(code=FAILED, message="user already exists")
        user = biz.create_user(request_username, request_password)
        if user is None:
            return StandardResponse(code=FAILED, message="create user failed")
        return StandardResponse(code=SUCCESS, message="create user success")
    elif request.action =='login':
        request_username = request.get_param("username", "")
        request_password = request.get_param("password", "")
        logger.info(f"login: {request_username} {request_password}")
        biz = UserBusiness(db)
        user = biz.get_user_by_username(request_username)
        if user is None:
            return StandardResponse(code=FAILED, message="user not exists")
        if not biz.verify_password(request_password, user.hashed_password):
            return StandardResponse(code=FAILED, message="password error")

        session = SessionBusiness(db)
        old_session = session.get_session_by_user_id(user.id)
        if old_session is not None:
            logger.info("delete old session")
            session.delete_session(old_session.session_id)
            
        logger.info("create new session")
        new_session = session.create_session(user)
        return StandardResponse(code=SUCCESS, message="login success", records=[new_session])
    elif request.action == "login_session":
        session_id = request.get_param("session_id", "")
        session = SessionBusiness(db)
        old_session = session.get_session(session_id)
        if old_session is None:
            return StandardResponse(code=FAILED, message="session not exists")
        return StandardResponse(code=SUCCESS, message="login success", records=[old_session])
    elif request.action == "logout":
        session_id = request.get_param("session_id", "")
        session = SessionBusiness(db)
        session.delete_session(session_id)
        return StandardResponse(code=SUCCESS, message="logout success")

@router.get("/logout/{session_id}", response_model=StandardResponse)
def logout(session_id: str, db: Session = Depends(get_db)):
    session = SessionBusiness(db)
    session.delete_session(session_id)
    return StandardResponse(code=SUCCESS, message="logout success")

@router.post("/signin", response_model=StandardResponse)
def signin(request: BasicRequest, db: Session = Depends(get_db)):
    if request.action == 'signin':
        biz = UserBusiness(db)
        request_username = request.get_param("username", "")
        request_password = request.get_param("password", "")
        request_fullname = request.get_param("full_name", "")
        request_email = request.get_param("email", "")
        user = biz.get_user_by_username(request_username)
        if user is not None:
            return StandardResponse(code=FAILED, message="用户名已存在")
        user = biz.create_user(request_username, request_password, request_fullname, request_email)
        if user is None:
            return StandardResponse(code=FAILED, message="创建用户失败")
        return StandardResponse(code=SUCCESS, message="成功创建用户，请继续登录")
    return StandardResponse(code=FAILED, message="invalid action")
user_router = router