Sicherhaven
384797551f
- Add eventify_id CharField (unique, indexed, editable=False) to User - Auto-generate on save() with charset excluding I/O/0/1 for clarity - Migration 0012: add field nullable, backfill all existing users, make non-null - Sync migration 0011 (allowed_modules) pulled from server - Expose eventify_id in accounts/api.py, partner/api.py serializers - Expose eventify_id in mobile_api login response (populates localStorage)
948 lines
34 KiB
Python
948 lines
34 KiB
Python
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",
|
|
"eventify_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"] = 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)
|
|
|