### Django Rest Framework Libs ### from rest_framework import viewsets, permissions, filters, status from rest_framework.decorators import action from rest_framework.response import Response from rest_framework.permissions import IsAuthenticated from rest_framework_simplejwt.authentication import JWTAuthentication from rest_framework_simplejwt.tokens import RefreshToken from django.contrib.auth import authenticate from rest_framework.parsers import MultiPartParser, FormParser ### Django Libs ### from django.conf import settings from django_filters import rest_framework as django_filters from django.contrib.auth.hashers import check_password import time from bson import ObjectId import os from django.core.files.base import ContentFile from django.core.files.storage import default_storage from drf_yasg.utils import swagger_auto_schema from drf_yasg import openapi ### App Libs ### from .utils import build_aggregation_pipeline, build_absolute_uri from .models import * from .serializers import * from .permissions import * from auth_app.models import JWTToken class CustomViewSet(viewsets.ModelViewSet): permission_classes = [IsAuthenticated, HasNivelPermission] authentication_classes = [JWTAuthentication] def get_required_permissions(self): action_permissions = { 'list': 'view', 'retrieve': 'view', 'create': 'add', 'update': 'change', 'partial_update': 'change', 'destroy': 'delete', } model_name = self.model._meta.object_name.lower() action = action_permissions.get(self.action, '') codename = f"{action}_{model_name}" if action else '' #print(f"Action: {self.action}") app_name = self.model._meta.app_label return (app_name, codename) def handle_uploaded_file(self, file, instance): # Ensure the directory exists directory = f'{instance.__class__.__name__.lower()}s/img/{instance.id}/' os.makedirs(directory, exist_ok=True) file_path = default_storage.save(os.path.join(directory, file.name), ContentFile(file.read())) if file_path: media = settings.MEDIA_URL return file_path.split(media)[-1] def apply_field_filters(self, data): """ Filter fields based on exclude_fields and query parameters. Exclude specified fields and include only fields listed in query parameters if provided. """ try: # Ensure the data is a list of dictionaries if not isinstance(data, list) or not all(isinstance(item, dict) for item in data): raise ValueError("Data should be a list of dictionaries") # Exclude specified fields if hasattr(self, 'exclude_fields'): data = [{k: v for k, v in item.items() if k not in self.exclude_fields} for item in data] # Include only fields listed in query parameters if provided if 'fields' in self.request.query_params: fields = self.request.query_params['fields'].split(',') print(f"Fields: {fields}") # Prepare a structure to handle nested fields nested_fields = {} for field in fields: if '__' in field: parent, child = field.split('__', 1) if parent not in nested_fields: nested_fields[parent] = [] nested_fields[parent].append(child) else: nested_fields[field] = None def filter_dict(d, keys): """ Recursively filter dictionary based on the given keys. """ result = {} for k, v in d.items(): if k in keys: if isinstance(v, dict) and nested_fields[k]: result[k] = filter_dict(v, nested_fields[k]) elif isinstance(v, list) and nested_fields[k]: result[k] = [filter_dict(item, nested_fields[k]) for item in v] else: result[k] = v return result # Apply the filtering logic data = [filter_dict(item, nested_fields) for item in data] return data except Exception as e: print(f"Error applying field filters: {e}") return data def get_queryset(self): pipeline, pipeline_filter = build_aggregation_pipeline(self.request, self.filter_fields) results = list(self.model.objects.mongo_aggregate(pipeline)) total = list(self.model.objects.mongo_aggregate(pipeline_filter)) ids = [r['_id'] for r in results] queryset = self.model.objects.filter(id__in=ids) queryset.total = total[0]['total'] if total else 0 return queryset def retrieve(self, request, pk=None): ### method for retrieving a single item try: from bson import ObjectId serializer = self.serializer_class(self.queryset.get(id=ObjectId(pk))) #print(f"Serializer Data: {serializer.data}") if not serializer: return Response({'message': 'No data found'}, status=status.HTTP_400_BAD_REQUEST) if serializer: filtered_data = self.apply_field_filters([serializer.data])[0] return Response(filtered_data, status=status.HTTP_200_OK) except Exception as e: return Response({'message': f'Item {pk} not found => {e}'}, status=status.HTTP_200_OK) def create(self, request): print(f"Request Data: {request.data}") try: serializer = self.serializer_class(data=request.data) if serializer.is_valid(): serializer.save() print(f"Serializer Data: {serializer.data}") filtered_data = self.apply_field_filters([serializer.data])[0] return Response({'message':'Data created successfully!', 'results': filtered_data}, status=status.HTTP_201_CREATED) return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST) except Exception as e: return Response({'message': f'Error creating item => {e}'}, status=status.HTTP_400_BAD_REQUEST) def list(self, request): try: start_time = time.time() queryset = self.get_queryset() print(f"Queryset: {queryset}") try: serializer = self.get_serializer(queryset, many=True) try: default_page_size = settings.REST_FRAMEWORK.get('PAGE_SIZE', 10) page_size = int(request.query_params.get('page_size', default_page_size)) page = int(request.query_params.get('page', 1)) total = queryset.total total_per_page = len(queryset) total_pages = (total // page_size) + (1 if total % page_size > 0 else 0) if page >= total_pages: next_page = None else: next_page = build_absolute_uri(request, {'page': page + 1}) if page <= 1: previous_page = None else: previous_page = build_absolute_uri(request, {'page': page - 1}) filtered_data = self.apply_field_filters(serializer.data) response = { 'count': total, 'num_pages': total_pages, 'current_count': total_per_page, 'next': next_page, 'previous': previous_page, 'response_time': time.time() - start_time, 'results': filtered_data } # Remove empty or None values from the response response = {key: value for key, value in response.items() if value is not None and value != ''} if len(serializer.data) > 0: return Response(response, status=status.HTTP_200_OK) else: return Response({'message': 'No item found'}, status=status.HTTP_200_OK) except Exception as e: return Response({'message': f'Error generating response => {e}'}, status=status.HTTP_400_BAD_REQUEST) except Exception as e: return Response({'message': 'No Item found'}, status=status.HTTP_200_OK) except Exception as e: response = {'message': 'Error listing items', 'Error': str(e)} return Response(response, status=status.HTTP_400_BAD_REQUEST) def destroy(self, request, pk=None): ### method for deleting a single item try: instance = self.queryset.get(id=ObjectId(pk)) instance.delete() return Response(status=status.HTTP_204_NO_CONTENT) except Exception as e: return Response({'message': f'Error deleting item => {e}'}, status=status.HTTP_400_BAD_REQUEST) class NivelViewSet(CustomViewSet): model = Nivel queryset = Nivel.objects.all() serializer_class = NivelSerializer #permission_classes = [IsAuthenticated, HasNivelPermission] #authentication_classes = [JWTAuthentication] #filter_backends = [django_filters.DjangoFilterBackend, filters.SearchFilter, filters.OrderingFilter] #search_fields='__all__' #ordering_fields = '__all__' exclude_fields = [] filter_fields = { 'nivel_id': ['in', 'exact'], 'nome': ['icontains', 'exact'], } @swagger_auto_schema( operation_description="""List all Niveis. **httpie Example**: http -A bearer -a $token -f GET $http_domain/accounts/niveis/ """, operation_id='list', operation_summary='List all Niveis', tags=['Niveis'], responses={200: openapi.Response('Data successfully retrieved', NivelSerializer), 400: openapi.Response('Error Generating Response', NivelSerializer)} ) def list(self, request): return super().list(request) @swagger_auto_schema( operation_description="""Retrieve a Nivel. **httpie Example**: http -A bearer -a $token -f GET $http_domain/accounts/niveis/66743218d8236dbea8742435/ """, operation_id='retrieve', operation_summary='Retrieve a Nivel', tags=['Niveis'], responses={200: openapi.Response('Data successfully retrieved', NivelSerializer), 400: openapi.Response('Error generating Response', NivelSerializer)} ) def retrieve(self, request, pk=None): return super().retrieve(request, pk) @swagger_auto_schema( operation_description="""Delete a Nivel. **httpie Example**: http -A bearer -a $token -f DELETE $http_domain/accounts/niveis/66743218d8236dbea8742435/ """, operation_id='destroy', operation_summary='Delete a Nivel', tags=['Niveis'], responses={204: openapi.Response(status.HTTP_204_NO_CONTENT, NivelSerializer), 400: openapi.Response('Error deleting item', NivelSerializer)} ) def destroy(self, request, pk=None): return super().destroy(request, pk) name_param = openapi.Parameter('name', openapi.IN_PATH, description="Nivel name", type=openapi.TYPE_STRING, required=True) permissions_param = openapi.Parameter('permissions', openapi.IN_PATH, description="Nivel permissions", type=openapi.TYPE_ARRAY,required=True, items=openapi.Items(type=openapi.TYPE_STRING)) manual_parameters = [name_param,permissions_param] @swagger_auto_schema( operation_description="""Create Nivel Item. **httpie Example**: http -A bearer -a $token POST $http_domain/accounts/niveis/ name="Acesso Total Universidade" permissions:=['{"app_label":"panel","codename":"add_universidade","name": "Can add Universidade"},{"app_label":"panel","codename":"change_universidade","name":"Can change Universidade"},{"app_label":"panel","codename":"delete_universidade","name":"Can delete Universidade"},{"app_label":"panel","codename":"view_universidade","name":"Can view Universidade"}'] """, operation_id='create', operation_summary='Create Nivel Item', manual_parameters=manual_parameters, tags=['Niveis'], responses={201: openapi.Response('Item created successfully', NivelSerializer), 400: openapi.Response('Erro creating item', NivelSerializer)} ) def create(self, request): try: serializer = NivelOnlyWriteSerializer(data=request.data) if serializer.is_valid(): serializer.save() serializer = self.serializer_class(self.queryset.get(id=ObjectId(serializer.data['id']))) filtered_data = self.apply_field_filters([serializer.data])[0] return Response({'message':'Item created successfully!', 'results': filtered_data}, status=status.HTTP_201_CREATED) return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST) except Exception as e: return Response({'message': f'Error creating item => {e}'}, status=status.HTTP_400_BAD_REQUEST) @swagger_auto_schema( operation_description="""Update Nivel Item. **httpie Example**: http -A bearer -a $token PUT $http_domain/accounts/niveis/66743218d8236dbea8742435/ name="Acesso Total Universidade 2" permissions:=['{"app_label":"panel","codename":"add_universidade","name": "Can add Universidade"},{"app_label":"panel","codename":"change_universidade","name":"Can change Universidade"},{"app_label":"panel","codename":"delete_universidade","name":"Can delete Universidade"},{"app_label":"panel","codename":"view_universidade","name":"Can view Universidade"},{"app_label":"panel","codename":"view_remaxnomundo","name":"Can view RE/MAX no Mundo"},{"app_label": "accounts","codename":"view_usuario","name":"Can view Usuario"}'] """, operation_id='update', operation_summary='Update Nivel Item', tags=['Niveis'], responses={200: openapi.Response('Data successfully updated', NivelSerializer), 400: openapi.Response('Error updating data', NivelSerializer), } ) def update(self, request, pk=None): try: instance = self.queryset.get(id=ObjectId(pk)) serializer = NivelOnlyWriteSerializer(instance, data=request.data, partial=True) if serializer.is_valid(): serializer.save() serializer = self.serializer_class(self.queryset.get(id=ObjectId(pk))) filtered_data = self.apply_field_filters([serializer.data])[0] return Response({'message':'Data successfully updated','results':filtered_data}, status=status.HTTP_200_OK) return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST) except Exception as e: return Response({'message': f'Error updating item => {e}'}, status=status.HTTP_400_BAD_REQUEST) @swagger_auto_schema( operation_description="""Partial Update Nivel Item. **httpie Example**: http -A bearer -a $token PATCH $http_domain/accounts/niveis/66743218d8236dbea8742435/ name="Acesso Total Universidade 2" permissions:=['{"app_label":"panel","codename":"add_universidade","name": "Can add Universidade"},{"app_label":"panel","codename":"change_universidade","name":"Can change Universidade"},{"app_label":"panel","codename":"delete_universidade","name":"Can delete Universidade"},{"app_label":"panel","codename":"view_universidade","name":"Can view Universidade"},{"app_label":"panel","codename":"view_remaxnomundo","name":"Can view RE/MAX no Mundo"},{"app_label": "accounts","codename":"view_usuario","name":"Can view Usuario"}'] """, operation_id='partial_update', operation_summary='Partial Update Nivel Item', tags=['Niveis'], responses={200: openapi.Response('Data successfully updated', NivelSerializer), 400: openapi.Response('Error updating data', NivelSerializer), } ) def partial_update(self, request, *args, **kwargs): try: instance = self.queryset.get(id=ObjectId(pk)) serializer = NivelOnlyWriteSerializer(instance, data=request.data, partial=True) if serializer.is_valid(): serializer.save() serializer = self.serializer_class(self.queryset.get(id=ObjectId(pk))) filtered_data = self.apply_field_filters([serializer.data])[0] return Response({'message':'Data successfully updated','results':filtered_data}, status=status.HTTP_200_OK) return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST) except Exception as e: return Response({'message': f'Error updating item => {e}'}, status=status.HTTP_400_BAD_REQUEST) class UsuarioViewSet(CustomViewSet): model = Usuario queryset = model.objects.all() serializer_class = UsuarioSerializer #permission_classes = [IsAuthenticated] #authentication_classes = [JWTAuthentication] #filter_backends = [django_filters.DjangoFilterBackend, filters.SearchFilter, filters.OrderingFilter] #search_fields='__all__' #ordering_fields = '__all__' parser_classes = [MultiPartParser, FormParser] exclude_fields = ['password', 'is_superuser'] filter_fields = { 'id': ['in', 'exact'], 'first_name': ['icontains', 'exact'], 'last_name': ['icontains', 'exact'], 'email': ['exact'], 'telefone': ['exact'], 'is_active': ['exact'], 'is_staff': ['exact'], 'is_superuser': ['exact'], 'nivel__nivel_id': ['in', 'exact'], 'categoria': ['icontains','exact'], } @swagger_auto_schema( operation_description="""Retrieve a user. **httpie Example**: http -A bearer -a $token -f GET $http_domain/accounts/usuarios/667373482c19a5c7a2f9c9e4/ """, operation_id='retrieve', operation_summary='Retrieve a user', tags=['Usuarios'], ) def retrieve(self, request, pk=None): return super().retrieve(request, pk) @swagger_auto_schema( operation_description="""List all users. **httpie Example**: http -A bearer -a $token -f GET $http_domain/accounts/usuarios/ """, operation_id='list', operation_summary='List all users', tags=['Usuarios'], ) def list(self, request): return super().list(request) @swagger_auto_schema( operation_description="""Delete a user. **httpie Example**: http -A bearer -a $token -f DELETE $http_domain/accounts/usuarios/667373482c19a5c7a2f9c9e4/ """, operation_id='destroy', operation_summary='Delete a user', tags=['Usuarios'], ) def destroy(self, request, pk=None): return super().destroy(request, pk) @swagger_auto_schema( operation_description="""Create a new user. **httpie Example**: http -A bearer -a $token -f POST $http_domain/accounts/usuarios/ email='usuario30@teste.com' password='123456' categoria='admin' http -A bearer -a $token -f POST $http_domain/accounts/usuarios/ email='usuario31@teste.com' password='123456' categoria='admin' first_name='Usuario Teste 31' telefone='123456789' foto@'/home/felipe/pixerama/remax/api/PortalREMax/files/categorias/img/6671c9512e992b0cb3339549/logo_boulevard.png' nivel='6668ad8f50feaaeb415bfa5d' """, operation_id='create', operation_summary='Create a new user', tags=['Usuarios'], ) def create(self, request): print(f"Request Data: {request.data}") obj = self.model.objects.mongo_find_one({'email': request.data['email']}) if obj: return Response({'message': 'Este e-mail já está cadastrado!'}, status=status.HTTP_400_BAD_REQUEST) else: try: data = self.request.data.copy() # Create a mutable copy of request data if 'foto' in data and data['foto'] is not None: data.pop('foto') if 'password' in data: data['password'] = make_password(data['password']) serializer = UsuarioOnlyWriteSerializer(data=data, partial=True) if serializer.is_valid(): serializer.save() instance = self.queryset.get(id=ObjectId(serializer.data['id'])) data = request.data.copy() # Create a mutable copy of request data # Handle file upload if 'foto' in request.FILES: if not request.FILES['foto'].content_type.startswith('image'): return Response({'message': 'Invalid file type. Only images are allowed.'}, status=status.HTTP_400_BAD_REQUEST) file_url = self.handle_uploaded_file(request.FILES['foto'], instance) data['foto'] = file_url # Update data with the file URL serializer = UsuarioOnlyWriteSerializer(instance, data=data, partial=True) if serializer.is_valid(): serializer.save() instance = self.queryset.get(id=ObjectId(serializer.data['id'])) serializer = UsuarioSerializer(instance) filtered_data = self.apply_field_filters([serializer.data])[0] return Response({'message':'Data Successfully created!', 'results':filtered_data}, status=status.HTTP_201_CREATED) return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST) except Exception as e: return Response({'message': f'Error creating item => {e}'}, status=status.HTTP_400_BAD_REQUEST) @swagger_auto_schema( operation_description="""Update a user. **httpie Example**: http -A bearer -a $token -f PUT $http_domain/accounts/usuarios/66737d260784bf2b096bba01/ email='usuario35@teste.com' password='123456' categoria='admin' first_name='Usuario Teste 35' telefone='12345678910' foto@'/home/felipe/pixerama/remax/api/PortalREMax/files/usuarios/fotos/curly_hair.jpg' nivel='6668ad8f50feaaeb415bfa5d' """, operation_id='update', operation_summary='Update a user', tags=['Usuarios'], responses={200: openapi.Response('Data successfully updated', UsuarioSerializer), 400: openapi.Response('Invalid data', UsuarioSerializer), 400: openapi.Response('Error Generating Response', UsuarioSerializer)} ) def update(self, request, pk=None): ### method for updating a single item try: instance = self.queryset.get(id=ObjectId(pk)) data = request.data.copy() # Create a mutable copy of request data # Handle file upload if 'foto' in request.FILES: print(f"File: {request.FILES['foto']}") if not request.FILES['foto'].content_type.startswith('image'): return Response({'message': 'Invalid file type. Only images are allowed.'}, status=status.HTTP_400_BAD_REQUEST) file_url = self.handle_uploaded_file(request.FILES['foto'], instance) data['foto'] = file_url # Update data with the file URL if 'password' in data: data['password'] = make_password(data['password']) serializer = UsuarioOnlyWriteSerializer(instance, data=data, partial=True) if serializer.is_valid(): serializer.save() instance = self.queryset.get(id=ObjectId(pk)) serializer = UsuarioSerializer(instance) filtered_data = self.apply_field_filters([serializer.data])[0] return Response({'message':'Data successfully updated','results':filtered_data}, status=status.HTTP_200_OK) return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST) except Exception as e: return Response({'message': f'Error updating item2 => {e}'}, status=status.HTTP_400_BAD_REQUEST) @swagger_auto_schema( operation_description="""Update a user. **httpie Example**: http -A bearer -a $token -f PUT $http_domain/accounts/usuarios/66737d260784bf2b096bba01/ email='usuario35@teste.com' password='123456' categoria='admin' first_name='Usuario Teste 35' telefone='12345678910' foto@'/home/felipe/pixerama/remax/api/PortalREMax/files/usuarios/fotos/curly_hair.jpg' nivel='6668ad8f50feaaeb415bfa5d' """, operation_id='partial_update', operation_summary='Partial Update a user', tags=['Usuarios'], responses={200: openapi.Response('Data successfully updated', UsuarioSerializer), 400: openapi.Response('Error Updating item', UsuarioSerializer)} ) def partial_update(self, request, pk=None): ### method for partially updating a single item try: instance = self.queryset.get(id=ObjectId(pk)) data = request.data.copy() # Create a mutable copy of request data # Handle file upload if 'foto' in request.FILES: if not request.FILES['foto'].content_type.startswith('image'): return Response({'message': 'Invalid file type. Only images are allowed.'}, status=status.HTTP_400_BAD_REQUEST) file_url = self.handle_uploaded_file(request.FILES['foto'], instance) data['foto'] = file_url # Update data with the file URL if 'password' in data: data['password'] = make_password(data['password']) serializer = UsuarioOnlyWriteSerializer(instance, data=data, partial=True) if serializer.is_valid(): serializer.save() instance = self.queryset.get(id=ObjectId(pk)) serializer = UsuarioSerializer(instance) filtered_data = self.apply_field_filters([serializer.data])[0] return Response({'message':'Data successfully updated','results':filtered_data}, status=status.HTTP_200_OK) return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST) except Exception as e: return Response({'message': f'Error updating item => {e}'}, status=status.HTTP_400_BAD_REQUEST) @swagger_auto_schema( operation_description="""Create a new client user. **httpie Example**: http -A bearer -a $token -f POST $http_domain/accounts/usuarios/create_client/ email='usuario30@teste.com' password='123456' categoria='cliente' http -A bearer -a $token -f POST $http_domain/accounts/usuarios/ email='usuario31@teste.com' password='123456' categoria='cliente' first_name='Usuario Teste 31' telefone='123456789' foto@'/home/felipe/pixerama/remax/api/PortalREMax/files/categorias/img/6671c9512e992b0cb3339549/logo_boulevard.png' nivel='6668ad8f50feaaeb415bfa5d' """, operation_id='create', operation_summary='Create a new user', tags=['Usuarios'], ) @action(detail=False, methods=['post'], url_path='create_client', url_name='create_client', permission_classes=[permissions.AllowAny], authentication_classes=[]) def create_client(self, request): print(f"Request Data: {request.data}") obj = self.model.objects.mongo_find_one({'email': request.data['email']}) if obj: return Response({'message': 'Este e-mail já está cadastrado!'}, status=status.HTTP_400_BAD_REQUEST) else: try: data = self.request.data.copy() # Create a mutable copy of request data data['categoria'] = 'cliente' if 'foto' in data and data['foto'] is not None: data.pop('foto') if 'password' in data: data['password'] = make_password(data['password']) serializer = UsuarioOnlyWriteSerializer(data=data, partial=True) if serializer.is_valid(): serializer.save() instance = self.queryset.get(id=ObjectId(serializer.data['id'])) data = request.data.copy() # Create a mutable copy of request data # Handle file upload if 'foto' in request.FILES: if not request.FILES['foto'].content_type.startswith('image'): return Response({'message': 'Invalid file type. Only images are allowed.'}, status=status.HTTP_400_BAD_REQUEST) file_url = self.handle_uploaded_file(request.FILES['foto'], instance) data['foto'] = file_url # Update data with the file URL serializer = UsuarioOnlyWriteSerializer(instance, data=data, partial=True) if serializer.is_valid(): serializer.save() instance = self.queryset.get(id=ObjectId(serializer.data['id'])) serializer = UsuarioSerializer(instance) filtered_data = self.apply_field_filters([serializer.data])[0] return Response({'message':'Data Successfully created!', 'results':filtered_data}, status=status.HTTP_201_CREATED) return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST) except Exception as e: return Response({'message': f'Error creating item => {e}'}, status=status.HTTP_400_BAD_REQUEST) ### create view my user, to retrieve, update, partial_update and destroy only the authenticated user class MyUserViewSet(CustomViewSet): model = Usuario queryset = model.objects.all() serializer_class = UsuarioSerializer permission_classes = [IsAuthenticated] authentication_classes = [JWTAuthentication] exclude_fields = ['password', 'is_superuser'] filter_fields = { 'id': ['in', 'exact'], 'first_name': ['icontains', 'exact'], 'last_name': ['icontains', 'exact'], 'email': ['exact'], 'telefone': ['exact'], 'is_active': ['exact'], 'is_staff': ['exact'], 'is_superuser': ['exact'], 'nivel__nivel_id': ['in', 'exact'], 'categoria': ['icontains','exact'], } ### list method not alloed for my user def list(self, request): return Response({'message': 'Method not allowed'}, status=status.HTTP_405_METHOD_NOT_ALLOWED) def retrieve(self, request, pk=None): return Response({'message': 'Method not allowed'}, status=status.HTTP_405_METHOD_NOT_ALLOWED) def create(self, request): return Response({'message': 'Method not allowed'}, status=status.HTTP_405_METHOD_NOT_ALLOWED) def update(self, request, pk=None): return Response({'message': 'Method not allowed'}, status=status.HTTP_405_METHOD_NOT_ALLOWED) def partial_update(self, request, pk=None): return Response({'message': 'Method not allowed'}, status=status.HTTP_405_METHOD_NOT_ALLOWED) @swagger_auto_schema( operation_description="""Retrieve my user. **httpie Example**: http -A bearer -a $token -f GET $http_domain/accounts/meu-usuario/retrieve/ """, operation_id='retrieve', operation_summary='Retrieve my user', tags=['My User'], ) @action(detail=False, methods=['get'] ,url_path='retrieve', url_name='retrieve', permission_classes=[IsAuthenticated], authentication_classes=[JWTAuthentication]) def retrieve_meu_usuario(self, request, pk=None): if request.user.id: pk = request.user.id return super().retrieve(request, pk) else: return Response({'message': 'User not authenticated'}, status=status.HTTP_400_BAD_REQUEST) @swagger_auto_schema( operation_description="""Update my user. **httpie Example**: http -A bearer -a $token -f PUT $http_domain/accounts/meu-usuario/update """, operation_id='update', operation_summary='Update my user', tags=['Meu Usuario'], ) @action(detail=False, methods=['put'], url_path='update', url_name='update', permission_classes=[IsAuthenticated], authentication_classes=[JWTAuthentication]) def update_meu_usuario(self, request): if request.user.id: pk = request.user.id try: instance = self.queryset.get(id=pk) data = request.data.copy() # Create a mutable copy of request data # Handle file upload if 'foto' in request.FILES: if not request.FILES['foto'].content_type.startswith('image'): return Response({'message': 'Invalid file type. Only images are allowed.'}, status=status.HTTP_400_BAD_REQUEST) file_url = self.handle_uploaded_file(request.FILES['foto'], instance) data['foto'] = file_url # Update data with the file URL if 'password' in data: data['password'] = make_password(data['password']) serializer = UsuarioOnlyWriteSerializer(instance, data=data, partial=True) if serializer.is_valid(): serializer.save() instance = self.queryset.get(id=pk) serializer = UsuarioSerializer(instance) filtered_data = self.apply_field_filters([serializer.data])[0] return Response({'message':'Data successfully updated','results':filtered_data}, status=status.HTTP_200_OK) return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST) except Exception as e: return Response({'message': f'Error updating item => {e}'}, status=status.HTTP_400_BAD_REQUEST) else: return Response({'message': 'User not authenticated'}, status=status.HTTP_400_BAD_REQUEST) @swagger_auto_schema( operation_description="""Partial Update my user. **httpie Example**: http -A bearer -a $token -f PATCH $http_domain/accounts/meu-usuario/partial/ """, operation_id='partial_update', operation_summary='Partial Update my user', tags=['My User'], ) @action(detail=False, methods=['patch'], url_path='partial', url_name='partial', permission_classes=[IsAuthenticated], authentication_classes=[JWTAuthentication]) def partial_meu_usuario(self, request): if request.user.id: pk = request.user.id try: instance = self.queryset.get(id=pk) data = request.data.copy() # Create a mutable copy of request data # Handle file upload if 'foto' in request.FILES: if not request.FILES['foto'].content_type.startswith('image'): return Response({'message': 'Invalid file type. Only images are allowed.'}, status=status.HTTP_400_BAD_REQUEST) file_url = self.handle_uploaded_file(request.FILES['foto'], instance) data['foto'] = file_url # Update data with the file URL if 'password' in data: data['password'] = make_password(data['password']) serializer = UsuarioOnlyWriteSerializer(instance, data=data, partial=True) if serializer.is_valid(): serializer.save() instance = self.queryset.get(id=pk) serializer = UsuarioSerializer(instance) filtered_data = self.apply_field_filters([serializer.data])[0] return Response({'message':'Data successfully updated','results':filtered_data}, status=status.HTTP_200_OK) return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST) except Exception as e: return Response({'message': f'Error updating item => {e}'}, status=status.HTTP_400_BAD_REQUEST) else: return Response({'message': 'User not authenticated'}, status=status.HTTP_400_BAD_REQUEST) @swagger_auto_schema( operation_description="""Delete my user. **httpie Example**: http -A bearer -a $token -f DELETE $http_domain/accounts/meu-usuario/delete/ """, operation_id='destroy', operation_summary='Delete my user', tags=['My User'], ) def destroy(self, request, pk=None): if request.user.id: pk = request.user.id return super().destroy(request, pk) else: return Response({'message': 'User not authenticated'}, status=status.HTTP_400_BAD_REQUEST) # class AdminViewSet(CustomViewSet): # model = Usuario # queryset = model.objects.filter(categoria='admin') # serializer_class = UsuarioSerializer # #permission_classes = [IsAuthenticated] # #authentication_classes = [JWTAuthentication] # filter_backends = [django_filters.DjangoFilterBackend, filters.SearchFilter, filters.OrderingFilter] # search_fields='__all__' # ordering_fields = '__all__' # exclude_fields = ['password', 'is_superuser'] # filter_fields = { # 'id': ['in', 'exact'], # 'first_name': ['icontains', 'exact'], # 'last_name': ['icontains', 'exact'], # 'email': ['exact'], # 'telefone': ['exact'], # 'is_active': ['exact'], # 'is_staff': ['exact'], # 'is_superuser': ['exact'], # 'nivel__nivel_id': ['in', 'exact'], # 'categoria': ['icontains','exact'], # } # def get_queryset(self): # pipeline, pipeline_filter = build_aggregation_pipeline(self.request, self.filter_fields) # results = list(self.model.objects.mongo_aggregate(pipeline)) # total = list(self.model.objects.mongo_aggregate(pipeline_filter)) # ids = [r['_id'] for r in results] # queryset = self.model.objects.filter(id__in=ids, categoria='admin') # queryset.total = total[0]['total'] if total else 0 # return queryset # class ClienteViewSet(CustomViewSet): # model = Usuario # queryset = model.objects.filter(categoria='cliente') # serializer_class = UsuarioSerializer # #permission_classes = [IsAuthenticated] # #authentication_classes = [JWTAuthentication] # filter_backends = [django_filters.DjangoFilterBackend, filters.SearchFilter, filters.OrderingFilter] # search_fields='__all__' # ordering_fields = '__all__' # exclude_fields = ['password', 'is_superuser'] # pre_filter = {'categoria': 'cliente'} # filter_fields = { # 'id': ['in', 'exact'], # 'first_name': ['icontains', 'exact'], # 'last_name': ['icontains', 'exact'], # 'email': ['exact'], # 'telefone': ['exact'], # 'is_active': ['exact'], # 'is_staff': ['exact'], # 'is_superuser': ['exact'], # 'nivel__nivel_id': ['in', 'exact'], # 'categoria': ['icontains','exact'], # } ### create filter class for permissions class PermissionFilter(django_filters.FilterSet): class Meta: model = Permission fields = { 'id': ['in', 'exact'], 'name': ['icontains', 'exact'], 'codename': ['icontains', 'exact'], 'content_type__app_label': ['icontains', 'exact'], 'content_type__model': ['icontains', 'exact'], } from django.contrib.auth.models import Permission class PermissionViewSet(viewsets.ReadOnlyModelViewSet): permission_classes = [IsAuthenticated] # authentication_classes = [JWTAuthentication] queryset = Permission.objects.filter(content_type__app_label__in=['auth_app', 'accounts', 'panel']) serializer_class = PermissionSerializer filter_backends = [django_filters.DjangoFilterBackend, filters.OrderingFilter, filters.SearchFilter] filterset_class = PermissionFilter search_fields = ['name', 'codename', 'content_type__app_label', 'content_type__model'] ordering_fields = ['id', 'name', 'codename'] def list(self, request): try: permissions = self.filter_queryset(self.get_queryset()) serializer = self.get_serializer(permissions, many=True) if permissions: return Response(serializer.data, status=status.HTTP_200_OK) else: return Response({'message': 'Not allowed'}, status=status.HTTP_401_UNAUTHORIZED) except Exception as e: response = {'message': 'Error listing items', 'Error': str(e)} return Response(response, status=status.HTTP_400_BAD_REQUEST) @swagger_auto_schema(responses={405: 'Method not allowed'}, auto_schema=None) def create(self, request): return Response({'message': 'Method not allowed'}, status=status.HTTP_405_METHOD_NOT_ALLOWED) @swagger_auto_schema(responses={405: 'Method not allowed'}, auto_schema=None) def retrieve(self, request, pk=None): return Response({'message': 'Method not allowed'}, status=status.HTTP_405_METHOD_NOT_ALLOWED) @swagger_auto_schema(responses={405: 'Method not allowed'}, auto_schema=None) def update(self, request, pk=None): return Response({'message': 'Method not allowed'}, status=status.HTTP_405_METHOD_NOT_ALLOWED) @swagger_auto_schema(responses={405: 'Method not allowed'}, auto_schema=None) def partial_update(self, request, pk=None): return Response({'message': 'Method not allowed'}, status=status.HTTP_405_METHOD_NOT_ALLOWED) @swagger_auto_schema(responses={405: 'Method not allowed'}, auto_schema=None) def destroy(self, request, pk=None): return Response({'message': 'Method not allowed'}, status=status.HTTP_405_METHOD_NOT_ALLOWED)