Spaces:
Running
Running
File size: 2,772 Bytes
478dec6 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 | from externals.databases.database import get_db
from externals.databases.pg_crud import (
get_user_by_username,
create_user,
get_user_by_email,
# deactivate_user,
# get_tenant_by_name,
# create_tenant,
)
from externals.databases.schemas.user import UserCreate, UserResponse
from externals.databases.schemas.auth import LoginRequest, TokenResponse
from fastapi import APIRouter, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordRequestForm
from sqlalchemy.ext.asyncio import AsyncSession
from utils.logger import get_logger
from utils.security import verify_password
from utils.jwt import create_access_token
logger = get_logger("user management")
router = APIRouter(prefix="/admin", tags=["User"])
@router.post(
"/users",
status_code=status.HTTP_201_CREATED,
response_model=UserResponse,
)
async def create_user_endpoint(
user: UserCreate,
db: AsyncSession = Depends(get_db),
):
logger.info("Create user request", extra={"username": user.username})
existing = await get_user_by_username(db, user.username)
if existing:
raise HTTPException(
status_code=status.HTTP_409_CONFLICT,
detail="Username already exists",
)
return await create_user(db, user)
# @router.get(
# "/users/{username}",
# response_model=UserResponse,
# )
# async def get_user_endpoint(
# username: str,
# db: AsyncSession = Depends(get_db),
# ):
# logger.info("Get user request", extra={"username": username})
# user = await get_user_by_username(db, username)
# if not user:
# raise HTTPException(
# status_code=status.HTTP_404_NOT_FOUND,
# detail="User not found",
# )
# return user
@router.post("/login", response_model=TokenResponse)
async def login(
form_data: OAuth2PasswordRequestForm = Depends(),
db: AsyncSession = Depends(get_db),
):
user = await get_user_by_email(db, form_data.username)
if not user or not verify_password(form_data.password, user.hashed_password):
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Invalid email or password",
)
token = create_access_token(
data={
"sub": str(user.user_id),
"email": user.email,
"role": user.role,
"full_name": user.full_name,
}
)
return TokenResponse(access_token=token)
from interfaces.api.deps import get_current_user
@router.get("/me", response_model=UserResponse)
async def get_me(
current_user: UserResponse = Depends(get_current_user),
):
logger.info(
"Get current user",
extra={"user_id": current_user.user_id},
)
return current_user |