The new updates of partners and user
Made-with: Cursor
This commit is contained in:
Vivek P Prakash
946
accounts/api.py
Normal file
946
accounts/api.py
Normal file
@@ -0,0 +1,946 @@
|
||||
from django.http import JsonResponse
|
||||
from django.utils.decorators import method_decorator
|
||||
from django.views.decorators.csrf import csrf_exempt
|
||||
from django.forms.models import model_to_dict
|
||||
from django.contrib.auth import authenticate, logout
|
||||
from rest_framework.views import APIView
|
||||
from rest_framework.authtoken.models import Token
|
||||
import json
|
||||
|
||||
from .models import User
|
||||
from mobile_api.utils import validate_token_and_get_user
|
||||
|
||||
|
||||
def _partner_user_to_dict(user, request=None):
|
||||
"""Serialize partner-related User for JSON (same structure as _user_to_dict)."""
|
||||
data = model_to_dict(
|
||||
user,
|
||||
fields=[
|
||||
"id",
|
||||
"username",
|
||||
"email",
|
||||
"phone_number",
|
||||
"role",
|
||||
"is_staff",
|
||||
"is_customer",
|
||||
"is_user",
|
||||
"pincode",
|
||||
"district",
|
||||
"state",
|
||||
"country",
|
||||
"place",
|
||||
"latitude",
|
||||
"longitude",
|
||||
"first_name",
|
||||
"last_name",
|
||||
],
|
||||
)
|
||||
# Add profile picture URL if exists
|
||||
if getattr(user, "profile_picture", None):
|
||||
if request:
|
||||
data["profile_picture"] = request.build_absolute_uri(user.profile_picture.url)
|
||||
else:
|
||||
data["profile_picture"] = user.profile_picture.url
|
||||
else:
|
||||
data["profile_picture"] = None
|
||||
return data
|
||||
|
||||
|
||||
def _user_to_dict(user, request=None):
|
||||
"""Serialize any User for JSON (admin/staff or partner)."""
|
||||
return _partner_user_to_dict(user, request)
|
||||
|
||||
|
||||
@method_decorator(csrf_exempt, name="dispatch")
|
||||
class PartnerLoginAPI(APIView):
|
||||
"""
|
||||
Partner Login API.
|
||||
Body: username (or email), password (required).
|
||||
Returns: token, user data.
|
||||
"""
|
||||
|
||||
def post(self, request):
|
||||
try:
|
||||
# Parse JSON or form data
|
||||
is_multipart = request.content_type and "multipart/form-data" in request.content_type
|
||||
if is_multipart:
|
||||
data = request.POST.dict()
|
||||
else:
|
||||
try:
|
||||
data = json.loads(request.body)
|
||||
except json.JSONDecodeError:
|
||||
return JsonResponse(
|
||||
{"status": "error", "message": "Invalid JSON"},
|
||||
status=400,
|
||||
)
|
||||
|
||||
username = data.get("username") or data.get("email")
|
||||
password = data.get("password")
|
||||
|
||||
if not username or not password:
|
||||
return JsonResponse(
|
||||
{"status": "error", "message": "username and password are required."},
|
||||
status=400,
|
||||
)
|
||||
|
||||
# Authenticate user
|
||||
user = authenticate(request, username=username, password=password)
|
||||
if not user:
|
||||
return JsonResponse(
|
||||
{"status": "error", "message": "Invalid username or password."},
|
||||
status=401,
|
||||
)
|
||||
|
||||
# Check if user has partner role
|
||||
partner_roles = ["partner", "partner_manager", "partner_staff"]
|
||||
if user.role not in partner_roles:
|
||||
return JsonResponse(
|
||||
{"status": "error", "message": "You are not authorized to access partner portal."},
|
||||
status=403,
|
||||
)
|
||||
|
||||
# Get or create token
|
||||
token, _ = Token.objects.get_or_create(user=user)
|
||||
|
||||
return JsonResponse(
|
||||
{
|
||||
"status": "success",
|
||||
"message": "Login successful",
|
||||
"token": token.key,
|
||||
"user": _partner_user_to_dict(user, request),
|
||||
},
|
||||
status=200,
|
||||
)
|
||||
|
||||
except Exception as e:
|
||||
return JsonResponse({"status": "error", "message": str(e)}, status=500)
|
||||
|
||||
|
||||
@method_decorator(csrf_exempt, name="dispatch")
|
||||
class PartnerLogoutAPI(APIView):
|
||||
"""
|
||||
Partner Logout API.
|
||||
Body: token, username (required).
|
||||
Returns: success message.
|
||||
"""
|
||||
|
||||
def post(self, request):
|
||||
try:
|
||||
user, token, data, error_response = validate_token_and_get_user(request, error_status_code=True)
|
||||
if error_response:
|
||||
return error_response
|
||||
|
||||
# Check if user has partner role
|
||||
partner_roles = ["partner", "partner_manager", "partner_staff"]
|
||||
if user.role not in partner_roles:
|
||||
return JsonResponse(
|
||||
{"status": "error", "message": "You are not authorized to access partner portal."},
|
||||
status=403,
|
||||
)
|
||||
|
||||
# Delete token
|
||||
token.delete()
|
||||
|
||||
return JsonResponse(
|
||||
{
|
||||
"status": "success",
|
||||
"message": "Logged out successfully.",
|
||||
},
|
||||
status=200,
|
||||
)
|
||||
|
||||
except Exception as e:
|
||||
return JsonResponse({"status": "error", "message": str(e)}, status=500)
|
||||
|
||||
|
||||
@method_decorator(csrf_exempt, name="dispatch")
|
||||
class PartnerDashboardAPI(APIView):
|
||||
"""
|
||||
Partner Dashboard API.
|
||||
Body: token, username (required).
|
||||
Returns: dashboard statistics.
|
||||
"""
|
||||
|
||||
def post(self, request):
|
||||
try:
|
||||
user, token, data, error_response = validate_token_and_get_user(request, error_status_code=True)
|
||||
if error_response:
|
||||
return error_response
|
||||
|
||||
# Check if user has partner role
|
||||
partner_roles = ["partner", "partner_manager", "partner_staff"]
|
||||
if user.role not in partner_roles:
|
||||
return JsonResponse(
|
||||
{"status": "error", "message": "You are not authorized to access this page."},
|
||||
status=403,
|
||||
)
|
||||
|
||||
# Get statistics for partner users (including partner_customer)
|
||||
all_partner_roles = ["partner", "partner_manager", "partner_staff", "partner_customer"]
|
||||
partner_users = User.objects.filter(role__in=all_partner_roles)
|
||||
total_partner_users = partner_users.count()
|
||||
|
||||
return JsonResponse(
|
||||
{
|
||||
"status": "success",
|
||||
"dashboard": {
|
||||
"total_partner_users": total_partner_users,
|
||||
},
|
||||
},
|
||||
status=200,
|
||||
)
|
||||
|
||||
except Exception as e:
|
||||
return JsonResponse({"status": "error", "message": str(e)}, status=500)
|
||||
|
||||
|
||||
@method_decorator(csrf_exempt, name="dispatch")
|
||||
class PartnerListUsersAPI(APIView):
|
||||
"""
|
||||
Partner List Users API.
|
||||
Body: token, username (required);
|
||||
role (optional filter: partner, partner_manager, partner_staff, partner_customer).
|
||||
Returns: list of partner users.
|
||||
"""
|
||||
|
||||
def post(self, request):
|
||||
try:
|
||||
user, token, data, error_response = validate_token_and_get_user(request, error_status_code=True)
|
||||
if error_response:
|
||||
return error_response
|
||||
|
||||
# Check if user has partner role
|
||||
partner_roles = ["partner", "partner_manager", "partner_staff"]
|
||||
if user.role not in partner_roles:
|
||||
return JsonResponse(
|
||||
{"status": "error", "message": "You are not authorized to access this page."},
|
||||
status=403,
|
||||
)
|
||||
|
||||
# Filter users by partner-related roles
|
||||
all_partner_roles = ["partner", "partner_manager", "partner_staff", "partner_customer"]
|
||||
qs = User.objects.filter(role__in=all_partner_roles).order_by("-id")
|
||||
|
||||
# Optional role filter
|
||||
role_filter = data.get("role")
|
||||
if role_filter:
|
||||
if role_filter not in all_partner_roles:
|
||||
return JsonResponse(
|
||||
{
|
||||
"status": "error",
|
||||
"message": f"Invalid role filter. Must be one of: {', '.join(all_partner_roles)}",
|
||||
},
|
||||
status=400,
|
||||
)
|
||||
qs = qs.filter(role=role_filter)
|
||||
|
||||
users = [_partner_user_to_dict(u, request) for u in qs]
|
||||
|
||||
return JsonResponse(
|
||||
{
|
||||
"status": "success",
|
||||
"users": users,
|
||||
"total_count": len(users),
|
||||
},
|
||||
status=200,
|
||||
)
|
||||
|
||||
except Exception as e:
|
||||
return JsonResponse({"status": "error", "message": str(e)}, status=500)
|
||||
|
||||
|
||||
@method_decorator(csrf_exempt, name="dispatch")
|
||||
class PartnerCreateUserAPI(APIView):
|
||||
"""
|
||||
Partner Create User API.
|
||||
Body: token, username, username (for new user), email, password, role (required);
|
||||
full_name, phone_number, pincode, district, state, country, place, latitude, longitude (optional).
|
||||
Returns: created user data.
|
||||
"""
|
||||
|
||||
def post(self, request):
|
||||
try:
|
||||
user, token, data, error_response = validate_token_and_get_user(request, error_status_code=True)
|
||||
if error_response:
|
||||
return error_response
|
||||
|
||||
# Check if user has partner role
|
||||
partner_roles = ["partner", "partner_manager", "partner_staff"]
|
||||
if user.role not in partner_roles:
|
||||
return JsonResponse(
|
||||
{"status": "error", "message": "You are not authorized to access this page."},
|
||||
status=403,
|
||||
)
|
||||
|
||||
# Extract user data
|
||||
new_username = data.get("username")
|
||||
email = data.get("email")
|
||||
password = data.get("password")
|
||||
role = data.get("role")
|
||||
full_name = data.get("full_name", "").strip()
|
||||
|
||||
if not all([new_username, email, password, role]):
|
||||
return JsonResponse(
|
||||
{
|
||||
"status": "error",
|
||||
"message": "username, email, password, and role are required.",
|
||||
},
|
||||
status=400,
|
||||
)
|
||||
|
||||
# Validate role - must be one of the partner-related roles
|
||||
valid_partner_roles = ["partner", "partner_manager", "partner_staff", "partner_customer"]
|
||||
if role not in valid_partner_roles:
|
||||
return JsonResponse(
|
||||
{
|
||||
"status": "error",
|
||||
"message": f"Invalid role. Must be one of: {', '.join(valid_partner_roles)}",
|
||||
},
|
||||
status=400,
|
||||
)
|
||||
|
||||
# Check if username already exists
|
||||
if User.objects.filter(username=new_username).exists():
|
||||
return JsonResponse(
|
||||
{"status": "error", "message": "Username already exists."},
|
||||
status=400,
|
||||
)
|
||||
|
||||
# Check if email already exists
|
||||
if User.objects.filter(email=email).exists():
|
||||
return JsonResponse(
|
||||
{"status": "error", "message": "Email already exists."},
|
||||
status=400,
|
||||
)
|
||||
|
||||
# Create user
|
||||
new_user = User.objects.create_user(
|
||||
username=new_username,
|
||||
email=email,
|
||||
password=password,
|
||||
role=role,
|
||||
phone_number=data.get("phone_number"),
|
||||
pincode=data.get("pincode"),
|
||||
district=data.get("district"),
|
||||
state=data.get("state"),
|
||||
country=data.get("country"),
|
||||
place=data.get("place"),
|
||||
)
|
||||
|
||||
# Handle full_name - split into first_name and last_name
|
||||
if full_name:
|
||||
parts = full_name.split(None, 1)
|
||||
new_user.first_name = parts[0]
|
||||
if len(parts) > 1:
|
||||
new_user.last_name = parts[1]
|
||||
|
||||
# Set location coordinates if provided
|
||||
if data.get("latitude") is not None:
|
||||
try:
|
||||
latitude = float(data["latitude"])
|
||||
if latitude < -90 or latitude > 90:
|
||||
return JsonResponse(
|
||||
{"status": "error", "message": "latitude must be between -90 and 90."},
|
||||
status=400,
|
||||
)
|
||||
new_user.latitude = latitude
|
||||
except (TypeError, ValueError):
|
||||
return JsonResponse(
|
||||
{"status": "error", "message": "latitude must be numeric."},
|
||||
status=400,
|
||||
)
|
||||
|
||||
if data.get("longitude") is not None:
|
||||
try:
|
||||
longitude = float(data["longitude"])
|
||||
if longitude < -180 or longitude > 180:
|
||||
return JsonResponse(
|
||||
{"status": "error", "message": "longitude must be between -180 and 180."},
|
||||
status=400,
|
||||
)
|
||||
new_user.longitude = longitude
|
||||
except (TypeError, ValueError):
|
||||
return JsonResponse(
|
||||
{"status": "error", "message": "longitude must be numeric."},
|
||||
status=400,
|
||||
)
|
||||
|
||||
# Handle profile picture upload if provided
|
||||
if "profile_picture" in request.FILES:
|
||||
new_user.profile_picture = request.FILES["profile_picture"]
|
||||
|
||||
new_user.save()
|
||||
|
||||
return JsonResponse(
|
||||
{
|
||||
"status": "success",
|
||||
"message": f"User created successfully with role: {role}.",
|
||||
"user": _partner_user_to_dict(new_user, request),
|
||||
},
|
||||
status=201,
|
||||
)
|
||||
|
||||
except Exception as e:
|
||||
return JsonResponse({"status": "error", "message": str(e)}, status=500)
|
||||
|
||||
|
||||
@method_decorator(csrf_exempt, name="dispatch")
|
||||
class PartnerUpdateUserAPI(APIView):
|
||||
"""
|
||||
Partner Update User API.
|
||||
Body: token, username, user_id (required);
|
||||
email, phone_number, role, full_name, pincode, district, state,
|
||||
country, place, latitude, longitude, password, profile_picture (optional).
|
||||
Returns: updated user data.
|
||||
"""
|
||||
|
||||
def post(self, request):
|
||||
try:
|
||||
user, token, data, error_response = validate_token_and_get_user(request, error_status_code=True)
|
||||
if error_response:
|
||||
return error_response
|
||||
|
||||
# Check if user has partner role
|
||||
partner_roles = ["partner", "partner_manager", "partner_staff"]
|
||||
if user.role not in partner_roles:
|
||||
return JsonResponse(
|
||||
{"status": "error", "message": "You are not authorized to access this page."},
|
||||
status=403,
|
||||
)
|
||||
|
||||
user_id = data.get("user_id")
|
||||
if not user_id:
|
||||
return JsonResponse(
|
||||
{"status": "error", "message": "user_id is required."},
|
||||
status=400,
|
||||
)
|
||||
|
||||
try:
|
||||
target_user = User.objects.get(id=user_id)
|
||||
except User.DoesNotExist:
|
||||
return JsonResponse(
|
||||
{"status": "error", "message": "User not found."},
|
||||
status=404,
|
||||
)
|
||||
|
||||
# Validate that the user has a partner-related role
|
||||
all_partner_roles = ["partner", "partner_manager", "partner_staff", "partner_customer"]
|
||||
if target_user.role not in all_partner_roles:
|
||||
return JsonResponse(
|
||||
{
|
||||
"status": "error",
|
||||
"message": "User is not a partner-related user. Only users with partner roles can be updated.",
|
||||
},
|
||||
status=400,
|
||||
)
|
||||
|
||||
# Update fields if provided
|
||||
if data.get("email") is not None:
|
||||
new_email = data["email"]
|
||||
# Check if email already exists for another user
|
||||
if User.objects.filter(email=new_email).exclude(id=user_id).exists():
|
||||
return JsonResponse(
|
||||
{"status": "error", "message": "Email already exists for another user."},
|
||||
status=400,
|
||||
)
|
||||
target_user.email = new_email
|
||||
|
||||
if data.get("phone_number") is not None:
|
||||
target_user.phone_number = data["phone_number"] or None
|
||||
|
||||
if data.get("role") is not None:
|
||||
new_role = data["role"]
|
||||
if new_role not in all_partner_roles:
|
||||
return JsonResponse(
|
||||
{
|
||||
"status": "error",
|
||||
"message": f"Invalid role. Must be one of: {', '.join(all_partner_roles)}",
|
||||
},
|
||||
status=400,
|
||||
)
|
||||
target_user.role = new_role
|
||||
|
||||
# Handle full_name
|
||||
if data.get("full_name"):
|
||||
full_name = data["full_name"].strip()
|
||||
if full_name:
|
||||
parts = full_name.split(None, 1)
|
||||
target_user.first_name = parts[0]
|
||||
if len(parts) > 1:
|
||||
target_user.last_name = parts[1]
|
||||
else:
|
||||
target_user.last_name = ""
|
||||
|
||||
if "pincode" in data:
|
||||
target_user.pincode = data["pincode"] or None
|
||||
if "district" in data:
|
||||
target_user.district = data["district"] or None
|
||||
if "state" in data:
|
||||
target_user.state = data["state"] or None
|
||||
if "country" in data:
|
||||
target_user.country = data["country"] or None
|
||||
if "place" in data:
|
||||
target_user.place = data["place"] or None
|
||||
|
||||
if data.get("latitude") is not None:
|
||||
try:
|
||||
latitude = float(data["latitude"])
|
||||
if latitude < -90 or latitude > 90:
|
||||
return JsonResponse(
|
||||
{"status": "error", "message": "latitude must be between -90 and 90."},
|
||||
status=400,
|
||||
)
|
||||
target_user.latitude = latitude
|
||||
except (TypeError, ValueError):
|
||||
return JsonResponse(
|
||||
{"status": "error", "message": "latitude must be numeric."},
|
||||
status=400,
|
||||
)
|
||||
|
||||
if data.get("longitude") is not None:
|
||||
try:
|
||||
longitude = float(data["longitude"])
|
||||
if longitude < -180 or longitude > 180:
|
||||
return JsonResponse(
|
||||
{"status": "error", "message": "longitude must be between -180 and 180."},
|
||||
status=400,
|
||||
)
|
||||
target_user.longitude = longitude
|
||||
except (TypeError, ValueError):
|
||||
return JsonResponse(
|
||||
{"status": "error", "message": "longitude must be numeric."},
|
||||
status=400,
|
||||
)
|
||||
|
||||
# Handle profile picture upload if provided
|
||||
if "profile_picture" in request.FILES:
|
||||
target_user.profile_picture = request.FILES["profile_picture"]
|
||||
|
||||
# Handle password update if provided
|
||||
if data.get("password"):
|
||||
target_user.set_password(data["password"])
|
||||
|
||||
target_user.save()
|
||||
|
||||
return JsonResponse(
|
||||
{
|
||||
"status": "success",
|
||||
"message": "Partner user updated successfully.",
|
||||
"user": _partner_user_to_dict(target_user, request),
|
||||
},
|
||||
status=200,
|
||||
)
|
||||
|
||||
except Exception as e:
|
||||
return JsonResponse({"status": "error", "message": str(e)}, status=500)
|
||||
|
||||
|
||||
@method_decorator(csrf_exempt, name="dispatch")
|
||||
class PartnerDeleteUserAPI(APIView):
|
||||
"""
|
||||
Partner Delete User API.
|
||||
Body: token, username, user_id (required).
|
||||
Returns: success message.
|
||||
"""
|
||||
|
||||
def post(self, request):
|
||||
try:
|
||||
user, token, data, error_response = validate_token_and_get_user(request, error_status_code=True)
|
||||
if error_response:
|
||||
return error_response
|
||||
|
||||
# Check if user has partner role
|
||||
partner_roles = ["partner", "partner_manager", "partner_staff"]
|
||||
if user.role not in partner_roles:
|
||||
return JsonResponse(
|
||||
{"status": "error", "message": "You are not authorized to access this page."},
|
||||
status=403,
|
||||
)
|
||||
|
||||
user_id = data.get("user_id")
|
||||
if not user_id:
|
||||
return JsonResponse(
|
||||
{"status": "error", "message": "user_id is required."},
|
||||
status=400,
|
||||
)
|
||||
|
||||
try:
|
||||
target_user = User.objects.get(id=user_id)
|
||||
except User.DoesNotExist:
|
||||
return JsonResponse(
|
||||
{"status": "error", "message": "User not found."},
|
||||
status=404,
|
||||
)
|
||||
|
||||
# Validate that the user has a partner-related role
|
||||
all_partner_roles = ["partner", "partner_manager", "partner_staff", "partner_customer"]
|
||||
if target_user.role not in all_partner_roles:
|
||||
return JsonResponse(
|
||||
{
|
||||
"status": "error",
|
||||
"message": "User is not a partner-related user. Only users with partner roles can be deleted.",
|
||||
},
|
||||
status=400,
|
||||
)
|
||||
|
||||
# Prevent deleting yourself
|
||||
if target_user.id == user.id:
|
||||
return JsonResponse(
|
||||
{"status": "error", "message": "You cannot delete your own account."},
|
||||
status=400,
|
||||
)
|
||||
|
||||
username = target_user.username
|
||||
target_user.delete()
|
||||
|
||||
return JsonResponse(
|
||||
{
|
||||
"status": "success",
|
||||
"message": f"Partner user '{username}' deleted successfully.",
|
||||
},
|
||||
status=200,
|
||||
)
|
||||
|
||||
except Exception as e:
|
||||
return JsonResponse({"status": "error", "message": str(e)}, status=500)
|
||||
|
||||
|
||||
@method_decorator(csrf_exempt, name="dispatch")
|
||||
class LoginAPI(APIView):
|
||||
"""
|
||||
Admin/Staff Login API (accounts).
|
||||
Body: username (or email), password (required).
|
||||
Returns: token and user details for admin/manager/staff roles.
|
||||
"""
|
||||
|
||||
def post(self, request):
|
||||
try:
|
||||
# Parse JSON or form data
|
||||
is_multipart = request.content_type and "multipart/form-data" in request.content_type
|
||||
if is_multipart:
|
||||
data = request.POST.dict()
|
||||
else:
|
||||
try:
|
||||
data = json.loads(request.body)
|
||||
except json.JSONDecodeError:
|
||||
return JsonResponse(
|
||||
{"status": "error", "message": "Invalid JSON"},
|
||||
status=400,
|
||||
)
|
||||
|
||||
username = data.get("username") or data.get("email")
|
||||
password = data.get("password")
|
||||
|
||||
if not username or not password:
|
||||
return JsonResponse(
|
||||
{"status": "error", "message": "username and password are required."},
|
||||
status=400,
|
||||
)
|
||||
|
||||
user = authenticate(request, username=username, password=password)
|
||||
if not user:
|
||||
return JsonResponse(
|
||||
{"status": "error", "message": "Invalid username or password."},
|
||||
status=401,
|
||||
)
|
||||
|
||||
# Only allow admin/manager/staff to use this login
|
||||
allowed_roles = ["admin", "manager", "staff"]
|
||||
if user.role not in allowed_roles:
|
||||
return JsonResponse(
|
||||
{"status": "error", "message": "You are not authorized to access the admin portal."},
|
||||
status=403,
|
||||
)
|
||||
|
||||
token, _ = Token.objects.get_or_create(user=user)
|
||||
|
||||
return JsonResponse(
|
||||
{
|
||||
"status": "success",
|
||||
"message": "Login successful",
|
||||
"token": token.key,
|
||||
"user": _user_to_dict(user, request),
|
||||
},
|
||||
status=200,
|
||||
)
|
||||
|
||||
except Exception as e:
|
||||
return JsonResponse({"status": "error", "message": str(e)}, status=500)
|
||||
|
||||
|
||||
@method_decorator(csrf_exempt, name="dispatch")
|
||||
class LogoutAPI(APIView):
|
||||
"""
|
||||
Logout API for token-based sessions.
|
||||
Body: token, username (required).
|
||||
"""
|
||||
|
||||
def post(self, request):
|
||||
try:
|
||||
user, token, data, error_response = validate_token_and_get_user(request, error_status_code=True)
|
||||
if error_response:
|
||||
return error_response
|
||||
|
||||
logout(request)
|
||||
token.delete()
|
||||
|
||||
return JsonResponse(
|
||||
{
|
||||
"status": "success",
|
||||
"message": "Logout successful.",
|
||||
},
|
||||
status=200,
|
||||
)
|
||||
|
||||
except Exception as e:
|
||||
return JsonResponse({"status": "error", "message": str(e)}, status=500)
|
||||
|
||||
|
||||
@method_decorator(csrf_exempt, name="dispatch")
|
||||
class UserListAPI(APIView):
|
||||
"""
|
||||
List users (admin / manager / staff only).
|
||||
Body: token, username (required); optional role filter.
|
||||
"""
|
||||
|
||||
def post(self, request):
|
||||
try:
|
||||
user, token, data, error_response = validate_token_and_get_user(request, error_status_code=True)
|
||||
if error_response:
|
||||
return error_response
|
||||
|
||||
# Only allow admin/manager/staff to list users
|
||||
allowed_roles = ["admin", "manager", "staff"]
|
||||
if user.role not in allowed_roles:
|
||||
return JsonResponse(
|
||||
{"status": "error", "message": "You are not authorized to list users."},
|
||||
status=403,
|
||||
)
|
||||
|
||||
qs = User.objects.all().order_by("-id")
|
||||
role_filter = data.get("role")
|
||||
if role_filter:
|
||||
qs = qs.filter(role=role_filter)
|
||||
|
||||
users = [_user_to_dict(u, request) for u in qs]
|
||||
|
||||
return JsonResponse(
|
||||
{
|
||||
"status": "success",
|
||||
"users": users,
|
||||
"total_count": len(users),
|
||||
},
|
||||
status=200,
|
||||
)
|
||||
|
||||
except Exception as e:
|
||||
return JsonResponse({"status": "error", "message": str(e)}, status=500)
|
||||
|
||||
|
||||
@method_decorator(csrf_exempt, name="dispatch")
|
||||
class UserCreateAPI(APIView):
|
||||
"""
|
||||
Create a user (admin / manager / staff only).
|
||||
Body: token, username, new_username, email, password, role ('admin'|'manager'|'staff'), phone_number (optional).
|
||||
"""
|
||||
|
||||
def post(self, request):
|
||||
try:
|
||||
user, token, data, error_response = validate_token_and_get_user(request, error_status_code=True)
|
||||
if error_response:
|
||||
return error_response
|
||||
|
||||
allowed_roles = ["admin", "manager", "staff"]
|
||||
if user.role not in allowed_roles:
|
||||
return JsonResponse(
|
||||
{"status": "error", "message": "You are not authorized to create users."},
|
||||
status=403,
|
||||
)
|
||||
|
||||
new_username = data.get("username") or data.get("new_username")
|
||||
email = data.get("email")
|
||||
password = data.get("password")
|
||||
role = data.get("role")
|
||||
|
||||
if not all([new_username, email, password, role]):
|
||||
return JsonResponse(
|
||||
{
|
||||
"status": "error",
|
||||
"message": "username, email, password, and role are required.",
|
||||
},
|
||||
status=400,
|
||||
)
|
||||
|
||||
valid_roles = ["admin", "manager", "staff"]
|
||||
if role not in valid_roles:
|
||||
return JsonResponse(
|
||||
{
|
||||
"status": "error",
|
||||
"message": f"Invalid role. Must be one of: {', '.join(valid_roles)}",
|
||||
},
|
||||
status=400,
|
||||
)
|
||||
|
||||
if User.objects.filter(username=new_username).exists():
|
||||
return JsonResponse(
|
||||
{"status": "error", "message": "Username already exists."},
|
||||
status=400,
|
||||
)
|
||||
|
||||
if User.objects.filter(email=email).exists():
|
||||
return JsonResponse(
|
||||
{"status": "error", "message": "Email already exists."},
|
||||
status=400,
|
||||
)
|
||||
|
||||
new_user = User.objects.create_user(
|
||||
username=new_username,
|
||||
email=email,
|
||||
password=password,
|
||||
role=role,
|
||||
phone_number=data.get("phone_number"),
|
||||
)
|
||||
|
||||
return JsonResponse(
|
||||
{
|
||||
"status": "success",
|
||||
"message": f"User created successfully with role: {role}.",
|
||||
"user": _user_to_dict(new_user, request),
|
||||
},
|
||||
status=201,
|
||||
)
|
||||
|
||||
except Exception as e:
|
||||
return JsonResponse({"status": "error", "message": str(e)}, status=500)
|
||||
|
||||
|
||||
@method_decorator(csrf_exempt, name="dispatch")
|
||||
class UserUpdateAPI(APIView):
|
||||
"""
|
||||
Update a user (admin / manager / staff only).
|
||||
Body: token, username, user_id (required); email, phone_number, role (optional).
|
||||
"""
|
||||
|
||||
def post(self, request):
|
||||
try:
|
||||
user, token, data, error_response = validate_token_and_get_user(request, error_status_code=True)
|
||||
if error_response:
|
||||
return error_response
|
||||
|
||||
allowed_roles = ["admin", "manager", "staff"]
|
||||
if user.role not in allowed_roles:
|
||||
return JsonResponse(
|
||||
{"status": "error", "message": "You are not authorized to update users."},
|
||||
status=403,
|
||||
)
|
||||
|
||||
user_id = data.get("user_id")
|
||||
if not user_id:
|
||||
return JsonResponse(
|
||||
{"status": "error", "message": "user_id is required."},
|
||||
status=400,
|
||||
)
|
||||
|
||||
try:
|
||||
target_user = User.objects.get(id=user_id)
|
||||
except User.DoesNotExist:
|
||||
return JsonResponse(
|
||||
{"status": "error", "message": "User not found."},
|
||||
status=404,
|
||||
)
|
||||
|
||||
if data.get("email") is not None:
|
||||
new_email = data["email"]
|
||||
if User.objects.filter(email=new_email).exclude(id=user_id).exists():
|
||||
return JsonResponse(
|
||||
{"status": "error", "message": "Email already exists for another user."},
|
||||
status=400,
|
||||
)
|
||||
target_user.email = new_email
|
||||
|
||||
if data.get("phone_number") is not None:
|
||||
target_user.phone_number = data["phone_number"] or None
|
||||
|
||||
if data.get("role") is not None:
|
||||
new_role = data["role"]
|
||||
valid_roles = ["admin", "manager", "staff"]
|
||||
if new_role not in valid_roles:
|
||||
return JsonResponse(
|
||||
{
|
||||
"status": "error",
|
||||
"message": f"Invalid role. Must be one of: {', '.join(valid_roles)}",
|
||||
},
|
||||
status=400,
|
||||
)
|
||||
target_user.role = new_role
|
||||
|
||||
target_user.save()
|
||||
|
||||
return JsonResponse(
|
||||
{
|
||||
"status": "success",
|
||||
"message": "User updated successfully.",
|
||||
"user": _user_to_dict(target_user, request),
|
||||
},
|
||||
status=200,
|
||||
)
|
||||
|
||||
except Exception as e:
|
||||
return JsonResponse({"status": "error", "message": str(e)}, status=500)
|
||||
|
||||
|
||||
@method_decorator(csrf_exempt, name="dispatch")
|
||||
class UserDeleteAPI(APIView):
|
||||
"""
|
||||
Delete a user (admin / manager / staff only).
|
||||
Body: token, username, user_id (required).
|
||||
"""
|
||||
|
||||
def post(self, request):
|
||||
try:
|
||||
user, token, data, error_response = validate_token_and_get_user(request, error_status_code=True)
|
||||
if error_response:
|
||||
return error_response
|
||||
|
||||
allowed_roles = ["admin", "manager", "staff"]
|
||||
if user.role not in allowed_roles:
|
||||
return JsonResponse(
|
||||
{"status": "error", "message": "You are not authorized to delete users."},
|
||||
status=403,
|
||||
)
|
||||
|
||||
user_id = data.get("user_id")
|
||||
if not user_id:
|
||||
return JsonResponse(
|
||||
{"status": "error", "message": "user_id is required."},
|
||||
status=400,
|
||||
)
|
||||
|
||||
try:
|
||||
target_user = User.objects.get(id=user_id)
|
||||
except User.DoesNotExist:
|
||||
return JsonResponse(
|
||||
{"status": "error", "message": "User not found."},
|
||||
status=404,
|
||||
)
|
||||
|
||||
if target_user.id == user.id:
|
||||
return JsonResponse(
|
||||
{"status": "error", "message": "You cannot delete your own account."},
|
||||
status=400,
|
||||
)
|
||||
|
||||
username = target_user.username
|
||||
target_user.delete()
|
||||
|
||||
return JsonResponse(
|
||||
{
|
||||
"status": "success",
|
||||
"message": f"User '{username}' deleted successfully.",
|
||||
},
|
||||
status=200,
|
||||
)
|
||||
|
||||
except Exception as e:
|
||||
return JsonResponse({"status": "error", "message": str(e)}, status=500)
|
||||
|
||||
Reference in New Issue
Block a user