Spaces:
Running
Running
File size: 4,371 Bytes
d3cce17 |
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 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 |
from datetime import datetime
from src.utils.mongo import UserCRUD
from src.utils.logger import logger
from bson import ObjectId
from fastapi.responses import JSONResponse
from typing import Dict
async def list_users_controller():
try:
users = await UserCRUD.find_all()
# Convert datetime fields to ISO format
for user in users:
if "created_at" in user and isinstance(user["created_at"], datetime):
user["created_at"] = user["created_at"].isoformat()
if "updated_at" in user and isinstance(user["updated_at"], datetime):
user["updated_at"] = user["updated_at"].isoformat()
if "expire_at" in user and isinstance(user["expire_at"], datetime):
user["expire_at"] = user["expire_at"].isoformat()
return JSONResponse(
content={
"status": "success",
"data": users,
},
status_code=200,
)
except Exception as e:
logger.error(f"Error listing users: {e}")
return JSONResponse(
content={
"status": "error",
"message": str(e),
},
status_code=500,
)
async def update_user_controller(user_id: str, user_data: dict):
try:
user = await UserCRUD.update({"_id": ObjectId(user_id)}, user_data)
# Convert datetime fields to ISO format
if user:
if "created_at" in user and isinstance(user["created_at"], datetime):
user["created_at"] = user["created_at"].isoformat()
if "updated_at" in user and isinstance(user["updated_at"], datetime):
user["updated_at"] = user["updated_at"].isoformat()
if "expire_at" in user and isinstance(user["expire_at"], datetime):
user["expire_at"] = user["expire_at"].isoformat()
return JSONResponse(
content={
"status": "success",
"data": user,
},
status_code=200,
)
else:
return JSONResponse(
content={
"status": "error",
"message": "User not found",
},
status_code=404,
)
except Exception as e:
logger.error(f"Error updating user: {e}")
return JSONResponse(
content={
"status": "error",
"message": str(e),
},
status_code=500,
)
async def delete_user_controller(user_id: str):
try:
user = await UserCRUD.delete_one({"_id": ObjectId(user_id)})
return JSONResponse(
content={
"status": "success",
"data": user,
},
status_code=200,
)
except Exception as e:
logger.error(f"Error deleting user: {e}")
return JSONResponse(
content={
"status": "error",
"message": str(e),
},
status_code=500,
)
async def get_user_by_id_controller(user_id: str):
try:
user = await UserCRUD.find_by_id(user_id)
if not user:
return JSONResponse(
content={
"status": "error",
"message": "User not found",
},
status_code=404,
)
# Convert datetime fields to ISO format
if "created_at" in user and isinstance(user["created_at"], datetime):
user["created_at"] = user["created_at"].isoformat()
if "updated_at" in user and isinstance(user["updated_at"], datetime):
user["updated_at"] = user["updated_at"].isoformat()
if "expire_at" in user and isinstance(user["expire_at"], datetime):
user["expire_at"] = user["expire_at"].isoformat()
return JSONResponse(
content={
"status": "success",
"data": user,
},
status_code=200,
)
except Exception as e:
logger.error(f"Error getting user by id: {e}")
return JSONResponse(
content={
"status": "error",
"message": str(e),
},
status_code=500,
)
|