from rest_framework import viewsets, status, filters from rest_framework.response import Response from rest_framework.decorators import action from .models import * from .serializers import * from django_filters import rest_framework as django_filters from rest_framework.pagination import PageNumberPagination import json import logging import datetime import decimal from drf_yasg.utils import swagger_auto_schema from drf_yasg import openapi import time import urllib.parse from django.db.models import Q from haversine import haversine, Unit from rest_framework.permissions import AllowAny, IsAuthenticatedOrReadOnly, IsAdminUser #from rest_framework.authentication import TokenAuthentication #from rest_framework.permissions import AllowAny, IsAuthenticated, IsAuthenticatedOrReadOnly, IsAdminUser #from rest_framework_simplejwt.authentication import JWTAuthentication #from django.conf import settings # Create your views here. logger = logging.getLogger(__name__) #Initializing the logging object. class SetPagination(PageNumberPagination): page_query_param = 'page' page_size_query_param = 'page_size' page_size = 20 max_page_size = 10000 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(',') # Prepare a structure to handle nested fields nested_fields = {} for field in fields: parts = field.split('__') current = nested_fields for part in parts[:-1]: current = current.setdefault(part, {}) current[parts[-1]] = 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 the value is a dictionary and we have nested fields, recurse if isinstance(v, dict) and keys[k]: result[k] = filter_dict(v, keys[k]) # If the value is a list, handle lists of dictionaries or other nested structures elif isinstance(v, list) and keys[k]: # Recursively apply filtering for each dictionary in the list result[k] = [filter_dict(item, keys[k]) if isinstance(item, dict) else item 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}") logger.error(f"Error applying field filters: {e}") return data def filter_queryset_dict(queryset_dict, exclude_fields=None): def filter_dict(data): filtered_data = {} for key, value in data.items(): if isinstance(value, dict): filtered_data[key] = filter_dict(value) elif value is not None and value != '' and key != '_state' and key not in exclude_fields: if isinstance(value, (datetime.datetime, datetime.date)): filtered_data[key] = value.strftime('%Y-%m-%d %H:%M:%S') elif isinstance(value, decimal.Decimal): filtered_data[key] = float(value) else: filtered_data[key] = value return filtered_data try: filtered_results = [] if exclude_fields is None: exclude_fields = [] if isinstance(queryset_dict, list): for item in queryset_dict: filtered_item = filter_dict(item) # Additional processing for list response for key in list(filtered_item.keys()): if filtered_item[key] is None or filtered_item[key] == '' or key == '_state' or key in exclude_fields: del filtered_item[key] filtered_results.append(filtered_item) else: filtered_results = filter_dict(queryset_dict) json_data = json.dumps(filtered_results) decoded_data = json.loads(json_data) return decoded_data except Exception as e: return e def filter_data_dict(serializer, exclude_fields=None): def filter_dict(data): filtered_data = {} for key, value in data.items(): if isinstance(value, dict): filtered_data[key] = filter_dict(value) elif value is not None and value != '' and key != '_state' and key not in exclude_fields: if isinstance(value, (datetime.datetime, datetime.date)): filtered_data[key] = value.strftime('%Y-%m-%d %H:%M:%S') elif isinstance(value, decimal.Decimal): filtered_data[key] = float(value) else: filtered_data[key] = value return filtered_data try: filtered_results = [] if exclude_fields is None: exclude_fields = [] if isinstance(serializer.data, list): for item in serializer.data: filtered_item = filter_dict(item) # Additional processing for list response for key in list(filtered_item.keys()): if filtered_item[key] is None or filtered_item[key] == '' or key == '_state' or key in exclude_fields: del filtered_item[key] filtered_results.append(filtered_item) else: filtered_results = filter_dict(serializer.data) json_data = json.dumps(filtered_results) decoded_data = json.loads(json_data) return decoded_data except Exception as e: return e def get_list_response(queryset, serializer, filtered_data, paginator): response = { 'count': queryset.count(), 'num_pages': paginator.page.paginator.num_pages, 'current_count': len(serializer.data), 'next': paginator.get_next_link(), 'previous': paginator.get_previous_link(), 'response_time': 0, '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 != ''} return response ## Office Viewset class OfficeFilter(django_filters.FilterSet): complements_info = django_filters.CharFilter(field_name='complements_info', method='filter_complements_info') city__in = django_filters.CharFilter(field_name='city', method='filter_city_in') country__in = django_filters.CharFilter(field_name='country', method='filter_country_in') state__in = django_filters.CharFilter(field_name='state', method='filter_state_in') #geo = django_filters.CharFilter(field_name='distance', method='filter_geo') # def filter_geo(self, queryset, name, value): # """ # Step 1: Calculate the distance from the fixed point. # Step 2: Use the calculated distance to filter entries within a range of the `distance` field. # """ # try: # # Parse the input: "latitude,longitude,radius" # lat, lng, radius = map(float, value.split(',')) # # Fixed reference point (e.g., center of the city) # fixed_point = (-23.533773, -46.625290) # Replace with your actual fixed point # # Step 1: Calculate distance from the fixed point to the given latitude/longitude # calculated_distance = haversine(fixed_point, (lat, lng), unit=Unit.METERS) # print(f"Calculated Distance: {calculated_distance}") # # Step 2: Use the precomputed `distance` field for filtering # min_distance = max(0, calculated_distance - radius) # Prevent negative values # max_distance = calculated_distance + radius # print(f"Filtering Range: Min={min_distance}, Max={max_distance}") # return queryset.filter(distance__gte=min_distance, distance__lte=max_distance) # except ValueError as ve: # print(f"ValueError: {ve}") # except Exception as e: # import traceback # print(f"Error filtering queryset: {e}") # print(traceback.format_exc()) # # Return the unfiltered queryset in case of an error # return queryset # Custom filter for geo queries geo = django_filters.CharFilter(method='filter_geo', label='Geolocation Filter') def filter_geo(self, queryset, name, value): """ Custom filter to filter offices based on a geo parameter (latitude, longitude, and radius). The geo parameter format is "latitude,longitude,radius". """ try: # Parse the input: "latitude,longitude,radius" lat, lng, radius = map(float, value.split(',')) #print(f"Latitude: {lat}, Longitude: {lng}, Radius: {radius}") # Convert radius to degrees radius_in_degrees_lat = radius / 111000.0 import math radius_in_degrees_lng = radius / (111000.0 * math.cos(math.radians(lat))) # Calculate bounding box min_lat = round(lat - radius_in_degrees_lat,6) max_lat = round(lat + radius_in_degrees_lat,6) min_lng = round(lng - radius_in_degrees_lng,6) max_lng = round(lng + radius_in_degrees_lng,6) #print(f"Bounding Box: Min lat: {min_lat}, Max lat: {max_lat}, Min lng: {min_lng}, Max lng: {max_lng}") # Apply bounding box filter return queryset.filter( lat__gte=min_lat, lat__lte=max_lat, lng__gte=min_lng, lng__lte=max_lng ) except ValueError as ve: #print(f"ValueError in filter_geo: {ve}") logger.error(f"ValueError in filter_geo: {ve}") except Exception as e: import traceback #print(f"Error in filter_geo: {e}") #print(traceback.format_exc()) logger.error(f"Error in filter_geo: {e}") logger.info(traceback.format_exc()) return queryset def filter_country_in(self, queryset, name, value): """ Filter the queryset based on the country field with multiple values separated by commas. Arguments: queryset -- The initial queryset. name -- The name of the field to filter on. value -- The value to filter by, with multiple values separated by commas. Returns: The filtered queryset. """ countries = [country.strip() for country in value.split(',')] q_objects = Q() for country in countries: q_objects |= Q(**{f"{name}__icontains": country}) return queryset.filter(q_objects) def filter_state_in(self, queryset, name, value): """ Filter the queryset based on the state field with multiple values separated by commas. Arguments: queryset -- The initial queryset. name -- The name of the field to filter on. value -- The value to filter by, with multiple values separated by commas. Returns: The filtered queryset. """ states = [state.strip() for state in value.split(',')] q_objects = Q() for state in states: q_objects |= Q(**{f"{name}__icontains": state}) return queryset.filter(q_objects) def filter_city_in(self, queryset, name, value): """ Filter the queryset based on the city field with multiple values separated by commas. Arguments: queryset -- The initial queryset. name -- The name of the field to filter on. value -- The value to filter by, with multiple values separated by commas. Returns: The filtered queryset. """ cities = [city.strip() for city in value.split(',')] q_objects = Q() for city in cities: q_objects |= Q(**{f"{name}__icontains": city}) return queryset.filter(q_objects) def filter_complements_info(self, queryset, name, value): """ Filter the queryset based on the complements_info field. Arguments: queryset -- The initial queryset. name -- The name of the field to filter on. value -- The value to filter by. Returns: The filtered queryset. """ lookup = f"complements_info__{name}__icontains" if isinstance(value, str) else f"complements_info__{name}" return queryset.filter(**{lookup: value}) class Meta: model = Offices fields = { 'office_id': ['exact', 'in'], 'name': ['icontains'], 'email': ['icontains', 'in'], 'address': ['icontains', 'in'], 'complement': ['icontains'], 'street_number': ['icontains'], 'postal_code': ['icontains', 'in'], 'website': ['icontains'], 'logo': ['icontains'], 'telephone': ['icontains'], 'lat': ['exact', 'in'], 'lng': ['exact', 'in'], 'total_properties': ['in', 'exact'], 'creci': ['icontains','exact', 'in'], 'neighborhood': ['icontains', 'in'], 'region_name': ['icontains'], 'city': ['icontains', 'in'], 'state': ['icontains', 'in'], 'state_abbreviation': ['icontains', 'in'], 'country': ['icontains'], 'country_abbreviation': ['icontains'], } #class OfficeViewSet(viewsets.ModelViewSet): class OfficeViewSet(viewsets.ReadOnlyModelViewSet): """ A viewset for handling Office objects. Methods Allowed (ReadOnly): retrieve -- Retrieve a specific Office object by ID. list -- List all Office objects (not allowed). Methods Not Allowed: create -- Create a new Office object (not allowed). update -- Update an existing Office object (not allowed). partial_update -- Partially update an existing Office object (not allowed). destroy -- Delete an existing Office object (not allowed). """ name = "Offices" description = "Offices Endpoint" #if not settings.DEBUG: #authentication_classes = [TokenAuthentication] #authentication_classes = [JWTAuthentication] queryset = Offices.objects.using('mongodb').all() serializer_class = OfficeSerializer filter_backends = [django_filters.DjangoFilterBackend, filters.SearchFilter, filters.OrderingFilter] filterset_class = OfficeFilter search_fields='__all__' ordering_fields = '__all__' ordering = ['office_id'] pagination_class = SetPagination allowed_methods = ['GET', 'HEAD', 'OPTIONS', 'TRACE'] def get_permissions(self): """ Instantiates and returns the list of permissions that this view requires. """ if self.action in ['list', 'retrieve']: permission_classes = [IsAuthenticatedOrReadOnly] else: permission_classes = [IsAdminUser] return [permission() for permission in permission_classes] office_pk_param = openapi.Parameter('pk', openapi.IN_PATH, description="Office ID or slug", type=openapi.TYPE_STRING, required=True) manual_parameters = [office_pk_param] @swagger_auto_schema(operation_description=""" Retrieve a specific Office object by ID or slug. Arguments: request -- The HTTP request object. pk -- The ID of the Office object to retrieve. slug -- The slug of the Office object to retrieve. Returns: A Response object containing the retrieved Office object as JSON. """, manual_parameters=manual_parameters,\ responses={ 200: 'Office fields listed', 400: 'Data not Found', }, \ operation_id='retrieve_office', operation_summary='Retrieve a given office from database using the office Id or slug.', tags=['Offices']) def retrieve(self, request, pk=None): """ Retrieve a specific Office object by ID or slug. Arguments: request -- The HTTP request object. pk -- The ID of the Office object to retrieve. slug -- The slug of the Office object to retrieve. Returns: A Response object containing the retrieved Office object as JSON. """ try: start_time = time.time() if pk: if pk.isdigit(): office = self.queryset.get(office_id=pk) else: # Clean slug - trim, lower and eliminate "/" or "\" #slug = pk.strip().lower().replace('/', '').replace('\\', '') office_id = Offices.objects.mongo_aggregate([ {'$match': {'complements_info.slug': pk}}, {'$project': {'office_id': 1, '_id': 0}}, {'$limit': 1} ]) if office_id: office_id_list = list(office_id) if office_id_list: office_id = office_id_list[0]['office_id'] else: return Response({'message': 'Office not found'}, status=status.HTTP_404_NOT_FOUND) office = self.queryset.get(office_id=office_id) else: return Response({'message': 'Office not found'}, status=status.HTTP_404_NOT_FOUND) else: return Response({'message': 'ID or slug must be provided'}, status=status.HTTP_400_BAD_REQUEST) serializer = self.serializer_class(office) try: filtered_data = filter_data_dict(serializer) filtered_data['response_time'] = time.time() - start_time return Response(filtered_data, status=status.HTTP_200_OK) except Exception as e: return Response({'message': f'Error trying to retrieve item {pk} => {e}'}, status=status.HTTP_400_BAD_REQUEST) except Exception as e: return Response({'message': f'Item {pk} not found => {e}'}, status=status.HTTP_404_NOT_FOUND) # if not request.user.is_authenticated: # return Response({'message': 'Authentication credentials were not provided.'}, status=status.HTTP_401_UNAUTHORIZED) # else: # try: # start_time = time.time() # serializer = self.serializer_class(self.queryset.get(pk=pk)) # data = {} # for key, value in serializer.data.items(): # if value is not None and value != '' and key != '_state': # data[key] = value # data['response_time'] = time.time() - start_time # json_data = json.dumps(data) # decoded_data = json.loads(json_data) # return Response(decoded_data, status=status.HTTP_200_OK) # except Exception as e: # return Response({'message': f'Office Id not found => {e}'}, status=status.HTTP_404_NOT_FOUND) @swagger_auto_schema(operation_description=""" List all Office objects. Arguments: request -- The HTTP request object. Returns: A Response object containing all the Office objects as JSON. Authentication: If the user is not authenticated, return a 401 Unauthorized response """, responses={ 200: 'Offices listed', 400: 'Error listing', }, \ operation_id='list_offices', operation_summary='List all the obtained offices', tags=['Offices']) def list(self, request): try: start_time = time.time() queryset = self.filter_queryset(self.get_queryset()) paginator = SetPagination() page = paginator.paginate_queryset(queryset, request) try: serializer = self.get_serializer(page, many=True) try: filtered_data = filter_data_dict(serializer) if not filtered_data: return Response({'message': 'No Item found'}, status=status.HTTP_200_OK) response = get_list_response(self.queryset, serializer, filtered_data, paginator) response['count'] = queryset.count() response['total_count'] = self.queryset.count() response['response_time'] = time.time() - start_time return Response(response, 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': f'No Item found => {e}'}, status=status.HTTP_404_NOT_FOUND) except Exception as e: response = {'message': f'Error to return items => {e}'} return Response(response, status=status.HTTP_400_BAD_REQUEST) # try: # queryset = self.filter_queryset(self.get_queryset()) # paginator = SetPagination() # page = paginator.paginate_queryset(queryset, request) # serializer_data = self.get_serializer(page, many=True).data # authorized = request.user.is_authenticated # return list_filter(serializer_data, authorized, paginator, request) # except Exception as e: # return Response({'message': f'Error listing offices => {e}'}, status=status.HTTP_400_BAD_REQUEST) # if not request.user.is_authenticated: # return Response({'message': 'Authentication credentials were not provided.'}, status=status.HTTP_401_UNAUTHORIZED) # else: # try: # start_time = time.time() # queryset = self.filter_queryset(self.get_queryset()) # paginator = SetPagination() # page = paginator.paginate_queryset(queryset, request) # serializer = self.get_serializer(page, many=True) # if not serializer.data: # return Response({'message': 'No Item found'}, status=status.HTTP_404_NOT_FOUND) # for data in serializer.data: # for key in list(data.keys()): # if data[key] is None or data[key] == '': # del data[key] # response = { # 'count': self.queryset.count(), # 'num_pages': paginator.page.paginator.num_pages, # 'current_count': len(serializer.data), # 'next': paginator.get_next_link(), # 'previous': paginator.get_previous_link(), # 'response_time': time.time() - start_time, # 'results': serializer.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 != ''} # return Response(response, status=status.HTTP_200_OK) # except Exception as e: # response = {'message': 'Error listing offices', 'Error': str(e)} # return Response(response, status=status.HTTP_400_BAD_REQUEST) ### Method not allowed @swagger_auto_schema(operation=None, auto_schema=None) def create(self, request): response = {'message': 'Method not allowed!'} return Response(response, status=status.HTTP_405_METHOD_NOT_ALLOWED) @swagger_auto_schema(operation=None, auto_schema=None) def update(self, request, pk=None): response = {'message': 'Method not allowed!'} return Response(response, status=status.HTTP_405_METHOD_NOT_ALLOWED) @swagger_auto_schema(operation=None, auto_schema=None) def destroy(self, request, pk=None): response = {'message': 'Method not allowed'} return Response(response, status=status.HTTP_405_METHOD_NOT_ALLOWED) @swagger_auto_schema(operation=None, auto_schema=None) def partial_update(self, request, pk=None): response = {'message': 'Method not allowed!'} return Response(response, status=status.HTTP_405_METHOD_NOT_ALLOWED) ## Agents Viewset class AgentFilter(django_filters.FilterSet): geo = django_filters.CharFilter(method='filter_geo', label='Geolocation Filter') def filter_geo(self, queryset, name, value): """ Custom filter to filter offices based on a geo parameter (latitude, longitude, and radius). The geo parameter format is "latitude,longitude,radius". """ try: # Parse the input: "latitude,longitude,radius" lat, lng, radius = map(float, value.split(',')) #print(f"Latitude: {lat}, Longitude: {lng}, Radius: {radius}") # Convert radius to degrees radius_in_degrees_lat = radius / 111000.0 import math radius_in_degrees_lng = radius / (111000.0 * math.cos(math.radians(lat))) # Calculate bounding box min_lat = round(lat - radius_in_degrees_lat,6) max_lat = round(lat + radius_in_degrees_lat,6) min_lng = round(lng - radius_in_degrees_lng,6) max_lng = round(lng + radius_in_degrees_lng,6) #print(f"Bounding Box: Min lat: {min_lat}, Max lat: {max_lat}, Min lng: {min_lng}, Max lng: {max_lng}") # Apply bounding box filter return queryset.filter( lat__gte=min_lat, lat__lte=max_lat, lng__gte=min_lng, lng__lte=max_lng ) except ValueError as ve: #print(f"ValueError in filter_geo: {ve}") logger.error(f"ValueError in filter_geo: {ve}") except Exception as e: import traceback #print(f"Error in filter_geo: {e}") #print(traceback.format_exc()) logger.error(f"Error in filter_geo: {e}") logger.info(traceback.format_exc()) return queryset has_profile_picture = django_filters.BooleanFilter(field_name='profile_picture', method='filter_has_profile_picture') def filter_has_profile_picture(self, queryset, name, value): if value: return queryset.filter(profile_picture='') else: return queryset.exclude(profile_picture='') has_creci = django_filters.BooleanFilter(field_name='creci', method='filter_has_creci') def filter_creci(self, queryset, name, value): if value: return queryset.filter(creci='') else: return queryset.exclude(creci='') complements_info = django_filters.CharFilter(field_name='complements_info', method='filter_complements_info') def filter_complements_info(self, queryset, name, value): """ Filter the queryset based on the complements_info field. Arguments: queryset -- The initial queryset. name -- The name of the field to filter on. value -- The value to filter by. Returns: The filtered queryset. """ lookup = f"complements_info__{name}__icontains" if isinstance(value, str) else f"complements_info__{name}" return queryset.filter(**{lookup: value}) office__city__in_str = django_filters.CharFilter(field_name='office__city', method='filter_office_city_in_str') def filter_office_city_in_str(self, queryset, name, value): """ Filter the queryset based on the office city field with multiple values separated by commas. Arguments: queryset -- The initial queryset. name -- The name of the field to filter on. value -- The value to filter by, with multiple values separated by commas. Returns: The filtered queryset. """ cities = [city.strip() for city in value.split(',')] q_objects = Q() for city in cities: q_objects |= Q(**{f"{name}__icontains": city}) return queryset.filter(q_objects) class Meta: model = Agents fields = { 'agent_id': ['exact','in'], 'name': ['icontains', 'in'], 'email': ['contains', 'in'], 'telephone': ['icontains'], 'creci': ['icontains','exact', 'in'], 'website': ['icontains'], 'region_id': ['exact', 'in'], 'profile_picture': ['contains', 'in'], 'total_properties': ['in', 'exact'], 'address': ['icontains', 'in'], 'street_number': ['icontains'], 'complement': ['icontains'], 'postal_code': ['icontains', 'in'], 'neighborhood': ['icontains', 'in'], 'region_name': ['icontains','in'], 'city': ['icontains', 'in'], 'state': ['icontains', 'in'], 'state_abbreviation': ['icontains', 'in'], 'country': ['icontains'], 'country_abbreviation': ['icontains'], 'lng': ['exact', 'in'], 'lat': ['exact', 'in'], 'office__office_id': ['exact', 'in'], 'office__name': ['icontains'], 'office__address': ['icontains', 'in'], 'office__complement': ['icontains'], 'office__street_number': ['icontains'], 'office__postal_code': ['icontains', 'in'], 'office__website': ['icontains'], 'office__logo': ['icontains'], 'office__telephone': ['icontains'], 'office__lat': ['exact', 'in'], 'office__lng': ['exact', 'in'], 'office__total_properties': ['in', 'exact'], 'office__creci': ['icontains','exact', 'in'], 'office__neighborhood': ['icontains', 'in'], 'office__region_name': ['icontains'], 'office__city': ['icontains', 'in'], 'office__state': ['icontains', 'in'], 'office__state_abbreviation': ['icontains', 'in'], 'office__country': ['icontains'], 'office__country_abbreviation': ['icontains'], 'complements_info': ['icontains','exact', 'in'], } class AgentOrderingFilter(filters.OrderingFilter): def get_ordering(self, request, queryset, view): params = request.query_params.get(self.ordering_param) if params: fields = [param.strip() for param in params.split(',')] ordering = [] for field in fields: if field == 'has_profile_picture': ordering.append('profile_picture') elif field == 'has_creci': ordering.append('creci') elif field.startswith('-'): ordering.append(f'-{field[1:]}') else: ordering.append(field) return ordering return self.get_default_ordering(view) class AgentViewSet(viewsets.ReadOnlyModelViewSet): """ A viewset for handling Agent objects. Methods Allowed: retrieve -- Retrieve a specific Agent object by ID. list -- List all Agent objects (not allowed). Methods Not Allowed: create -- Create a new Agent object (not allowed). update -- Update an existing Agent object (not allowed). destroy -- Delete an existing Agent object (not allowed). """ name = "Agents" description = "Agents Endpoint" queryset = Agents.objects.using('mongodb').all() serializer_class = AgentSerializer # if not settings.DEBUG: # authentication_classes = [TokenAuthentication] # authentication_classes = [JWTAuthentication] filter_backends = [django_filters.DjangoFilterBackend, filters.SearchFilter, filters.OrderingFilter] filterset_class = AgentFilter search_fields='__all__' ordering_fields = '__all__' ordering_class = AgentOrderingFilter ordering = ['-agent_id'] pagination_class = SetPagination allowed_methods = ['GET','HEAD', 'OPTIONS', 'TRACE'] def get_permissions(self): """ Instantiates and returns the list of permissions that this view requires. """ if self.action in ['list', 'retrieve']: permission_classes = [IsAuthenticatedOrReadOnly] pass else: permission_classes = [IsAdminUser] return [permission() for permission in permission_classes] agent_pk_param = openapi.Parameter('pk', openapi.IN_PATH, description="AgentId or Slug", type=openapi.TYPE_STRING, required=True) manual_parameters = [agent_pk_param] @swagger_auto_schema(operation_description=""" Retrieve a specific Agent object by ID or slug. Arguments: request -- The HTTP request object. pk -- The ID of the Agent object to retrieve. slug -- The slug of the Agent object to retrieve. Returns: A Response object containing the retrieved Agent object as JSON. """, manual_parameters=manual_parameters,\ responses={ 200: 'Agent fields listed', 400: 'Data not Found', }, \ operation_id='retrieve_agent', operation_summary='Retrieve a given agent from database using the agent Id or slug.', tags=['Agents']) def retrieve(self, request, pk=None): """ Retrieve a specific Agent object by ID or slug. Arguments: request -- The HTTP request object. pk -- The ID of the Agent object to retrieve. slug -- The slug of the Agent object to retrieve. Returns: A Response object containing the retrieved Agent object as JSON. """ try: start_time = time.time() if pk: if pk.isdigit(): agent = self.queryset.get(agent_id=pk) else: # Clean slug - trim, lower and eliminate "/" or "\" #slug = pk.strip().lower().replace('/', '').replace('\\', '') agent_id = Agents.objects.mongo_aggregate([ {'$match': {'complements_info.slug': pk}}, {'$project': {'agent_id': 1, '_id': 0}}, {'$limit': 1} ]) if agent_id: agent_id_list = list(agent_id) if agent_id_list: agent_id = agent_id_list[0]['agent_id'] else: return Response({'message': 'Agent not found'}, status=status.HTTP_404_NOT_FOUND) agent = self.queryset.get(agent_id=agent_id) else: return Response({'message': 'Agent not found'}, status=status.HTTP_404_NOT_FOUND) else: return Response({'message': 'ID or slug must be provided'}, status=status.HTTP_400_BAD_REQUEST) serializer = self.serializer_class(agent) try: filtered_data = filter_data_dict(serializer) filtered_data['response_time'] = time.time() - start_time return Response(filtered_data, status=status.HTTP_200_OK) except Exception as e: return Response({'message': f'Error trying to retrieve item {pk} => {e}'}, status=status.HTTP_400_BAD_REQUEST) except Exception as e: return Response({'message': f'Item {pk} not found => {e}'}, status=status.HTTP_404_NOT_FOUND) @swagger_auto_schema(operation_description=""" List all Agent objects. Arguments: request -- The HTTP request object. Returns: A Response object containing all the Agent objects as JSON. Authentication: If the user is not authenticated, return a 401 Unauthorized response """, responses={ 200: 'Agents listed', 400: 'Error listing', }, \ operation_id='list_agents', operation_summary='List all the obtained agents', tags=['Agents']) def list(self, request): try: start_time = time.time() queryset = self.filter_queryset(self.get_queryset()) paginator = SetPagination() page = paginator.paginate_queryset(queryset, request) try: serializer = self.get_serializer(page, many=True) try: filtered_data = filter_data_dict(serializer) if not filtered_data: return Response({'message': 'No Item found'}, status=status.HTTP_200_OK) response = get_list_response(self.queryset, serializer, filtered_data, paginator) response['count'] = queryset.count() response['total_count'] = self.queryset.count() response['response_time'] = time.time() - start_time return Response(response, 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': f'No Item found => {e}'}, status=status.HTTP_404_NOT_FOUND) except Exception as e: response = {'message': f'Error to return items => {e}'} return Response(response, status=status.HTTP_400_BAD_REQUEST) ### Method not allowed @swagger_auto_schema(operation=None, auto_schema=None) def create(self, request): response = {'message': 'Method not allowed!'} return Response(response, status=status.HTTP_405_METHOD_NOT_ALLOWED) @swagger_auto_schema(operation=None, auto_schema=None) def update(self, request, pk=None): response = {'message': 'Method not allowed!'} return Response(response, status=status.HTTP_405_METHOD_NOT_ALLOWED) @swagger_auto_schema(operation=None, auto_schema=None) def destroy(self, request, pk=None): response = {'message': 'Method not allowed'} return Response(response, status=status.HTTP_405_METHOD_NOT_ALLOWED) @swagger_auto_schema(operation=None, auto_schema=None) def partial_update(self, request, pk=None): response = {'message': 'Method not allowed!'} return Response(response, status=status.HTTP_405_METHOD_NOT_ALLOWED) class FeatureFilter(django_filters.FilterSet): class Meta: model = Features fields = { 'feature_id': ['exact'], 'feature_name_en': ['icontains'], 'feature_name_pt': ['icontains'], } class FeatureViewSet(viewsets.ReadOnlyModelViewSet): """ A viewset for handling Feature objects. Methods Allowed: retrieve -- Retrieve a specific Feature object by ID. list -- List all Feature objects (not allowed). obtain_feature -- Obtain a feature from OpenAI based on provided parameters. obtain_quick_feature -- Obtain a quick feature from OpenAI based on provided parameters. Methods Not Allowed: create -- Create a new Feature object (not allowed). update -- Update an existing Feature object (not allowed). destroy -- Delete an existing Feature object (not allowed). """ name = "Features" description = "Features Endpoint" queryset = Features.objects.using('mongodb').all() serializer_class = FeatureSerializer # if not settings.DEBUG: # authentication_classes = [TokenAuthentication] # authentication_classes = [JWTAuthentication] filter_backends = [django_filters.DjangoFilterBackend, filters.SearchFilter, filters.OrderingFilter] filterset_class = FeatureFilter search_fields=['feature_name_en', 'feature_name_pt',] ordering_fields = '__all__' ordering = ['feature_id'] pagination_class = PageNumberPagination allowed_methods = ['GET', 'POST', 'HEAD', 'OPTIONS', 'TRACE'] def get_permissions(self): """ Instantiates and returns the list of permissions that this view requires. """ if self.action in ['list', 'retrieve']: permission_classes = [IsAuthenticatedOrReadOnly] pass else: permission_classes = [IsAdminUser] return [permission() for permission in permission_classes] feature_id_param = openapi.Parameter('id', openapi.IN_PATH, description="Feature register ID", type=openapi.TYPE_INTEGER, required=True) manual_parameters = [feature_id_param] @swagger_auto_schema(operation_description=""" Retrieve a specific Feature object by ID. Arguments: request -- The HTTP request object. pk -- The ID of the Feature object to retrieve. Returns: A Response object containing the retrieved Feature object as JSON. """, manual_parameters=manual_parameters,\ responses={ 200: 'Feature fields listed', 400: 'Data not Found', }, \ operation_id='retrieve_feature', operation_summary='Retrieve a given feature from database using the feature Id.', tags=['Features']) def retrieve(self, request, pk=None): """ Retrieve a specific Feature object by ID. Arguments: request -- The HTTP request object. pk -- The ID of the Feature object to retrieve. Returns: A Response object containing the retrieved Feature object as JSON. """ try: start_time = time.time() serializer = self.serializer_class(self.queryset.get(feature_id=pk)) try: filtered_data = filter_data_dict(serializer) filtered_data['response_time'] = time.time() - start_time return Response(filtered_data, status=status.HTTP_200_OK) except Exception as e: return Response({'message': f'Error trying to retrieve item {pk} => {e}'}, status=status.HTTP_400_BAD_REQUEST) except Exception as e: return Response({'message': f'Item {pk} not found => {e}'}, status=status.HTTP_404_NOT_FOUND) @swagger_auto_schema(operation_description=""" List all Feature objects. Arguments: request -- The HTTP request object. Returns: A Response object containing all the Feature objects as JSON. Authentication: If the user is not authenticated, return a 401 Unauthorized response """, responses={ 200: 'Features listed', 400: 'Error listing', }, \ operation_id='list_features', operation_summary='List all the obtained features', tags=['Features']) def list(self, request): try: start_time = time.time() queryset = self.filter_queryset(self.get_queryset()) paginator = SetPagination() page = paginator.paginate_queryset(queryset, request) try: serializer = self.get_serializer(page, many=True) try: filtered_data = filter_data_dict(serializer) response = get_list_response(self.queryset, serializer, filtered_data, paginator) response['count'] = queryset.count() response['total_count'] = self.queryset.count() response['response_time'] = time.time() - start_time return Response(response, 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': f'No Item found => {e}'}, status=status.HTTP_404_NOT_FOUND) except Exception as e: response = {'message': f'Error to return items => {e}'} return Response(response, status=status.HTTP_400_BAD_REQUEST) ### Method not allowed @swagger_auto_schema(operation=None, auto_schema=None) def create(self, request): response = {'message': 'Method not allowed!'} return Response(response, status=status.HTTP_405_METHOD_NOT_ALLOWED) @swagger_auto_schema(operation=None, auto_schema=None) def update(self, request, pk=None): response = {'message': 'Method not allowed!'} return Response(response, status=status.HTTP_405_METHOD_NOT_ALLOWED) @swagger_auto_schema(operation=None, auto_schema=None) def destroy(self, request, pk=None): response = {'message': 'Method not allowed'} return Response(response, status=status.HTTP_405_METHOD_NOT_ALLOWED) @swagger_auto_schema(operation=None, auto_schema=None) def partial_update(self, request, pk=None): response = {'message': 'Method not allowed!'} return Response(response, status=status.HTTP_405_METHOD_NOT_ALLOWED) ### Usage Viewset class UsageFilter(django_filters.FilterSet): class Meta: model = Usages fields = { 'usage_id': ['exact'], 'usage_name_en': ['icontains'], 'usage_name_pt': ['icontains'], } class UsageViewSet(viewsets.ReadOnlyModelViewSet): """ A viewset for handling Usage objects. Methods Allowed: retrieve -- Retrieve a specific Usage object by ID. list -- List all Usage objects (not allowed). Methods Not Allowed: create -- Create a new Usage object (not allowed). update -- Update an existing Usage object (not allowed). destroy -- Delete an existing Usage object (not allowed). """ name = "Usages" description = "Usages Endpoint" queryset = Usages.objects.using('mongodb').all() serializer_class = UsageSerializer # if not settings.DEBUG: # authentication_classes = [TokenAuthentication] # authentication_classes = [JWTAuthentication] filter_backends = [django_filters.DjangoFilterBackend, filters.SearchFilter, filters.OrderingFilter] filterset_class = UsageFilter search_fields=['usage_name_en', 'usage_name_pt',] ordering_fields = '__all__' ordering = ['usage_id'] pagination_class = PageNumberPagination allowed_methods = ['GET', 'POST', 'HEAD', 'OPTIONS', 'TRACE'] def get_permissions(self): """ Instantiates and returns the list of permissions that this view requires. """ if self.action in ['list', 'retrieve']: permission_classes = [IsAuthenticatedOrReadOnly] pass else: permission_classes = [IsAdminUser] return [permission() for permission in permission_classes] usage_id_param = openapi.Parameter('id', openapi.IN_PATH, description="Usage register ID", type=openapi.TYPE_INTEGER, required=True) manual_parameters = [usage_id_param] @swagger_auto_schema(operation_description=""" Retrieve a specific Usage object by ID. Arguments: request -- The HTTP request object. pk -- The ID of the Usage object to retrieve. Returns: A Response object containing the retrieved Usage object as JSON. """, manual_parameters=manual_parameters,\ responses={ 200: 'Usage fields listed', 400: 'Data not Found', }, \ operation_id='retrieve_usage', operation_summary='Retrieve a given usage from database using the usage Id.', tags=['Usages']) def retrieve(self, request, pk=None): """ Retrieve a specific Usage object by ID. Arguments: request -- The HTTP request object. pk -- The ID of the Usage object to retrieve. Returns: A Response object containing the retrieved Usage object as JSON. """ try: start_time = time.time() serializer = self.serializer_class(self.queryset.get(usage_id=pk)) try: filtered_data = filter_data_dict(serializer) filtered_data['response_time'] = time.time() - start_time return Response(filtered_data, status=status.HTTP_200_OK) except Exception as e: return Response({'message': f'Error trying to retrieve item {pk} => {e}'}, status=status.HTTP_400_BAD_REQUEST) except Exception as e: return Response({'message': f'Item {pk} not found => {e}'}, status=status.HTTP_404_NOT_FOUND) @swagger_auto_schema(operation_description=""" List all Usage objects. Arguments: request -- The HTTP request object. Returns: A Response object containing all the Usage objects as JSON. Authentication: If the user is not authenticated, return a 401 Unauthorized response """, responses={ 200: 'Usages listed', 400: 'Error listing', }, \ operation_id='list_usages', operation_summary='List all the obtained usages', tags=['Usages']) def list(self, request): try: start_time = time.time() queryset = self.filter_queryset(self.get_queryset()) paginator = SetPagination() page = paginator.paginate_queryset(queryset, request) try: serializer = self.get_serializer(page, many=True) try: filtered_data = filter_data_dict(serializer) response = get_list_response(self.queryset, serializer, filtered_data, paginator) response['count'] = queryset.count() response['total_count'] = self.queryset.count() response['response_time'] = time.time() - start_time return Response(response, 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': f'No Item found => {e}'}, status=status.HTTP_404_NOT_FOUND) except Exception as e: response = {'message': f'Error to return items => {e}'} return Response(response, status=status.HTTP_400_BAD_REQUEST) ### Method not allowed @swagger_auto_schema(operation=None, auto_schema=None) def create(self, request): response = {'message': 'Method not allowed!'} return Response(response, status=status.HTTP_405_METHOD_NOT_ALLOWED) @swagger_auto_schema(operation=None, auto_schema=None) def update(self, request, pk=None): response = {'message': 'Method not allowed!'} return Response(response, status=status.HTTP_405_METHOD_NOT_ALLOWED) @swagger_auto_schema(operation=None, auto_schema=None) def destroy(self, request, pk=None): response = {'message': 'Method not allowed'} return Response(response, status=status.HTTP_405_METHOD_NOT_ALLOWED) @swagger_auto_schema(operation=None, auto_schema=None) def partial_update(self, request, pk=None): response = {'message': 'Method not allowed!'} return Response(response, status=status.HTTP_405_METHOD_NOT_ALLOWED) class TypeFilter(django_filters.FilterSet): class Meta: model = Types fields = { 'type_id': ['exact'], 'type_name_en': ['icontains'], 'type_name_pt': ['icontains'], 'usage__usage_name_en': ['icontains'], 'usage__usage_name_pt': ['icontains'], } class TypeViewSet(viewsets.ReadOnlyModelViewSet): """ A viewset for handling Type objects. Methods Allowed: retrieve -- Retrieve a specific Type object by ID. list -- List all Type objects (not allowed). Methods Not Allowed: create -- Create a new Type object (not allowed). update -- Update an existing Type object (not allowed). destroy -- Delete an existing Type object (not allowed). """ name = "Types" description = "Types Endpoint" queryset = Types.objects.using('mongodb').all() serializer_class = TypeSerializer # if not settings.DEBUG: # authentication_classes = [TokenAuthentication] # authentication_classes = [JWTAuthentication] filter_backends = [django_filters.DjangoFilterBackend, filters.SearchFilter, filters.OrderingFilter] filterset_class = TypeFilter search_fields=['type_name_en', 'type_name_pt', 'usage__usage_name_en', 'usage__usage_name_pt',] ordering_fields = '__all__' ordering = ['type_id'] pagination_class = PageNumberPagination allowed_methods = ['GET', 'POST', 'HEAD', 'OPTIONS', 'TRACE'] def get_permissions(self): """ Instantiates and returns the list of permissions that this view requires. """ if self.action in ['list', 'retrieve']: permission_classes = [IsAuthenticatedOrReadOnly] pass else: permission_classes = [IsAdminUser] return [permission() for permission in permission_classes] type_id_param = openapi.Parameter('id', openapi.IN_PATH, description="Type register ID", type=openapi.TYPE_INTEGER, required=True) manual_parameters = [type_id_param] @swagger_auto_schema(operation_description=""" Retrieve a specific Type object by ID. Arguments: request -- The HTTP request object. pk -- The ID of the Type object to retrieve. Returns: A Response object containing the retrieved Type object as JSON. """, manual_parameters=manual_parameters,\ responses={ 200: 'Type fields listed', 400: 'Data not Found', }, \ operation_id='retrieve_type', operation_summary='Retrieve a given type from database using the type Id.', tags=['Types']) def retrieve(self, request, pk=None): """ Retrieve a specific Type object by ID. Arguments: request -- The HTTP request object. pk -- The ID of the Type object to retrieve. Returns: A Response object containing the retrieved Type object as JSON. """ try: start_time = time.time() serializer = self.serializer_class(self.queryset.get(type_id=pk)) try: filtered_data = filter_data_dict(serializer) filtered_data['response_time'] = time.time() - start_time return Response(filtered_data, status=status.HTTP_200_OK) except Exception as e: return Response({'message': f'Error trying to retrieve item {pk} => {e}'}, status=status.HTTP_400_BAD_REQUEST) except Exception as e: return Response({'message': f'Item {pk} not found => {e}'}, status=status.HTTP_404_NOT_FOUND) @swagger_auto_schema(operation_description=""" List all Type objects. Arguments: request -- The HTTP request object. Returns: A Response object containing all the Type objects as JSON. Authentication: If the user is not authenticated, return a 401 Unauthorized response """, responses={ 200: 'Types listed', 400: 'Error listing', }, \ operation_id='list_types', operation_summary='List all the obtained types', tags=['Types']) def list(self, request): try: start_time = time.time() queryset = self.filter_queryset(self.get_queryset()) paginator = SetPagination() page = paginator.paginate_queryset(queryset, request) try: serializer = self.get_serializer(page, many=True) try: filtered_data = filter_data_dict(serializer) response = get_list_response(self.queryset, serializer, filtered_data, paginator) response['count'] = queryset.count() response['total_count'] = self.queryset.count() response['response_time'] = time.time() - start_time return Response(response, 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': f'No Item found => {e}'}, status=status.HTTP_404_NOT_FOUND) except Exception as e: response = {'message': f'Error to return items => {e}'} return Response(response, status=status.HTTP_400_BAD_REQUEST) ### Method not allowed @swagger_auto_schema(operation=None, auto_schema=None) def create(self, request): response = {'message': 'Method not allowed!'} return Response(response, status=status.HTTP_405_METHOD_NOT_ALLOWED) @swagger_auto_schema(operation=None, auto_schema=None) def update(self, request, pk=None): response = {'message': 'Method not allowed!'} return Response(response, status=status.HTTP_405_METHOD_NOT_ALLOWED) @swagger_auto_schema(operation=None, auto_schema=None) def destroy(self, request, pk=None): response = {'message': 'Method not allowed'} return Response(response, status=status.HTTP_405_METHOD_NOT_ALLOWED) @swagger_auto_schema(operation=None, auto_schema=None) def partial_update(self, request, pk=None): response = {'message': 'Method not allowed!'} return Response(response, status=status.HTTP_405_METHOD_NOT_ALLOWED) class LocationFilter(django_filters.FilterSet): class Meta: model = Location fields = { 'property_location_id': ['exact'], 'display_address': ['icontains'], 'country': ['icontains', 'exact'], 'country_abbreviation': ['icontains', 'exact'], 'state': ['icontains', 'exact'], 'city': ['icontains', 'exact'], 'zone': ['icontains', 'exact'], 'neighborhood': ['icontains', 'exact'], 'complement': ['icontains', 'exact'], 'address': ['icontains'], 'street_number': ['icontains'], 'postal_code': ['icontains'], 'lat': ['exact'], 'lng': ['exact'], } class LocationViewSet(viewsets.ReadOnlyModelViewSet): """ A viewset for handling Location objects. Methods Allowed: retrieve -- Retrieve a specific Location object by ID. list -- List all Location objects. Methods Not Allowed: create -- Create a new Location object. update -- Update an existing Location object. destroy -- Delete an existing Location object. """ name = "Locations" description = "Locations Endpoint" queryset = Location.objects.using('mongodb').all() serializer_class = LocationSerializer # if not settings.DEBUG: # authentication_classes = [TokenAuthentication] # authentication_classes = [JWTAuthentication] filter_backends = [django_filters.DjangoFilterBackend, filters.SearchFilter, filters.OrderingFilter] filterset_class = LocationFilter search_fields=['country', 'country_abbreviation', 'state', 'city', 'zone', 'neighborhood', 'complement', 'address', 'street_number', 'postal_code','lat', 'lng'] ordering_fields = '__all__' ordering = ['property_location_id'] pagination_class = PageNumberPagination allowed_methods = ['GET', 'POST', 'HEAD', 'OPTIONS', 'TRACE'] def get_permissions(self): """ Instantiates and returns the list of permissions that this view requires. """ if self.action in ['list', 'retrieve','aggregate_locations']: permission_classes = [IsAuthenticatedOrReadOnly] pass else: permission_classes = [IsAdminUser] return [permission() for permission in permission_classes] location_id_param = openapi.Parameter('id', openapi.IN_PATH, description="Location register ID", type=openapi.TYPE_INTEGER, required=True) manual_parameters = [location_id_param] @swagger_auto_schema(operation_description=""" Retrieve a specific Location object by ID. Arguments: request -- The HTTP request object. pk -- The ID of the Location object to retrieve. Returns: A Response object containing the retrieved Location object as JSON. """, manual_parameters=manual_parameters,\ responses={ 200: 'Location fields listed', 400: 'Data not Found', }, \ operation_id='retrieve_location', operation_summary='Retrieve a given location from database using the location Id.', tags=['Locations']) def retrieve(self, request, pk=None): """ Retrieve a specific Location object by ID. Arguments: request -- The HTTP request object. pk -- The ID of the Location object to retrieve. Returns: A Response object containing the retrieved Location object as JSON. """ try: start_time = time.time() serializer = self.serializer_class(self.queryset.get(property_location_id=pk)) try: filtered_data = filter_data_dict(serializer) filtered_data['response_time'] = time.time() - start_time return Response(filtered_data, status=status.HTTP_200_OK) except Exception as e: return Response({'message': f'Error trying to retrieve item {pk} => {e}'}, status=status.HTTP_400_BAD_REQUEST) except Exception as e: return Response({'message': f'Item {pk} not found => {e}'}, status=status.HTTP_404_NOT_FOUND) @swagger_auto_schema(operation_description=""" List all Location objects. Arguments: request -- The HTTP request object. Returns: A Response object containing all the Location objects as JSON. Authentication: If the user is not authenticated, return a 401 Unauthorized response """, responses={ 200: 'Locations listed', 400: 'Error listing', }, \ operation_id='list_locations', operation_summary='List all the obtained locations', tags=['Locations']) def list(self, request): try: start_time = time.time() queryset = self.filter_queryset(self.get_queryset()) paginator = SetPagination() page = paginator.paginate_queryset(queryset, request) try: serializer = self.get_serializer(page, many=True) try: filtered_data = filter_data_dict(serializer) response = get_list_response(self.queryset, serializer, filtered_data, paginator) response['count'] = queryset.count() response['total_count'] = self.queryset.count() response['response_time'] = time.time() - start_time return Response(response, 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': f'No Item found => {e}'}, status=status.HTTP_404_NOT_FOUND) except Exception as e: response = {'message': f'Error to return items => {e}'} return Response(response, status=status.HTTP_400_BAD_REQUEST) @swagger_auto_schema(operation_description=""" List Total of properties grouped by state and city. Arguments: request -- The HTTP request object. Filters: state= city= Returns: A Response object containing all the Location state and city objects as JSON. Authentication: If the user is not authenticated, return a 401 Unauthorized response """, responses={ 200: 'Locations listed', 400: 'Error listing', }, \ operation_id='aggregate_locations', operation_summary='Aggregate properties by location', tags=['Locations']) @action(detail=False, methods=['get'], url_path='aggregate', url_name='aggregate', permission_classes=[AllowAny]) def aggregate_locations(self, request, *args, **kwargs): """ Custom action to aggregate properties by location (state and city). """ state = request.query_params.get('state') city = request.query_params.get('city') #print(f"State: {state}, City: {city}") aggregated_data = aggregate_properties_by_location(state=state, city=city) context = {'state': state, 'city': city} serializer = PropertiesNestedAggregateSerializer({'location_aggregate': aggregated_data}, context=context) return Response(serializer.data, status=status.HTTP_200_OK) ### Method not allowed @swagger_auto_schema(operation=None, auto_schema=None) def create(self, request): response = {'message': 'Method not allowed!'} return Response(response, status=status.HTTP_405_METHOD_NOT_ALLOWED) @swagger_auto_schema(operation=None, auto_schema=None) def update(self, request, pk=None): response = {'message': 'Method not allowed!'} return Response(response, status=status.HTTP_405_METHOD_NOT_ALLOWED) @swagger_auto_schema(operation=None, auto_schema=None) def destroy(self, request, pk=None): response = {'message': 'Method not allowed'} return Response(response, status=status.HTTP_405_METHOD_NOT_ALLOWED) @swagger_auto_schema(operation=None, auto_schema=None) def partial_update(self, request, pk=None): response = {'message': 'Method not allowed!'} return Response(response, status=status.HTTP_405_METHOD_NOT_ALLOWED) class PropertyFilter(django_filters.FilterSet): list_price= django_filters.RangeFilter( field_name='list_price', ) living_area= django_filters.RangeFilter( field_name='living_area', ) publish_date= django_filters.DateFilter( field_name='publish_date', ) sync_date= django_filters.DateFilter( field_name='sync_date', ) class Meta: model = Properties fields = { 'listing_id': ['in', 'exact'], 'title': ['icontains'], 'transaction_type': ['icontains'], 'detail_view_url': ['icontains'], 'description': ['icontains'], 'list_price': ['range','exact'], 'list_price_currency': ['exact'], 'rental_price' : ['range','exact'], 'rental_price_currency': ['exact'], 'rental_price_period': ['icontains'], 'property_administration_fee' : ['range','exact'], 'property_administration_fee_period': ['icontains'], 'yearly_tax': ['range', 'exact'], 'yearly_tax_currency': ['exact'], 'living_area': ['range', 'exact'], 'living_area_unit': ['exact'], 'year_built' : ['in', 'exact'], 'bedrooms': ['in', 'exact', 'range','gte'], 'bathrooms': ['in','exact', 'range','gte'], 'garage': ['in', 'exact', 'range', 'gte'], 'garage_type': ['exact'], 'unit_floor': ['in', 'exact', 'range'], 'unit_number': ['in', 'exact'], 'publish_date': ['exact', 'range', 'icontains'], 'sync_date': ['exact', 'range'], 'region_id': ['in', 'exact'], 'office__office_id': ['in', 'exact'], 'office__name': ['in', 'exact','icontains'], 'office__website': ['icontains'], 'office__logo': ['icontains'], 'office__telephone': ['icontains'], 'office__country': ['icontains'], 'office__country_abbreviation': ['exact'], 'office__state': ['icontains'], 'office__state_abbreviation': ['exact'], 'office__city': ['icontains'], 'office__neighborhood': ['icontains'], 'office__complement': ['icontains'], 'office__address': ['icontains'], 'office__street_number': ['exact'], 'office__postal_code': ['exact'], 'office__total_properties': ['in', 'exact', 'range'], 'agent__agent_id': ['exact'], 'agent__name': ['exact', 'icontains'], 'agent__agent_id': ['exact'], 'agent__name': ['exact', 'icontains'], 'agent__email': ['icontains'], 'agent__total_properties':['in', 'exact', 'range'], 'agent__agent_id': ['exact'], 'agent__name': ['exact', 'icontains'], 'usage__usage_id': ['exact'], 'usage__usage_name_en': ['exact', 'icontains'], 'usage__usage_name_pt': ['exact', 'icontains'], 'type__type_id': ['exact'], 'type__type_name_en': ['exact', 'icontains'], 'type__type_name_pt': ['exact', 'icontains'], 'location__property_location_id': ['in', 'exact'], 'location__display_address': ['icontains'], 'location__country': ['in', 'exact','icontains'], 'location__country_abbreviation': ['in','icontains'], 'location__state': ['in', 'icontains'], 'location__zone': ['in', 'icontains'], 'location__neighborhood': ['in', 'icontains'], 'location__complement': ['icontains'], 'location__address': ['in', 'icontains'], 'location__street_number': ['icontains'], 'location__postal_code': ['icontains'], 'location__lat': ['exact'], 'location__lng': ['exact'], 'medias__url': ['icontains'], 'features__feature__feature_name_en': ['in', 'icontains'], 'features__feature__feature_name_pt': ['in', 'icontains'], } class PropertyViewSet(viewsets.ReadOnlyModelViewSet): """ A viewset for handling Property objects. Methods Allowed: retrieve -- Retrieve a specific Property object by ID. list -- List all Property objects. Methods Not Allowed: create -- Create a new Property object. update -- Update an existing Property object. destroy -- Delete an existing Property object. """ # if not settings.DEBUG: # authentication_classes = [TokenAuthentication] # authentication_classes = [JWTAuthentication] queryset = Properties.objects.using('mongodb').all() serializer_class = PropertySerializer name = "Properties" description = "Properties Endpoint" filter_backends = [django_filters.DjangoFilterBackend, filters.SearchFilter, filters.OrderingFilter] filterset_class = PropertyFilter search_fields = ['listing_id', 'title', 'property_type', 'description', 'list_price', 'rental_price', 'living_area', 'bedrooms', 'bathrooms', 'year_built', 'garage', 'garage_type', 'unit_floor', 'unit_number', 'office__name', 'office__website', 'office__telephone','agent__name', 'agent__email', 'usage__usage_name_en', 'usage__usage_name_pt', 'type__type_name_en', 'type__type_name_pt', 'location__display_address', 'location__country', 'location__country_abbreviation', 'location__state', 'location__zone', 'location__neighborhood', 'location__complement', 'location__address', 'location__postal_code', 'location__lat', 'location__lng',] ordering_fields = '__all__' ordering = ['property_id'] pagination_class = SetPagination allowed_methods = ['GET', 'HEAD', 'OPTIONS', 'TRACE'] def get_permissions(self): """ Instantiates and returns the list of permissions that this view requires. """ if self.action in ['list', 'retrieve','summary','simple_list','simple_retrieve', 'complete_list']: permission_classes = [IsAuthenticatedOrReadOnly] else: permission_classes = [IsAdminUser] return [permission() for permission in permission_classes] property_id_param = openapi.Parameter('id', openapi.IN_PATH, description="Property register ID", type=openapi.TYPE_INTEGER, required=True) manual_parameters = [property_id_param] @swagger_auto_schema(operation_description=""" Retrieve a specific Property object by ID. Arguments: request -- The HTTP request object. pk -- The ID of the Property object to retrieve. Returns: A Response object containing the retrieved Property object as JSON. """, manual_parameters=manual_parameters,\ responses={ 200: 'Property fields listed', 400: 'Data not Found', }, \ operation_id='retrieve_property', operation_summary='Retrieve a given property from database using the property Id.', tags=['Properties']) def retrieve(self, request, pk=None): """ Retrieve a specific Property object by ID. Arguments: request -- The HTTP request object. pk -- The ID of the Property object to retrieve. Returns: A Response object containing the retrieved Property object as JSON. """ try: start_time = time.time() serializer = self.serializer_class(self.queryset.get(property_id=pk)) try: filtered_data = filter_data_dict(serializer) filtered_data['response_time'] = time.time() - start_time return Response(filtered_data, status=status.HTTP_200_OK) except Exception as e: return Response({'message': f'Error trying to retrieve item {pk} => {e}'}, status=status.HTTP_400_BAD_REQUEST) except Exception as e: return Response({'message': f'Item {pk} not found => {e}'}, status=status.HTTP_404_NOT_FOUND) @swagger_auto_schema(operation_description=""" List all Property objects. Arguments: request -- The HTTP request object. Returns: A Response object containing all the Property objects as JSON. Authentication: If the user is not authenticated, return a 401 Unauthorized response """, responses={ 200: 'Properties listed', 400: 'Error listing', }, \ operation_id='list_properties', operation_summary='List all the obtained properties', tags=['Properties']) def list(self, request): try: start_time = time.time() queryset = self.filter_queryset(self.get_queryset()) paginator = SetPagination() page = paginator.paginate_queryset(queryset, request) try: serializer = self.get_serializer(page, many=True) try: # Get min and max values for list_price, rental_price and living_area from django.db.models import Min, Max min_list_price = self.queryset.aggregate(Min('list_price'))['list_price__min'] max_list_price = self.queryset.aggregate(Max('list_price'))['list_price__max'] min_rental_price = self.queryset.aggregate(Min('rental_price'))['rental_price__min'] max_rental_price = self.queryset.aggregate(Max('rental_price'))['rental_price__max'] min_living_area = self.queryset.aggregate(Min('living_area'))['living_area__min'] max_living_area = self.queryset.aggregate(Max('living_area'))['living_area__max'] # Filter data filtered_data = filter_data_dict(serializer) response = { 'count': self.queryset.count(), 'num_pages': paginator.page.paginator.num_pages, 'current_count': len(serializer.data), 'next': paginator.get_next_link(), 'previous': paginator.get_previous_link(), 'min_list_price': min_list_price, 'max_list_price': max_list_price, 'min_rental_price': min_rental_price, 'max_rental_price': max_rental_price, 'min_living_area': min_living_area, 'max_living_area': max_living_area, '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 != ''} return Response(response, 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 properties', 'Error': str(e)} return Response(response, status=status.HTTP_400_BAD_REQUEST) @swagger_auto_schema(operation_description=""" Get only the property list summary. Arguments: request -- The HTTP request object. Returns: A Response object containing a summary from the Property objects as JSON. Authentication: If the user is not authenticated, return a 401 Unauthorized response """, responses={ 200: 'Summary listed', 400: 'Error listing', }, \ operation_id='summary_properties', operation_summary='List all summary details from the obtained properties', tags=['Properties']) @action (detail=False, methods=['get'], url_path='summary', url_name='summary', permission_classes=[AllowAny]) def summary(self, request): """ Return a summary of the properties Arguments: request -- The HTTP request object. Returns: A Response object containing the summary of the properties. """ try: start_time = time.time() #queryset = self.filter_queryset(self.get_queryset()) try: # Get min and max values for list_price, rental_price and living_area from django.db.models import Min, Max min_list_price = self.queryset.aggregate(Min('list_price'))['list_price__min'] max_list_price = self.queryset.aggregate(Max('list_price'))['list_price__max'] min_rental_price = self.queryset.aggregate(Min('rental_price'))['rental_price__min'] max_rental_price = self.queryset.aggregate(Max('rental_price'))['rental_price__max'] min_living_area = self.queryset.aggregate(Min('living_area'))['living_area__min'] max_living_area = self.queryset.aggregate(Max('living_area'))['living_area__max'] # Filter data response = { 'count': self.queryset.count(), 'min_list_price': min_list_price, 'max_list_price': max_list_price, 'min_rental_price': min_rental_price, 'max_rental_price': max_rental_price, 'min_living_area': min_living_area, 'max_living_area': max_living_area, 'response_time': time.time() - start_time, } # Remove empty or None values from the response response = {key: value for key, value in response.items() if value is not None and value != ''} return Response(response, 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: response = {'message': 'Error listing properties', 'Error': str(e)} return Response(response, status=status.HTTP_400_BAD_REQUEST) @swagger_auto_schema(operation_description=""" List all Property objects in a simple list. Foreign Key depth 0 Arguments: request -- The HTTP request object. Returns: A Response object containing all the Property objects as JSON. Authentication: If the user is not authenticated, return a 401 Unauthorized response """, responses={ 200: 'Properties listed', 400: 'Error listing', }, \ operation_id='simple_list_properties', operation_summary='List all the obtained properties', tags=['Properties']) @action(detail=False, methods=['get'], url_path='simple_list', url_name='simple_list', permission_classes=[AllowAny]) def simple_list(self, request): """ Return a simple list of the properties Arguments: request -- The HTTP request object. Returns: A Response object containing a simple list of the properties. """ try: start_time = time.time() # Get min and max values for list_price, rental_price and living_area from django.db.models import Min, Max min_list_price = self.queryset.aggregate(Min('list_price'))['list_price__min'] max_list_price = self.queryset.aggregate(Max('list_price'))['list_price__max'] min_rental_price = self.queryset.aggregate(Min('rental_price'))['rental_price__min'] max_rental_price = self.queryset.aggregate(Max('rental_price'))['rental_price__max'] min_living_area = self.queryset.aggregate(Min('living_area'))['living_area__min'] max_living_area = self.queryset.aggregate(Max('living_area'))['living_area__max'] queryset = self.filter_queryset(self.get_queryset()) paginator = SetPagination() page = paginator.paginate_queryset(queryset, request) serializer = PropertyRawSerializer(page, many=True) filtered_data = filter_data_dict(serializer) try: # Get min and max values for list_price, rental_price and living_area from django.db.models import Min, Max min_list_price = self.queryset.aggregate(Min('list_price'))['list_price__min'] max_list_price = self.queryset.aggregate(Max('list_price'))['list_price__max'] min_rental_price = self.queryset.aggregate(Min('rental_price'))['rental_price__min'] max_rental_price = self.queryset.aggregate(Max('rental_price'))['rental_price__max'] min_living_area = self.queryset.aggregate(Min('living_area'))['living_area__min'] max_living_area = self.queryset.aggregate(Max('living_area'))['living_area__max'] response = { 'count': self.queryset.count(), 'num_pages': paginator.page.paginator.num_pages, 'current_count': len(serializer.data), 'next': paginator.get_next_link(), 'previous': paginator.get_previous_link(), 'min_list_price': min_list_price, 'max_list_price': max_list_price, 'min_rental_price': min_rental_price, 'max_rental_price': max_rental_price, 'min_living_area': min_living_area, 'max_living_area': max_living_area, '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 != ''} return Response(response, 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: response = {'message': 'Error listing properties', 'Error': str(e)} return Response(response, status=status.HTTP_400_BAD_REQUEST) property_id_param = openapi.Parameter('id', openapi.IN_PATH, description="Property register ID", type=openapi.TYPE_INTEGER, required=True) manual_parameters = [property_id_param] @swagger_auto_schema(operation_description=""" Retrieve a specific Property object by ID. Arguments: request -- The HTTP request object. pk -- The ID of the Property object to retrieve. Returns: A Response object containing the retrieved Property object as JSON. """, manual_parameters=manual_parameters,\ responses={ 200: 'Property fields listed', 400: 'Data not Found', }, \ operation_id='simple_retrieve_property', operation_summary='Retrieve a given property from database using the property Id.', tags=['Properties']) @action(detail=True, methods=['get'], url_path=r'simple_detail', url_name='simple_detail', permission_classes=[AllowAny]) def simple_retrieve(self, request, pk=None): """ Return a simple Properties object. Foreign Key depth 0 Arguments: request -- The HTTP request object. Returns: A Response object containing Properties detail. """ try: start_time = time.time() serializer = PropertyRawSerializer(self.queryset.get(pk=pk)) try: filtered_data = filter_data_dict(serializer) filtered_data['response_time'] = time.time() - start_time return Response(filtered_data, status=status.HTTP_200_OK) except Exception as e: return Response({'message': f'Error trying to retrieve item {pk} => {e}'}, status=status.HTTP_400_BAD_REQUEST) except Exception as e: return Response({'message': f'Item {pk} not found => {e}'}, status=status.HTTP_404_NOT_FOUND) @swagger_auto_schema(operation_description=""" List all Property objects directly from mongodn database Foreign Key depth 0 Arguments: request -- The HTTP request object. Returns: A Response object containing all the Property objects as JSON. Authentication: If the user is not authenticated, return a 401 Unauthorized response """, responses={ 200: 'Properties listed', 400: 'Error listing', }, \ operation_id='complete_list_properties', operation_summary='List all the obtained properties directly from mongodb', tags=['Properties']) @action(detail=False, methods=['get'], url_path='complete_list', url_name='complete_list', permission_classes=[AllowAny]) def complete_list(self, request): """ Return a complete list of the properties, directly from mongdb. Arguments: request -- The HTTP request object. Returns: A Response object containing a simple list of the properties. """ try: start_time = time.time() PropertiesComplete.connect_to_mongodb() queryset = PropertiesComplete.objects.all() fields_to_filter = list(PropertyFilter.get_fields()) # Initialize a filter dictionary filter_dict = {} # Iterate over request parameters and add filters for available fields for field in fields_to_filter: if field in request.query_params: filter_dict[field] = request.query_params[field] # Apply filters to the queryset if filter_dict: queryset = queryset.filter(**filter_dict) queryset_dict = list(queryset.as_pymongo()) for item in queryset_dict: item.pop('_id', None) paginator = SetPagination() paginated_queryset = paginator.paginate_queryset(queryset_dict, request) filtered_data = filter_queryset_dict(paginated_queryset) # Initialize variables with default values min_list_price = max_list_price = min_rental_price = max_rental_price = min_living_area = max_living_area = None try: # Get min and max values for list_price, rental_price and living_area pipeline = [ { '$group': { '_id': None, 'min_list_price': {'$min': '$list_price'}, 'max_list_price': {'$max': '$list_price'}, 'min_rental_price': {'$min': '$rental_price'}, 'max_rental_price': {'$max': '$rental_price'}, 'min_living_area': {'$min': '$living_area'}, 'max_living_area': {'$max': '$living_area'}, } } ] result_cursor = queryset.aggregate(*pipeline) result = next(result_cursor, None) if result: min_list_price = result.get('min_list_price') max_list_price = result.get('max_list_price') min_rental_price = result.get('min_rental_price') max_rental_price = result.get('max_rental_price') min_living_area = result.get('min_living_area') max_living_area = result.get('max_living_area') response = { 'count': queryset.count(), 'num_pages': paginator.page.paginator.num_pages, 'current_count': len(paginated_queryset), 'next': paginator.get_next_link(), 'previous': paginator.get_previous_link(), 'min_list_price': min_list_price, 'max_list_price': max_list_price, 'min_rental_price': min_rental_price, 'max_rental_price': max_rental_price, 'min_living_area': min_living_area, 'max_living_area': max_living_area, '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 != ''} return Response(response, 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: response = {'message': 'Error listing properties', 'Error': str(e)} return Response(response, status=status.HTTP_400_BAD_REQUEST) ### Methods not allowed @swagger_auto_schema(operation=None, auto_schema=None) def create(self, request): response = {'message': 'Method not allowed!'} return Response(response, status=status.HTTP_405_METHOD_NOT_ALLOWED) @swagger_auto_schema(operation=None, auto_schema=None) def update(self, request, pk=None): response = {'message': 'Method not allowed!'} return Response(response, status=status.HTTP_405_METHOD_NOT_ALLOWED) @swagger_auto_schema(operation=None, auto_schema=None) def destroy(self, request, pk=None): response = {'message': 'Method not allowed'} return Response(response, status=status.HTTP_405_METHOD_NOT_ALLOWED) @swagger_auto_schema(operation=None, auto_schema=None) def partial_update(self, request, pk=None): response = {'message': 'Method not allowed!'} return Response(response, status=status.HTTP_405_METHOD_NOT_ALLOWED) class PropertyMediaFilter(django_filters.FilterSet): """ main = django_filters.filters.ModelMultipleChoiceFilter( field_name='main', to_field_name='main', queryset=PropertiesMedia.objects.only('main').order_by('main').distinct() ) """ class Meta: model = PropertiesMedia fields = { 'property_media_id': ['in', 'exact'], 'type': ['contains'], 'main':['exact'], 'url': ['contains'], 'property__property_id': ['exact'], 'property__listing_id': ['exact'], 'property__title': ['icontains'], 'property__transaction_type': ['icontains'], 'property__detail_view_url': ['icontains'], 'property__usage__usage_name_en': ['icontains'], 'property__usage__usage_name_pt': ['icontains'], 'property__type__type_name_en': ['icontains'], 'property__type__type_name_pt': ['icontains'], 'property__description': ['icontains'], 'property__list_price': ['range','exact'], 'property__rental_price': ['range','exact'], 'property__list_price_currency': ['exact'], 'property__rental_price_currency': ['exact'], 'property__living_area': ['range', 'exact'], 'property__living_area_unit': ['exact'], 'property__year_built': ['exact'], 'property__bedrooms': ['exact', 'range', 'icontains'], 'property__bathrooms': ['exact', 'range', 'icontains'], 'property__garage': ['exact', 'range', 'icontains'], 'property__garage_type': ['exact'], 'property__unit_floor': ['exact', 'range', 'icontains'], 'property__unit_number': ['exact'], 'property__publish_date': ['exact', 'range', 'icontains'], 'property__sync_date': ['exact', 'range'], 'property__region_id': ['exact'], } class PropertyMediaViewSet(viewsets.ReadOnlyModelViewSet): """ A viewset for handling Property Media objects. Methods Allowed: retrieve -- Retrieve a specific Property Media object by ID. list -- List all Property Media objects. Methods Not Allowed: create -- Create a new Property Media object. update -- Update an existing Property Media object. destroy -- Delete an existing Property Media object. """ queryset = PropertiesMedia.objects.using('mongodb').all() serializer_class = PropertyMediaSerializer name = "Medias" description = "Properties Media Endpoint" filter_backends = [django_filters.DjangoFilterBackend, filters.SearchFilter, filters.OrderingFilter] filterset_class = PropertyMediaFilter # if not settings.DEBUG: # authentication_classes = [TokenAuthentication] # authentication_classes = [JWTAuthentication] search_fields= [] ordering_fields = '__all__' ordering = ['property_media_id',] pagination_class = PageNumberPagination allowed_methods = ['GET', 'POST', 'HEAD', 'OPTIONS', 'TRACE'] def get_permissions(self): """ Instantiates and returns the list of permissions that this view requires. """ if self.action in ['list', 'retrieve']: permission_classes = [IsAuthenticatedOrReadOnly] pass else: permission_classes = [IsAdminUser] return [permission() for permission in permission_classes] property_media_id_param = openapi.Parameter('id', openapi.IN_PATH, description="Property Media register ID", type=openapi.TYPE_INTEGER, required=True) manual_parameters = [property_media_id_param] @swagger_auto_schema(operation_description=""" Retrieve a specific PropertyMedia object by ID. Arguments: request -- The HTTP request object. pk -- The ID of the PropertyMedia object to retrieve. Returns: A Response object containing the retrieved PropertyMedia object as JSON. """, manual_parameters=manual_parameters,\ responses={ 200: 'PropertyMedia fields listed', 400: 'Data not Found', }, \ operation_id='retrieve_propertymedia', operation_summary='Retrieve a given propertymedia from database using the propertymedia Id.', tags=['PropertiesMedia']) def retrieve(self, request, pk=None): """ Retrieve a specific PropertyMedia object by ID. Arguments: request -- The HTTP request object. pk -- The ID of the PropertyMedia object to retrieve. Returns: A Response object containing the retrieved PropertyMedia object as JSON. """ try: start_time = time.time() serializer = self.serializer_class(self.queryset.get(property_media_id=pk)) try: filtered_data = filter_data_dict(serializer) filtered_data['response_time'] = time.time() - start_time return Response(filtered_data, status=status.HTTP_200_OK) except Exception as e: return Response({'message': f'Error trying to retrieve item {pk} => {e}'}, status=status.HTTP_400_BAD_REQUEST) except Exception as e: return Response({'message': f'Item {pk} not found => {e}'}, status=status.HTTP_404_NOT_FOUND) @swagger_auto_schema(operation_description=""" List all PropertyMedia objects. Arguments: request -- The HTTP request object. Returns: A Response object containing all the PropertyMedia objects as JSON. Authentication: If the user is not authenticated, return a 401 Unauthorized response """, responses={ 200: 'PropertiesMedia listed', 400: 'Error listing', }, \ operation_id='list_propertiesmedia', operation_summary='List all the obtained propertiesmedia', tags=['PropertiesMedia']) def list(self, request): try: start_time = time.time() queryset = self.filter_queryset(self.get_queryset()) paginator = SetPagination() page = paginator.paginate_queryset(queryset, request) try: serializer = self.get_serializer(page, many=True) try: filtered_data = filter_data_dict(serializer) response = get_list_response(self.queryset, serializer, filtered_data, paginator) response['count'] = queryset.count() response['total_count'] = self.queryset.count() response['response_time'] = time.time() - start_time return Response(response, 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': f'No Item found => {e}'}, status=status.HTTP_404_NOT_FOUND) except Exception as e: response = {'message': f'Error to return items => {e}'} return Response(response, status=status.HTTP_400_BAD_REQUEST) ### Method not allowed @swagger_auto_schema(operation=None, auto_schema=None) def create(self, request): response = {'message': 'Method not allowed!'} return Response(response, status=status.HTTP_405_METHOD_NOT_ALLOWED) @swagger_auto_schema(operation=None, auto_schema=None) def update(self, request, pk=None): response = {'message': 'Method not allowed!'} return Response(response, status=status.HTTP_405_METHOD_NOT_ALLOWED) @swagger_auto_schema(operation=None, auto_schema=None) def destroy(self, request, pk=None): response = {'message': 'Method not allowed'} return Response(response, status=status.HTTP_405_METHOD_NOT_ALLOWED) @swagger_auto_schema(operation=None, auto_schema=None) def partial_update(self, request, pk=None): response = {'message': 'Method not allowed!'} return Response(response, status=status.HTTP_405_METHOD_NOT_ALLOWED) class PropertyFeatureFilter(django_filters.FilterSet): """ item = django_filters.filters.ModelMultipleChoiceFilter( field_name='item', to_field_name='item', queryset=PropertiesFeatures.objects.all(), ) """ class Meta: model = PropertiesFeatures fields = { 'property_feature_id': ['in', 'exact'], 'feature__feature_name_en': ['in', 'exact', 'contains'], 'feature__feature_name_pt': ['in', 'exact', 'contains'], 'property__property_id': ['exact'], 'property__listing_id': ['exact'], 'property__listing_id': ['exact'], 'property__title': ['icontains'], 'property__transaction_type': ['icontains'], 'property__detail_view_url': ['icontains'], 'property__usage__usage_name_en': ['icontains'], 'property__usage__usage_name_pt': ['icontains'], 'property__type__type_name_en': ['icontains'], 'property__type__type_name_pt': ['icontains'], 'property__description': ['icontains'], 'property__list_price': ['range','exact'], 'property__rental_price': ['range','exact'], 'property__list_price_currency': ['exact'], 'property__rental_price_currency': ['exact'], 'property__living_area': ['range', 'exact'], 'property__living_area_unit': ['exact'], 'property__year_built': ['exact'], 'property__bedrooms': ['exact', 'range', 'icontains'], 'property__bathrooms': ['exact', 'range', 'icontains'], 'property__garage': ['exact', 'range', 'icontains'], 'property__garage_type': ['exact'], 'property__unit_floor': ['exact', 'range', 'icontains'], 'property__unit_number': ['exact'], 'property__publish_date': ['exact', 'range', 'icontains'], 'property__sync_date': ['exact', 'range'], 'property__region_id': ['exact'], } class PropertyFeatureViewSet(viewsets.ReadOnlyModelViewSet): """ A viewset for handling Property Feature objects. Methods Allowed: retrieve -- Retrieve a specific Property Feature object by ID. list -- List all Property Feature objects. Methods Not Allowed: create -- Create a new Property Feature object. update -- Update an existing Property Feature object. destroy -- Delete an existing Property Feature object. """ queryset = PropertiesFeatures.objects.using('mongodb').all() serializer_class = PropertyFeatureSerializer name = "Features" description = "Properties Feature Endpoint" filter_backends = [django_filters.DjangoFilterBackend, filters.SearchFilter, filters.OrderingFilter] filterset_class = PropertyFeatureFilter # if not settings.DEBUG: # authentication_classes = [TokenAuthentication] # authentication_classes = [JWTAuthentication] search_fields = '__all__' ordering_fields = '__all__' ordering = ['property_feature_id',] pagination_class = PageNumberPagination allowed_methods = ['GET', 'POST', 'HEAD', 'OPTIONS', 'TRACE'] def get_permissions(self): """ Instantiates and returns the list of permissions that this view requires. """ if self.action in ['list', 'retrieve']: permission_classes = [IsAuthenticatedOrReadOnly] pass else: permission_classes = [IsAdminUser] return [permission() for permission in permission_classes] property_feature_id_param = openapi.Parameter('id', openapi.IN_PATH, description="Property Feature register ID", type=openapi.TYPE_INTEGER, required=True) manual_parameters = [property_feature_id_param] @swagger_auto_schema(operation_description=""" Retrieve a specific PropertyFeature object by ID. Arguments: request -- The HTTP request object. pk -- The ID of the PropertyFeature object to retrieve. Returns: A Response object containing the retrieved PropertyFeature object as JSON. """, manual_parameters=manual_parameters,\ responses={ 200: 'PropertyFeature fields listed', 400: 'Data not Found', }, \ operation_id='retrieve_propertymedia', operation_summary='Retrieve a given propertymedia from database using the propertymedia Id.', tags=['PropertiesFeature']) def retrieve(self, request, pk=None): """ Retrieve a specific PropertyFeature object by ID. Arguments: request -- The HTTP request object. pk -- The ID of the PropertyFeature object to retrieve. Returns: A Response object containing the retrieved PropertyFeature object as JSON. """ try: start_time = time.time() serializer = self.serializer_class(self.queryset.get(property_feature_id=pk)) try: filtered_data = filter_data_dict(serializer) filtered_data['response_time'] = time.time() - start_time return Response(filtered_data, status=status.HTTP_200_OK) except Exception as e: return Response({'message': f'Error trying to retrieve item {pk} => {e}'}, status=status.HTTP_400_BAD_REQUEST) except Exception as e: return Response({'message': f'Item {pk} not found => {e}'}, status=status.HTTP_404_NOT_FOUND) @swagger_auto_schema(operation_description=""" List all PropertyFeature objects. Arguments: request -- The HTTP request object. Returns: A Response object containing all the PropertyFeature objects as JSON. Authentication: If the user is not authenticated, return a 401 Unauthorized response """, responses={ 200: 'PropertiesFeature listed', 400: 'Error listing', }, \ operation_id='list_propertiesmedia', operation_summary='List all the obtained propertiesmedia', tags=['PropertiesFeature']) def list(self, request): try: start_time = time.time() queryset = self.filter_queryset(self.get_queryset()) paginator = SetPagination() page = paginator.paginate_queryset(queryset, request) try: serializer = self.get_serializer(page, many=True) try: filtered_data = filter_data_dict(serializer) response = get_list_response(self.queryset, serializer, filtered_data, paginator) response['count'] = queryset.count() response['total_count'] = self.queryset.count() response['response_time'] = time.time() - start_time return Response(response, 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': f'No Item found => {e}'}, status=status.HTTP_404_NOT_FOUND) except Exception as e: response = {'message': f'Error to return items => {e}'} return Response(response, status=status.HTTP_400_BAD_REQUEST) ### Method not allowed @swagger_auto_schema(operation=None, auto_schema=None) def create(self, request): response = {'message': 'Method not allowed!'} return Response(response, status=status.HTTP_405_METHOD_NOT_ALLOWED) @swagger_auto_schema(operation=None, auto_schema=None) def update(self, request, pk=None): response = {'message': 'Method not allowed!'} return Response(response, status=status.HTTP_405_METHOD_NOT_ALLOWED) @swagger_auto_schema(operation=None, auto_schema=None) def destroy(self, request, pk=None): response = {'message': 'Method not allowed'} return Response(response, status=status.HTTP_405_METHOD_NOT_ALLOWED) @swagger_auto_schema(operation=None, auto_schema=None) def partial_update(self, request, pk=None): response = {'message': 'Method not allowed!'} return Response(response, status=status.HTTP_405_METHOD_NOT_ALLOWED) ### View with Methods Not Allowed for this endpoint ### MEthods Create, Update, Destroy and Partial Updated Implemented # class LocationFilter(django_filters.FilterSet): # class Meta: # model = Location # fields = { # 'property_location_id': ['exact'], # 'display_address': ['icontains'], # 'country': ['icontains'], # 'country_abbreviation': ['icontains'], # 'state': ['icontains'], # 'city': ['icontains'], # 'zone': ['icontains'], # 'neighborhood': ['icontains'], # 'complement': ['icontains'], # 'address': ['icontains'], # 'street_number': ['icontains'], # 'postal_code': ['icontains'], # } # class LocationViewSet(viewsets.ModelViewSet): # """ # A viewset for handling Location objects. # Methods Allowed: # retrieve -- Retrieve a specific Location object by ID. # list -- List all Location objects. # Methods Not Allowed: # create -- Create a new Location object. # update -- Update an existing Location object. # destroy -- Delete an existing Location object. # """ # name = "Locations" # description = "Locations Endpoint" # queryset = Location.objects.using('mongodb').all() # serializer_class = LocationSerializer # if not settings.DEBUG: # authentication_classes = [TokenAuthentication] # authentication_classes = [JWTAuthentication] # filter_backends = [django_filters.DjangoFilterBackend, filters.SearchFilter, filters.OrderingFilter] # filterset_class = LocationFilter # search_fields=['country', 'country_abbreviation', 'state', 'city', 'zone', 'neighborhood', 'complement', # 'address', 'street_number', 'postal_code'] # ordering_fields = '__all__' # ordering = ['property_location_id'] # pagination_class = PageNumberPagination # allowed_methods = ['GET', 'POST', 'HEAD', 'OPTIONS', 'TRACE'] # def get_permissions(self): # """ # Instantiates and returns the list of permissions that this view requires. # """ # if self.action in ['list', 'retrieve']: # permission_classes = [IsAuthenticated] # else: # permission_classes = [IsAdminUser] # return [permission() for permission in permission_classes] # location_id_param = openapi.Parameter('id', openapi.IN_PATH, description="Location register ID", type=openapi.TYPE_INTEGER, required=True) # manual_parameters = [location_id_param] # @swagger_auto_schema(operation_description=""" # Retrieve a specific Location object by ID. # Arguments: # request -- The HTTP request object. # pk -- The ID of the Location object to retrieve. # Returns: # A Response object containing the retrieved Location object as JSON. # """, manual_parameters=manual_parameters,\ # responses={ # 200: 'Location fields listed', # 400: 'Data not Found', }, \ # operation_id='retrieve_location', operation_summary='Retrieve a given location from database using the location Id.', tags=['Locations']) # def retrieve(self, request, pk=None): # """ # Retrieve a specific Location object by ID. # Arguments: # request -- The HTTP request object. # pk -- The ID of the Location object to retrieve. # Returns: # A Response object containing the retrieved Location object as JSON. # """ # serializer_dict = self.serializer_class(self.queryset.get(pk=pk)).data # authorized = request.user.is_authenticated # return retrieve_filter(serializer_dict, authorized) # @swagger_auto_schema(operation_description=""" # List all Location objects. # Arguments: # request -- The HTTP request object. # Returns: # A Response object containing all the Location objects as JSON. # Authentication: # If the user is not authenticated, return a 401 Unauthorized response # """, # responses={ # 200: 'Locations listed', # 400: 'Error listing', }, \ # operation_id='list_locations', operation_summary='List all the obtained locations', tags=['Locations']) # def list(self, request): # if not request.user.is_authenticated: # return Response({'message': 'Authentication credentials were not provided.'}, status=status.HTTP_401_UNAUTHORIZED) # else: # try: # start_time = time.time() # queryset = self.filter_queryset(self.queryset) # total_count = queryset.count() # if request.GET.get(settings.REST_FRAMEWORK['PAGE_SIZE_QUERY_PARAM']): # page_size = request.GET.get(settings.REST_FRAMEWORK['PAGE_SIZE_QUERY_PARAM'], self.pagination_class.page_size) # self.pagination_class.page_size = page_size # else: # page_size = settings.REST_FRAMEWORK['PAGE_SIZE'] # self.pagination_class.page_size = page_size # set back to default value # page = self.paginate_queryset(queryset) # if page is not None: # serializer_data = self.serializer_class(page, many=True).data # else: # serializer_data = self.serializer_class(queryset, many=True).data # if not serializer_data: # return Response({'message': 'No Location found'}, status=status.HTTP_404_NOT_FOUND) # # Filter out null or empty fields and exclude openai_api_key # for data in serializer_data: # for key in list(data.keys()): # if data[key] is None or data[key] == '': # del data[key] # current_count = len(serializer_data) # end_time = time.time() # response_time = end_time - start_time # response = {'count': {'total_count':total_count, 'current_count': current_count, 'response_time':response_time}, 'results': serializer_data} # return Response(response, status=status.HTTP_200_OK) # except Exception as e: # response = {'message': 'Error listing locations', 'Error': str(e)} # return Response(response, status=status.HTTP_400_BAD_REQUEST) # @swagger_auto_schema(operation_description=""" # Create a new Location object. # Arguments: # request -- The HTTP request object. # Returns: # A Response object containing the created Location object as JSON. # """, \ # responses={ # 201: 'Location created', # 400: 'Error creating', }, \ # operation_id='create_location', operation_summary='Create a new location', tags=['Locations']) # def create(self, request): # """ # Create a new Location object. # Arguments: # request -- The HTTP request object. # Returns: # A Response object containing the created Location object as JSON. # """ # if not request.user.is_authenticated: # return Response({'message': 'Authentication credentials were not provided.'}, status=status.HTTP_401_UNAUTHORIZED) # else: # try: # serializer = self.serializer_class(data=request.data) # if serializer.is_valid(): # serializer.save() # return Response(serializer.data, status=status.HTTP_201_CREATED) # else: # return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST) # except Exception as e: # response = {'message': 'Error creating location', 'Error': str(e)} # return Response(response, status=status.HTTP_400_BAD_REQUEST) # @swagger_auto_schema(operation_description=""" # Update an existing Location object. # Arguments: # request -- The HTTP request object. # pk -- The ID of the Location object to update. # Returns: # A Response object containing the updated Location object as JSON. # """, \ # responses={ # 200: 'Location updated', # 400: 'Error updating', }, \ # operation_id='update_location', operation_summary='Update an existing location', tags=['Locations']) # def update(self, request, pk=None): # """ # Update an existing Location object. # Arguments: # request -- The HTTP request object. # pk -- The ID of the Location object to update. # Returns: # A Response object containing the updated Location object as JSON. # """ # if not request.user.is_authenticated: # return Response({'message': 'Authentication credentials were not provided.'}, status=status.HTTP_401_UNAUTHORIZED) # else: # try: # item = self.queryset.get(pk=pk) # serializer = self.serializer_class(item, data=request.data) # if serializer.is_valid(): # serializer.save() # return Response(serializer.data, status=status.HTTP_200_OK) # else: # return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST) # except Exception as e: # response = {'message': f'Error updating location with id {pk}', 'Error': str(e)} # return Response(response, status=status.HTTP_400_BAD_REQUEST) # @swagger_auto_schema(operation_description=""" # Delete an existing Location object. # Arguments: # request -- The HTTP request object. # pk -- The ID of the Location object to delete. # Returns: # A Response object indicating that the Location object was deleted. # """, \ # responses={ # 204: 'Location deleted', # 400: 'Error deleting', }, \ # operation_id='delete_location', operation_summary='Delete an existing location', tags=['Locations']) # def destroy(self, request, pk=None): # """ # Delete an existing Location object. # Arguments: # request -- The HTTP request object. # pk -- The ID of the Location object to delete. # Returns: # A Response object indicating that the Location object was deleted. # """ # if not request.user.is_authenticated: # return Response({'message': 'Authentication credentials were not provided.'}, status=status.HTTP_401_UNAUTHORIZED) # else: # try: # item = self.queryset.get(pk=pk) # item.delete() # return Response(status=status.HTTP_204_NO_CONTENT) # except Exception as e: # response = {'message': f'Error deleting location with id {pk}', 'Error': str(e)} # return Response(response, status=status.HTTP_400_BAD_REQUEST) # @swagger_auto_schema(operation_description=""" # Partially update an existing Location object. # Arguments: # request -- The HTTP request object. # pk -- The ID of the Location object to partially update. # Returns: # A Response object containing the updated Location object as JSON. # """, \ # responses={ # 200: 'Location updated', # 400: 'Error updating', }, \ # operation_id='partial_update_location', operation_summary='Partially update an existing location', tags=['Locations']) # def partial_update(self, request, pk=None): # """ # Partially update an existing Location object. # Arguments: # request -- The HTTP request object. # pk -- The ID of the Location object to partially update. # Returns: # A Response object containing the updated Location object as JSON. # """ # if not request.user.is_authenticated: # return Response({'message': 'Authentication credentials were not provided.'}, status=status.HTTP_401_UNAUTHORIZED) # else: # try: # item = self.queryset.get(pk=pk) # serializer = self.serializer_class(item, data=request.data, partial=True) # if serializer.is_valid(): # serializer.save() # return Response(serializer.data, status=status.HTTP_200_OK) # else: # return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST) # except Exception as e: # response = {'message': f'Error partially updating location with id {pk}', 'Error': str(e)} # return Response(response, status=status.HTTP_400_BAD_REQUEST) # class PropertyNestedFilter(django_filters.FilterSet): # list_price= django_filters.RangeFilter( # field_name='list_price', # ) # living_area= django_filters.RangeFilter( # field_name='living_area', # ) # publish_date= django_filters.DateFilter( # field_name='publish_date', # ) # sync_date= django_filters.DateFilter( # field_name='sync_date', # ) # # Define filters for nested fields # office_office_id = django_filters.CharFilter(field_name='office__office_id', lookup_expr='exact') # office_name = django_filters.CharFilter(field_name='office__name', lookup_expr='icontains') # office_website = django_filters.CharFilter(field_name='office__website', lookup_expr='icontains') # office_telephone = django_filters.CharFilter(field_name='office__telephone', lookup_expr='icontains') # office_country = django_filters.CharFilter(field_name='office__country', lookup_expr='icontains') # office_country_abbreviation = django_filters.CharFilter(field_name='office__country_abbreviation', lookup_expr='exact') # office_state = django_filters.CharFilter(field_name='office__state', lookup_expr='icontains') # office_state_abbreviation = django_filters.CharFilter(field_name='office__state_abbreviation', lookup_expr='exact') # office_city = django_filters.CharFilter(field_name='office__city', lookup_expr='icontains') # office_zone = django_filters.CharFilter(field_name='office__zone', lookup_expr='icontains') # office_neighborhood = django_filters.CharFilter(field_name='office__neighborhood', lookup_expr='icontains') # office_complement = django_filters.CharFilter(field_name='office__complement', lookup_expr='icontains') # office_address = django_filters.CharFilter(field_name='office__address', lookup_expr='icontains') # office_street_number = django_filters.CharFilter(field_name='office__street_number', lookup_expr='exact') # office_postal_code = django_filters.CharFilter(field_name='office__postal_code', lookup_expr='exact') # office_total_properties = django_filters.CharFilter(field_name='office__total_properties', lookup_expr='exact') # agente_agent_id = django_filters.CharFilter(field_name='agent__agent_id', lookup_expr='exact') # agent_name = django_filters.CharFilter(field_name='agent__name', lookup_expr='icontains') # agent_email = django_filters.CharFilter(field_name='agent__email', lookup_expr='icontains') # agent_total_properties = django_filters.CharFilter(field_name='agent__total_properties', lookup_expr='exact') # usage_usage_id = django_filters.CharFilter(field_name='usage__usage_id', lookup_expr='exact') # usage_usage_name_en = django_filters.CharFilter(field_name='usage__usage_name_en', lookup_expr='icontains') # usage_usage_name_pt = django_filters.CharFilter(field_name='usage__usage_name_pt', lookup_expr='icontains') # type_type_id = django_filters.CharFilter(field_name='type__type_id', lookup_expr='exact') # type_type_name_en = django_filters.CharFilter(field_name='type__type_name_en', lookup_expr='icontains') # type_type_name_pt = django_filters.CharFilter(field_name='type__type_name_pt', lookup_expr='icontains') # location_property_location_id = django_filters.CharFilter(field_name='location__property_location_id', lookup_expr='exact') # location_display_address = django_filters.CharFilter(field_name='location__display_address', lookup_expr='icontains') # location_country = django_filters.CharFilter(field_name='location__country', lookup_expr='icontains') # location_country_abbreviation = django_filters.CharFilter(field_name='location__country_abbreviation', lookup_expr='icontains') # location_state = django_filters.CharFilter(field_name='location__state', lookup_expr='icontains') # location_zone = django_filters.CharFilter(field_name='location__zone', lookup_expr='icontains') # location_neighborhood = django_filters.CharFilter(field_name='location__neighborhood', lookup_expr='icontains') # location_complement = django_filters.CharFilter(field_name='location__complement', lookup_expr='icontains') # location_address = django_filters.CharFilter(field_name='location__address', lookup_expr='icontains') # location_street_number = django_filters.CharFilter(field_name='location__street_number', lookup_expr='icontains') # location_postal_code = django_filters.CharFilter(field_name='location__postal_code', lookup_expr='icontains') # location_lat = django_filters.CharFilter(field_name='location__lat', lookup_expr='exact') # location_lng = django_filters.CharFilter(field_name='location__lng', lookup_expr='exact') # medias_url = django_filters.CharFilter(field_name='medias__url', lookup_expr='icontains') # features_feature_name_pt = django_filters.CharFilter(field_name='features__feature_name_pt', lookup_expr='icontains') # features_feature_name_en = django_filters.CharFilter(field_name='features__feature_name_en', lookup_expr='icontains') # class Meta: # model = PropertiesNested # fields = { # 'property_id': ['in', 'exact'], # 'listing_id': ['in', 'exact'], # 'title': ['icontains'], # 'transaction_type': ['icontains'], # 'detail_view_url': ['icontains'], # 'description': ['icontains'], # 'list_price': ['range','exact'], # 'list_price_currency': ['exact'], # 'rental_price' : ['range','exact'], # 'rental_price_currency': ['exact'], # 'rental_price_period': ['icontains'], # 'property_administration_fee' : ['range','exact'], # 'property_administration_fee_period': ['icontains'], # 'yearly_tax': ['range', 'exact'], # 'yearly_tax_currency': ['exact'], # 'living_area': ['range', 'exact'], # 'living_area_unit': ['exact'], # 'year_built' : ['in', 'exact'], # 'bedrooms': ['in', 'exact', 'range','gte'], # 'bathrooms': ['in','exact', 'range','gte'], # 'garage': ['in', 'exact', 'range', 'gte'], # 'garage_type': ['exact'], # 'unit_floor': ['in', 'exact', 'range'], # 'unit_number': ['in', 'exact'], # 'publish_date': ['exact', 'range', 'icontains'], # 'sync_date': ['exact', 'range'], # 'region_id': ['in', 'exact'], # 'office_office_id': ['in', 'exact'], # 'office_name': ['in', 'exact','icontains'], # 'office_website': ['icontains'], # 'office_telephone': ['icontains'], # 'office_country': ['icontains'], # 'office_country_abbreviation': ['exact'], # 'office_state': ['icontains'], # 'office_state_abbreviation': ['exact'], # 'office_city': ['icontains'], # 'office_zone': ['icontains'], # 'office_neighborhood': ['icontains'], # 'office_complement': ['icontains'], # 'office_address': ['icontains'], # 'office_street_number': ['exact'], # 'office_postal_code': ['exact'], # 'office_total_properties': ['in', 'exact', 'range'], # 'agent_agent_id': ['exact'], # 'agent_name': ['exact', 'icontains'], # 'agent_email': ['icontains'], # 'agent_total_properties':['in', 'exact', 'range'], # 'usage_usage_id': ['exact'], # 'usage_usage_name_en': ['exact', 'icontains'], # 'usage_usage_name_pt': ['exact', 'icontains'], # 'type_type_id': ['exact'], # 'type_type_name_en': ['exact', 'icontains'], # 'type_type_name_pt': ['exact', 'icontains'], # 'location_property_location_id': ['in', 'exact'], # 'location_display_address': ['icontains'], # 'location_country': ['in', 'exact','icontains'], # 'location_country_abbreviation': ['in','icontains'], # 'location_state': ['in', 'icontains'], # 'location_zone': ['in', 'icontains'], # 'location_neighborhood': ['in', 'icontains'], # 'location_complement': ['icontains'], # 'location_address': ['in', 'icontains'], # 'location_street_number': ['icontains'], # 'location_postal_code': ['icontains'], # 'location_lat': ['exact'], # 'location_lng': ['exact'], # 'medias_url': ['icontains'], # 'features_feature_name_en': ['in', 'icontains'], # 'features_feature_name_pt': ['in', 'icontains'], # } # from rest_framework import filters # class JSONFieldFilterBackend(filters.BaseFilterBackend): # def filter_queryset(self, request, queryset, view): # # Get the query parameters from the request # params = request.query_params # # Loop through the query parameters to find JSONField filters # json_field_filters = {} # for param, value in params.items(): # # Check if the parameter starts with the name of the JSONField # if param.startswith('office__'): # # Extract the field name and value from the parameter # field_name = param.replace('office__', '') # # Construct the filter keyword argument dynamically # filter_kwargs = {'office__contains': {field_name: value}} # print("Filter kwargs: ", filter_kwargs) # # Perform manual filtering on the queryset # #queryset = queryset.filter(**filter_kwargs) # queryset = queryset.filter(office__office_id=value) # return queryset # class MongoFilterBackend: # def __init__(self, filter_fields): # self.filter_fields = filter_fields # self.filter_types = { # 'exact': lambda field, value: {field: int(value) if field.endswith('_id') else value}, # 'icontains': lambda field, value: {field: {'$regex': value, '$options': 'i'}}, # 'in': lambda field, value: {field: {'$in': [v.strip() for v in value.split(',')] if ',' in value else [value]}}, # 'range': lambda field, value: {field: {'$gte': float(value.split(',')[0]), '$lte': float(value.split(',')[1])}}, # 'gte': lambda field, value: {field: {'$gte': float(value)}}, # 'lt': lambda field, value: {field: {'$lt': float(value)}}, # 'lte': lambda field, value: {field: {'$lte': float(value)}}, # 'gt': lambda field, value: {field: {'$gt': float(value)}}, # 'regex': lambda field, value: {field: {'$regex': value}}, # 'iregex': lambda field, value: {field: {'$regex': value, '$options': 'i'}}, # 'iexact': lambda field, value: {field: {'$regex': f"^{value}$", '$options': 'i'}}, # 'startswith': lambda field, value: {field: {'$regex': f"^{value}"}}, # 'istartswith': lambda field, value: {field: {'$regex': f"^{value}", '$options': 'i'}}, # 'endswith': lambda field, value: {field: {'$regex': f"{value}$"}}, # 'iendswith': lambda field, value: {field: {'$regex': f"{value}$", '$options': 'i'}} # } # def get_mongo_match(self, query_params): # match = {} # for key, values in query_params.items(): # if not values: # continue # value = values[0] if isinstance(values, list) else values # parts = key.split('__') # field = '__'.join(parts[:-1]) if len(parts) > 1 and parts[-1] in self.filter_types else key # filter_type = parts[-1] if len(parts) > 1 and parts[-1] in self.filter_types else 'exact' # if field in self.filter_fields and filter_type in self.filter_fields[field]: # method = self.filter_types.get(filter_type) # field = field.replace('__', '.') # match.update(method(field, value)) # return match # def build_aggregation_pipeline(self, query_params): # match = self.get_mongo_match(query_params) # pipeline = [ # {'$match': match}, # {'$project': {'property_id': 1, '_id': 0}} # Adjust projection as needed # ] # return pipeline # def get_mongo_match(query_params, filter_fields): # """Generates a MongoDB match dictionary based on the Django-style query parameters.""" # match = {} # #print(f"Original Query Params: {query_params}") # Log incoming parameters # # Accepted MongoDB filter types # filter_types = { # 'exact': lambda field, value: {field: value if field == "listing_id" else int(value) if field.endswith('_id') else value}, # 'icontains': lambda field, value: {field: {'$regex': value, '$options': 'i'}}, # 'in': lambda field, value: {field: {'$in': [v.strip() for v in value.split(',')] if ',' in value else [value]}}, # 'range': lambda field, value: {field: {'$gte': float(value.split(',')[0]), '$lte': float(value.split(',')[1])}}, # 'gte': lambda field, value: {field: {'$gte': float(value)}}, # 'lt': lambda field, value: {field: {'$lt': float(value)}}, # 'lte': lambda field, value: {field: {'$lte': float(value)}}, # 'gt': lambda field, value: {field: {'$gt': float(value)}}, # 'regex': lambda field, value: {field: {'$regex': value}}, # 'iregex': lambda field, value: {field: {'$regex': value, '$options': 'i'}}, # 'iexact': lambda field, value: {field: {'$regex': f"^{value}$", '$options': 'i'}}, # 'startswith': lambda field, value: {field: {'$regex': f"^{value}"}}, # 'istartswith': lambda field, value: {field: {'$regex': f"^{value}", '$options': 'i'}}, # 'endswith': lambda field, value: {field: {'$regex': f"{value}$"}}, # 'iendswith': lambda field, value: {field: {'$regex': f"{value}$", '$options': 'i'}} # } # not_supported = ['ordering', 'page', 'page_size'] # for key, values in query_params.items(): # if not values: # continue # if key not in not_supported: # value = values[0] if isinstance(values, list) else values # Ensure proper value extraction # parts = key.split('__') # field = '__'.join(parts[:-1]) if len(parts) > 1 and parts[-1] in filter_types else key # filter_type = parts[-1] if len(parts) > 1 and parts[-1] in filter_types else 'exact' # # Process field if it's recognized and applicable filter is provided # if field in filter_fields and filter_type in filter_fields[field]: # method = filter_types.get(filter_type, lambda f, v: {f: v}) # ### change __ to . in field # field = field.replace('__', '.') # match.update(method(field, value)) # #print(f"Processed {field} with filter {filter_type} to {match[field]}") # else: # print(f"Field {field} or filter {filter_type} not recognized or not applicable.") # #print(f"Final MongoDB match criteria: {match}") # return match # def build_aggregation_pipeline(request): # """Builds an aggregation pipeline based on the request's query parameters.""" # filter_fields = { # 'property_id': ['in', 'exact'], # 'listing_id': ['in', 'exact'], # 'title': ['icontains'], # 'transaction_type': ['icontains'], # 'detail_view_url': ['icontains'], # 'description': ['icontains'], # 'list_price': ['range','exact'], # 'list_price_currency': ['exact'], # 'rental_price' : ['range','exact'], # 'rental_price_currency': ['exact'], # 'rental_price_period': ['icontains'], # 'property_administration_fee' : ['range','exact'], # 'property_administration_fee_period': ['icontains'], # 'yearly_tax': ['range', 'exact'], # 'yearly_tax_currency': ['exact'], # 'living_area': ['range', 'exact'], # 'living_area_unit': ['exact'], # 'year_built' : ['in', 'exact'], # 'bedrooms': ['in', 'exact', 'range','gte'], # 'bathrooms': ['in','exact', 'range','gte'], # 'garage': ['in', 'exact', 'range', 'gte'], # 'garage_type': ['exact'], # 'unit_floor': ['in', 'exact', 'range'], # 'unit_number': ['in', 'exact'], # 'publish_date': ['exact', 'range', 'icontains'], # 'sync_date': ['exact', 'range'], # 'region_id': ['in', 'exact'], # 'office__office_id': ['in', 'exact'], # 'office__name': ['in', 'exact','icontains'], # 'office__website': ['icontains'], # 'office__telephone': ['icontains'], # 'office__country': ['icontains'], # 'office__country_abbreviation': ['exact'], # 'office__state': ['icontains'], # 'office__state_abbreviation': ['exact'], # 'office__city': ['icontains'], # 'office__zone': ['icontains'], # 'office__neighborhood': ['icontains'], # 'office__complement': ['icontains'], # 'office__address': ['icontains'], # 'office__street_number': ['exact'], # 'office__postal_code': ['exact'], # 'office__total_properties': ['in', 'exact', 'range'], # 'agent__agent_id': ['exact','in','contains'], # 'agent__name': ['exact', 'icontains'], # 'agent__email': ['icontains'], # 'agent__total_properties':['in', 'exact', 'range'], # 'usage__usage_id': ['exact'], # 'usage__usage_name_en': ['exact', 'icontains'], # 'usage__usage_name_pt': ['exact', 'icontains'], # 'type__type_id': ['exact'], # 'type__type_name_en': ['exact', 'icontains'], # 'type__type_name_pt': ['exact', 'icontains'], # 'location__property_location_id': ['in', 'exact'], # 'location__display_address': ['icontains'], # 'location__country': ['in', 'exact','icontains'], # 'location__country_abbreviation': ['in','icontains'], # 'location__state': ['in', 'icontains'], # 'location__zone': ['in', 'icontains'], # 'location__neighborhood': ['in', 'icontains'], # 'location__city': ['in', 'icontains'], # 'location__complement': ['icontains'], # 'location__address': ['in', 'icontains'], # 'location__street_number': ['icontains'], # 'location__postal_code': ['icontains'], # 'location__lat': ['exact'], # 'location__lng': ['exact'], # 'medias__url': ['icontains'], # 'features__feature_name_en': ['in', 'icontains'], # 'features__feature_name_pt': ['in', 'icontains'], # } # match = get_mongo_match(request, filter_fields) # pipeline = [ # {'$match': match}, # {'$project': {'property_id': 1, '_id': 0}} # Example projection # ] # #print(f"Pipeline: {pipeline}") # return pipeline def get_mongo_match(query_params, exclude_fields, search_fields=[]): match = {} # Define MongoDB filter types filter_types = { 'exact': lambda field, value: {field: int(value) if field.endswith('_id') and field != 'listing_id' else value}, 'precise': lambda field, value: {field: float(value)}, 'icontains': lambda field, value: {field: {'$regex': value, '$options': 'i'}}, #'in': lambda field, value: {field: {'$in': [v.strip() for v in value.split(',')] if ',' in value else [value]}}, 'in': lambda field, value: {field: {'$in': [ObjectId(v.strip()) if ObjectId.is_valid(v.strip()) else (float(v.strip()) if v.strip().replace('.', '', 1).isdigit() else v.strip()) for v in value.split(',')] if ',' in value else [ObjectId(value) if ObjectId.is_valid(value) else (float(value) if value.replace('.', '', 1).isdigit() else value)]}}, 'in_obj': lambda field, value: {field: {'$in': [ObjectId(v.strip()) for v in value.split(',')] if ',' in value else [ObjectId(value)]}}, 'in_str': lambda field, value: {field: {'$in': [v.strip() for v in value.split(',')] if ',' in value else [value]}}, 'in_num': lambda field, value: {field: {'$in': [float(v.strip()) for v in value.split(',')] if ',' in value else [float(value)]}}, 'range': lambda field, value: {field: {'$gte': float(value.split(',')[0]), '$lte': float(value.split(',')[1])}}, 'gte': lambda field, value: {field: {'$gte': float(value)}}, 'lt': lambda field, value: {field: {'$lt': float(value)}}, 'lte': lambda field, value: {field: {'$lte': float(value)}}, 'gt': lambda field, value: {field: {'$gt': float(value)}}, 'regex': lambda field, value: {field: {'$regex': value}}, 'iregex': lambda field, value: {field: {'$regex': value, '$options': 'i'}}, 'iexact': lambda field, value: {field: {'$regex': f"^{value}$", '$options': 'i'}}, 'startswith': lambda field, value: {field: {'$regex': f"^{value}"}}, 'istartswith': lambda field, value: {field: {'$regex': f"^{value}", '$options': 'i'}}, 'endswith': lambda field, value: {field: {'$regex': f"{value}$"}}, 'iendswith': lambda field, value: {field: {'$regex': f"{value}$", '$options': 'i'}}, 'exists': lambda field, value: {field: {'$exists': True if value.lower() == 'true' else False}} } for key, values in query_params.items(): if not values: continue if key == 'ordering' or key == 'page_size' or key == 'page': continue value = values[0] if isinstance(values, list) else values # Decodificar a URL para garantir que espaços são convertidos corretamente value = urllib.parse.unquote(value) parts = key.split('__') field = '__'.join(parts[:-1]) if len(parts) > 1 and parts[-1] in filter_types else key filter_type = parts[-1] if len(parts) > 1 and parts[-1] in filter_types else 'exact' # if field in filter_fields and filter_type in filter_fields[field]: # method = filter_types.get(filter_type, lambda f, v: {f: v}) # field = field.replace('__', '.') # match.update(method(field, value)) special_params = ['has_medias', 'has_features', 'has_lat_lng','location_search', 'location_search_or', 'search', 'geo', 'medias_limit', 'page', 'page_size', 'ordering','fields', 'save_params','geo2'] if exclude_fields and field in exclude_fields: continue elif field in special_params: continue else: method = filter_types.get(filter_type, lambda f, v: {f: v}) field = field.replace('__', '.') match.update(method(field, value)) ### Search for existence or not of medias and features if 'has_medias' in query_params: if query_params.get('has_medias')[0] == '1': match.update({'medias': {'$exists': True, '$ne': []}}) else: match.update({'medias': {'$exists': True, '$eq': []}}) if 'has_features' in query_params: if query_params.get('has_features')[0] == '1': match.update({'features': {'$exists': True, '$ne': []}}) else: match.update({'features': {'$exists': True, '$eq': []}}) if 'has_lat_lng' in query_params: if query_params.get('has_lat_lng')[0] == '1': match.update({'location.lat': {'$exists': True, '$ne': None, '$ne': [], '$ne':""}, 'location.lng': {'$exists': True, '$ne': None, '$ne': [], '$ne':""}}) else: match.update({'location.lat': {'$exists': True, '$eq': None, '$eq': [], '$eq':""}, 'location.lng': {'$exists': True, '$eq': None, '$eq': [], '$eq':""}}) if 'search' in query_params: #match = {} search = urllib.parse.unquote(query_params.get('search')[0]) search_match = {'$or': [{field: {'$regex': search, '$options': 'i'}} for field in search_fields]} match.update(search_match) ### google maps return = Sao Paulo,SP,Brasil ### location_search = Sao Paulo,SP,Brasil ### extrair os 3 dados: city, state, country e buscar tal qual location__state__icontains=Sao Paulo&location__city=Sao Paulo&location__country=Brasil siglas_para_nomes_estados = { 'AC': 'Acre', 'AL': 'Alagoas', 'AP': 'Amapá', 'AM': 'Amazonas', 'BA': 'Bahia', 'CE': 'Ceará', 'DF': 'Distrito Federal', 'ES': 'Espírito Santo', 'GO': 'Goiás', 'MA': 'Maranhão', 'MT': 'Mato Grosso', 'MS': 'Mato Grosso do Sul', 'MG': 'Minas Gerais', 'PA': 'Pará', 'PB': 'Paraíba', 'PR': 'Paraná', 'PE': 'Pernambuco', 'PI': 'Piauí', 'RJ': 'Rio de Janeiro', 'RN': 'Rio Grande do Norte', 'RS': 'Rio Grande do Sul', 'RO': 'Rondônia', 'RR': 'Roraima', 'SC': 'Santa Catarina', 'SP': 'São Paulo', 'SE': 'Sergipe', 'TO': 'Tocantins' } if 'location_search' in query_params: location_search = urllib.parse.unquote(query_params.get('location_search')[0]) location_search = location_search.split(',') city = location_search[0] state = location_search[1] country = location_search[2] state = siglas_para_nomes_estados.get(state, state) match.update({ 'location.state': {'$regex': state, '$options': 'i'}, 'location.city': {'$regex': city, '$options': 'i'}, 'location.country': {'$regex': country, '$options': 'i'} }) if 'location_search_or' in query_params: location_search_or = urllib.parse.unquote(query_params.get('location_search_or')[0]) location_search_or = location_search_or.split('),(') city_in = [] state_in = [] country_in = [] for location in location_search_or: city, state, country = location.split(',') city_in.append(city.replace('(','').replace(')','')) state_in.append(state.replace('(','').replace(')','')) country_in.append(country.replace('(','').replace(')','')) state_in = [siglas_para_nomes_estados.get(state, state) for state in state_in] match.update({ '$or': [ { 'location.city': {'$regex': city_in[i], '$options': 'i'}, 'location.state': {'$regex': state_in[i], '$options': 'i'}, 'location.country': {'$regex': country_in[i], '$options': 'i'} } for i in range(len(city_in)) ] }) if 'geo' in query_params: # extract lat, lng and radius lat, lng, radius = map(float, query_params['geo'][0].split(',')) if lat and lng and radius: # distance = haversine((-23.533773, -46.625290),(lat,lng), unit=Unit.METERS) # distance_difference = distance - radius # distance_sum = distance + radius # distance_range = [distance_difference, distance_sum] #print(f"Latitude: {lat}, Longitude: {lng}, Radius: {radius}") # Convert radius to degrees # Giving more precision to longer distances if radius > 1000: radius = radius - 100 radius_in_degrees_lat = radius / 111000.0 import math radius_in_degrees_lng = radius / (111000.0 * math.cos(math.radians(lat))) # Calculate bounding box min_lat = round(lat - radius_in_degrees_lat,6) max_lat = round(lat + radius_in_degrees_lat,6) min_lng = round(lng - radius_in_degrees_lng,6) max_lng = round(lng + radius_in_degrees_lng,6) #print(f"Bounding Box: Min lat: {min_lat}, Max lat: {max_lat}, Min lng: {min_lng}, Max lng: {max_lng}") # lat__gte=min_lat, # lat__lte=max_lat, # lng__gte=min_lng, # lng__lte=max_lng match.update({ 'lat': {'$gte': min_lat, '$lte': max_lat}, 'lng': {'$gte': min_lng, '$lte': max_lng} }) query_params_keys = [key for key in query_params if key.startswith('query_params__')] if len(query_params_keys) > 0: for key in query_params_keys: value = query_params[key][0] if key.find('[') != -1: nested_key = key[key.find('[')+1:key.find(']')] else: nested_key = key[key.find('__')+2:] match.update({f"query_params.{nested_key}": {'$regex': f".*{value}.*", '$options': 'i'}}) return match def save_query_params(query_params, url, model, user=None): if query_params and 'save_params' in query_params and query_params.get('save_params')[0] == '1': query_params.pop('save_params', None) # Flatten and sort query_params query_params = [{k: v[0] if isinstance(v, list) else v for k, v in query_params.items()}] # Ordering query_params query_params_sorted = dict(sorted(query_params[0].items())) # Prepare additional fields for MongoDB #additional_fields = {f"query_param_{k}": v for k, v in query_params_sorted.items()} additional_fields = {k: v for k, v in query_params_sorted.items()} from django.utils import timezone created_at = timezone.now() if user and not user.is_anonymous: from panel.models import QueryParamsUser user_id = [user.id] saved_params = QueryParamsUser.objects.mongo_find({'user_id': user_id, 'origin': model}) if saved_params and saved_params.count() > 0: saved_params_id = saved_params[0]['_id'] try: QueryParamsUser.objects.mongo_update_one( {'id': ObjectId(saved_params_id)}, {'$set': {'query_params': [query_params_sorted], 'url': url, 'origin': model, 'updated_at': created_at}} ) except Exception as e: logger.error(f"Error updating query params: {e}") else: try: QueryParamsUser.objects.mongo_insert( {'user_id': user_id, 'query_params': [query_params_sorted], 'url': url, 'origin': model, 'created_at': created_at} ) except Exception as e: logger.error(f"Error saving query params: {e}") from panel.models import QueryParamsGeneral # save general query params try: document = { 'query_params': [query_params_sorted], 'url': url, 'origin': model, 'created_at': created_at, **additional_fields } QueryParamsGeneral.objects.mongo_insert(document) except Exception as e: logger.error(f"Error saving query params: {e}") else: logger.info("No user or query params to save") def build_aggregation_pipeline(request, search_fields=None, user=None, model=None): query_params = dict(request.query_params) match = get_mongo_match(query_params, search_fields) pipeline = [{'$match': match}] # Handle ordering ordering = request.query_params.get('ordering', '-property_id').strip() sort_order = -1 if ordering.startswith('-') else 1 sort_field = ordering.lstrip('-') # if sort_field == 'list_price' or sort_field == 'rental_price' or sort_field == 'property_administration_fee' or sort_field == 'yearly_tax' or sort_field == 'living_area': # sort_field_converted = f"{sort_field}_converted" # pipeline.append({'$addFields': {sort_field_converted: {'$toDouble': f"${sort_field}"}}}) pipeline.append({'$sort': {sort_field: sort_order}}) #print(pipeline) if 'geo2' in request.query_params: geo_ids = filter_based_geo(request, 'v1_properties_nested', pipeline) if geo_ids: filtered_ids = [property['property_id'] for property in geo_ids] distances = {property['property_id']: property['distance'] for property in geo_ids} if filtered_ids: match['property_id'] = {'$in': filtered_ids} page_size = len(filtered_ids) page = 1 else: # return no results match['property_id'] = {'$in': []} # Handle pagination ## check if variable page_size already exists if 'page_size' and 'page' in locals(): page_size = page_size page = page else: page_size = int(request.query_params.get('page_size', SetPagination.page_size)) page = int(request.query_params.get('page', 1)) skip = (page - 1) * page_size pipeline.extend([{'$skip': skip}, {'$limit': page_size}]) # Optionally add projection to reduce fields returned # if 'fields' in request.query_params: # fields = request.query_params['fields'].split(',') # projection = {field: 1 for field in fields} # projection['_id'] = 0 # pipeline.append({'$project': projection}) pipeline.append({'$project': {'property_id': 1, '_id': 0}}) # Pipeline Filter only with filters without limit, returning count documents pipeline_filter = [{'$match': match}] #Count the documents pipeline_filter.append({'$count': 'total'}) #print(f"Pipeline: {pipeline}") #print(f"Pipeline Filter: {pipeline_filter}") return pipeline, pipeline_filter # def filter_based_geo(request, collection, pipeline): # from haversine import haversine, Unit # geo = request.query_params.get('geo', None) # if geo: # try: # lat, lng, radius = map(float, geo.split(',')) # print(f"Lat: {lat}, {lng}, {radius}") # except Exception as e: # return None # if lat is not None and lng is not None and radius is not None: # from .connections import MongoDBConnection # db = MongoDBConnection().get_db() # collection = db[collection] # def is_within_radius(property_location): # property_lat = float(property_location[0]['lat']) # property_lng = float(property_location[0]['lng']) # property_coords = (property_lat, property_lng) # center_coords = (lat, lng) # print(f"Center Coords: {center_coords}, {property_coords}") # distance = haversine(center_coords, property_coords, unit=Unit.METERS) # print(f"Distance: {distance}") # print(f"Valid?{distance <= radius}") # return distance, distance <= radius # filtered_ids = [] # #print(f"Pipeline from GEO: {pipeline}") # collection = list(collection.aggregate(pipeline)) # for property in collection: # if property['location']: # property_location = property['location'] # if property_location[0]['lat'] and property_location[0]['lng']: # distance, within_radius = is_within_radius(property_location) # if within_radius: # property_location[0]['distance_meters'] = distance # if distance <= radius: # filtered_ids.append({property['property_id']: distance}) # return filtered_ids def filter_based_geo(request, collection, pipeline): geo = request.query_params.get('geo', None) if geo: try: lat, lng, radius = map(float, geo.split(',')) except ValueError: return None # Invalid geo parameters if lat is not None and lng is not None and radius is not None: from .connections import MongoDBConnection db = MongoDBConnection().get_db() collection = db[collection] def is_within_radius(property_location): try: property_lat = float(property_location[0]['lat']) property_lng = float(property_location[0]['lng']) except (KeyError, ValueError, TypeError): return None, False # Invalid lat/lng in property_location property_coords = (property_lat, property_lng) center_coords = (lat, lng) # Calculate the distance in meters distance = haversine(center_coords, property_coords, unit=Unit.METERS) # Check if distance is within the radius (including radius = 0) return distance, distance <= radius filtered_properties = [] pipeline.append({ '$project': { 'property_id': 1, 'location.lng': 1, 'location.lat': 1, '_id': 0 } }) properties = list(collection.aggregate(pipeline)) for property in properties: if property['location']: property_location = property['location'] if 'lat' in property_location[0] and 'lng' in property_location[0]: distance, within_radius = is_within_radius(property_location) # For radius = 0, only include exact matches (distance == 0) if within_radius and (radius > 0 or distance == 0): property_location[0]['distance_meters'] = distance filtered_properties.append({ 'property_id': property['property_id'], 'distance': distance }) #print(f"Filtered Properties: {filtered_properties}") return filtered_properties else: return None from urllib.parse import urlencode, urlparse, urlunparse, parse_qs def build_absolute_uri(request, new_params): url_parts = list(urlparse(request.build_absolute_uri())) query = parse_qs(url_parts[4]) query.update(new_params) url_parts[4] = urlencode(query, doseq=True) return urlunparse(url_parts) class CustomViewSet(viewsets.ReadOnlyModelViewSet): # def filter_based_geo(self): # from haversine import haversine, Unit # # geo = self.request.query_params.get('geo', None) # if geo: # lat, lng, radius = map(float, geo.split(',')) # # if lat is not None and lng is not None and radius is not None: # # def is_within_radius(property_location): # property_lat = float(property_location['lat']) # property_lng = float(property_location['lng']) # property_coords = (property_lat, property_lng) # center_coords = (lat, lng) # distance = haversine(center_coords, property_coords, unit=Unit.METERS) # return distance, distance <= radius # # filtered_queryset = [] # for property in queryset: # if property.location: # property_location = property.location[0] # distance, within_radius = is_within_radius(property_location) # if within_radius: # property_location['distance_meters'] = distance # if distance <= radius: # filtered_queryset.append(property) # # return filtered_queryset 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(',') # Prepare a structure to handle nested fields nested_fields = {} for field in fields: parts = field.split('__') current = nested_fields for part in parts[:-1]: current = current.setdefault(part, {}) current[parts[-1]] = 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 the value is a dictionary and we have nested fields, recurse if isinstance(v, dict) and keys[k]: result[k] = filter_dict(v, keys[k]) # If the value is a list, handle lists of dictionaries or other nested structures elif isinstance(v, list) and keys[k]: # Recursively apply filtering for each dictionary in the list result[k] = [filter_dict(item, keys[k]) if isinstance(item, dict) else item 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}") logger.error(f"Error applying field filters: {e}") return data # def get_queryset(self): # """Fetches the queryset based on the dynamically built aggregation pipeline.""" # query_params = dict(self.request.query_params) # pipeline = build_aggregation_pipeline(query_params) # query = PropertiesNested.objects.mongo_aggregate(pipeline) # queryset = list(query) # # Assuming property_id is the identifier to fetch the full object # property_ids = [item['property_id'] for item in queryset] # final_queryset = PropertiesNested.objects.filter(property_id__in=property_ids) # ordering = self.request.query_params.get('ordering', None) # if ordering: # final_queryset = final_queryset.order_by(ordering) # else: # final_queryset = final_queryset.order_by('property_id') # return final_queryset def get_queryset(self): unique_identifier_field = getattr(self, 'unique_identifier_field', self.model._meta.pk.name) user = self.request.user model = self.model._meta.model_name pipeline, pipeline_filter = build_aggregation_pipeline(self.request, self.search_fields, user=user, model=model) #Collation Object collation = {"locale": "pt", "strength": 1} from pymongo.collation import Collation collation = Collation(locale="pt", strength=1) results = list(self.model.objects.mongo_aggregate(pipeline, collation=collation)) total = list(self.model.objects.mongo_aggregate(pipeline_filter, collation=collation)) #print(f"Results: {results}") #print(f"Total: {total}") ids = [r[unique_identifier_field] for r in results] #print(f"IDs: {ids}") queryset = self.model.objects.filter(property_id__in=ids) #print(f"Queryset: {queryset}") queryset.total = total[0]['total'] if total else 0 return queryset def get_permissions(self): """ Instantiates and returns the list of permissions that this view requires. """ if self.action in ['list', 'retrieve','summary','simple_list','simple_retrieve', 'complete_list']: permission_classes = [IsAuthenticatedOrReadOnly] else: permission_classes = [IsAdminUser] return [permission() for permission in permission_classes] property_id_param = openapi.Parameter('id', openapi.IN_PATH, description="Property register ID", type=openapi.TYPE_INTEGER, required=True) manual_parameters = [property_id_param] @swagger_auto_schema(operation_description=""" Retrieve a specific Property object by ID. Arguments: request -- The HTTP request object. pk -- The ID of the Property object to retrieve. Returns: A Response object containing the retrieved Property object as JSON. """, manual_parameters=manual_parameters,\ responses={ 200: 'Property fields listed', 400: 'Data not Found', }, \ operation_id='retrieve_property_nested', operation_summary='Retrieve a given property from database using the property Id.', tags=['Properties']) def retrieve(self, request, pk=None): """ Retrieve a specific Property object by ID. Arguments: request -- The HTTP request object. pk -- The ID of the Property object to retrieve. Returns: A Response object containing the retrieved Property object as JSON. """ try: start_time = time.time() serializer = self.serializer_class(self.queryset.get(property_id=pk)) try: filtered_data = filter_data_dict(serializer) #print(f"Filtered data: {filtered_data}") filtered_data['response_time'] = time.time() - start_time # if request.query_params.get('cidade'): # if request.query_params.get('cidade') == '1': # # from panel.models import Cidade, Estado, Bairro # # from panel.serializers import CidadeReadOnlySerializer, EstadoSerializer, BairroReadOnlySerializer # try: # state = filtered_data['location'][0]['state'] # #Eprint(f"State: {state}") # # Fetch the state object from the database # estado_instance = State.objects.filter(state_name__iexact=state).first() # if estado_instance: # # If a state was found, serialize it # estado_serializer = StateSerializer(estado_instance) # estado = estado_serializer.data # #print("Estado found and serialized.") # else: # # Handle the case where no state was found # #print("No matching estado found.") # estado = [] # except: # estado = [] # try: # if estado['id']: # city = str(filtered_data['location'][0]['city']).strip() # #print(f"City: {city}") # from bson import ObjectId # # Fetch the city object from the database # cidade_instance = City.objects.filter(city_name__iexact=city, state_id=int(estado['id'])).first() # if cidade_instance: # # If a city was found, serialize it # cidade_serializer = CitySerializer(cidade_instance) # cidade = cidade_serializer.data # #print("Cidade found and serialized.") # else: # # Handle the case where no city was found # #print("No matching cidade found.") # cidade = [] # except: # cidade = [] # #print(f"Cidade: {cidade}") # try: # if estado['id'] and cidade['id']: # neighborhood = str(filtered_data['location'][0]['neighborhood']).strip() # #print(f"Neighborhood: {neighborhood}") # zone = str(filtered_data['location'][0]['zone']).strip() # #print(f"Zone: {zone}") # from bson import ObjectId # ### if neighborhood is not None, search for neighborhood, else search for zone # if neighborhood: # bairro = NeighborhoodSerializer(Neighborhood.objects.filter(neighborhood_name__iexact=neighborhood,city_id=[int(cidade['id'])],state_id=[int(estado['id'])]).first()).data # else: # bairro = NeighborhoodSerializer(Neighborhood.objects.filter(neighborhood_name__iexact=zone,city_id=[int(cidade['id'])],state_id=[int(estado['id'])]).first()).data # #print(f"Bairro: {bairro}") # except: # bairro = [] # #print(f"Bairro: {bairro}") # if 'id' in bairro: # filtered_data['location'][0]['info'] = [ # { # "bairro":[bairro] # } # ] # elif 'id' in cidade: # filtered_data['location'][0]['info'] = [ # { # "cidade":[cidade] # } # ] # elif 'id' in estado: # filtered_data['location'][0]['info'] = [ # { # "estado":[estado] # } # ] return Response(filtered_data, status=status.HTTP_200_OK) except Exception as e: return Response({'message': f'Error trying to retrieve item {pk} => {e}'}, status=status.HTTP_400_BAD_REQUEST) except Exception as e: return Response({'message': f'Item {pk} not found => {e}'}, status=status.HTTP_404_NOT_FOUND) @swagger_auto_schema(operation_description=""" List all Property objects. Arguments: request -- The HTTP request object. Returns: A Response object containing all the Property objects as JSON. Authentication: If the user is not authenticated, return a 401 Unauthorized response """, responses={ 200: 'Properties listed', 400: 'Error listing', }, \ operation_id='list_properties_nested', operation_summary='List all the obtained properties', tags=['Properties']) def list(self, request): try: start_time = time.time() queryset = self.get_queryset() #print(f"Queryset: {queryset}") # from inspect import currentframe, getframeinfo # frame = currentframe() # frameinfo = getframeinfo(frame) # print(f"File: {frameinfo.filename}, Line: {frameinfo.lineno}") # if "geo" in request.query_params: # try: # print(f"File: {__file__}") # queryset = self.filter_based_geo(queryset) # except Exception as e: # return Response({'message': f'Error generating geo response => {e}'}, status=status.HTTP_400_BAD_REQUEST) try: filtered_queryset = self.filter_queryset(queryset) serializer = self.get_serializer(filtered_queryset, many=True) #serializer = self.serializer_class(queryset, many=True) # Save query params if requested and data is available if len(serializer.data) > 0: if 'save_params' in self.request.query_params: query_params = dict(self.request.query_params) save_query_params(query_params=query_params, user=request.user, url=request.build_absolute_uri(), model=self.model._meta.model_name) try: # Get min and max values for list_price, rental_price and living_area from django.db.models import Min, Max min_list_price = self.queryset.aggregate(Min('list_price'))['list_price__min'] max_list_price = self.queryset.aggregate(Max('list_price'))['list_price__max'] min_rental_price = self.queryset.aggregate(Min('rental_price'))['rental_price__min'] max_rental_price = self.queryset.aggregate(Max('rental_price'))['rental_price__max'] min_living_area = self.queryset.aggregate(Min('living_area'))['living_area__min'] max_living_area = self.queryset.aggregate(Max('living_area'))['living_area__max'] page_size = int(request.query_params.get('page_size', SetPagination.page_size)) page = int(request.query_params.get('page', 1)) if request.query_params.get('geo') is not None: total = len(queryset) else: 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 = None else: next = build_absolute_uri(request, {'page': page + 1}) if page <= 1: previous = None else: previous = build_absolute_uri(request, {'page': page - 1}) filtered_data = serializer.data if "fields" in request.query_params: filtered_data = self.apply_field_filters(serializer.data) #print(f"Fields Filtered Data: {filtered_data}") if "medias_limit" in request.query_params: medias_limit = int(request.query_params.get('medias_limit', 0)) for item in filtered_data: if 'medias' in item: item['medias'] = item['medias'][:medias_limit] if total==total_per_page: total_pages = 1 next = None response = { 'count': total, 'num_pages': total_pages, 'current_count': total_per_page, 'next':next, 'previous':previous, 'min_list_price': min_list_price, 'max_list_price': max_list_price, 'min_rental_price': min_rental_price, 'max_rental_price': max_rental_price, 'min_living_area': min_living_area, 'max_living_area': max_living_area, '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 properties', 'Error': str(e)} return Response(response, status=status.HTTP_400_BAD_REQUEST) # class PropertiesNestedViewSet(CustomViewSet): # """ # A viewset for handling Property objects. # Methods Allowed: # retrieve -- Retrieve a specific Property object by ID. # list -- List all Property objects. # Methods Not Allowed: # create -- Create a new Property object. # update -- Update an existing Property object. # destroy -- Delete an existing Property object. # """ # model = PropertiesNested # unique_identifier_field = 'property_id' # queryset = model.objects.using('mongodb').all() # serializer_class = PropertiesNestedSerializer # name = "Properties" # description = "Properties Endpoint" # filter_backends = [filters.SearchFilter, filters.OrderingFilter] # search_fields = ['title', 'description', 'office.name', 'agent.name', 'usage.usage_name_en', 'usage.usage_name_pt', 'type.type_name_en', 'type.type_name_pt', 'location.display_address', 'location.country', 'location.state', 'location.zone', 'location.neighborhood', 'location.address', 'location.postal_code', 'medias.url', 'features.feature_name_en', 'features.feature_name_pt'] # ordering_fields = '__all__' # ordering = ['property_id'] # pagination_class = SetPagination # allowed_methods = ['GET', 'POST', 'HEAD', 'OPTIONS', 'TRACE'] # exclude_fields = [] # filter_fields = { # 'property_id': ['in', 'exact','in_num'], # 'listing_id': ['in', 'exact'], # 'title': ['icontains'], # 'transaction_type': ['icontains'], # 'detail_view_url': ['icontains'], # 'description': ['icontains'], # 'list_price': ['range','exact'], # 'list_price_currency': ['exact'], # 'rental_price' : ['range','exact'], # 'rental_price_currency': ['exact'], # 'rental_price_period': ['icontains'], # 'property_administration_fee' : ['range','exact'], # 'property_administration_fee_period': ['icontains'], # 'yearly_tax': ['range', 'exact'], # 'yearly_tax_currency': ['exact'], # 'living_area': ['range', 'exact'], # 'living_area_unit': ['exact'], # 'year_built' : ['in', 'exact'], # 'bedrooms': ['in', 'exact', 'range','gte','lte','lt','precise'], # 'bathrooms': ['in','exact', 'range','gte'], # 'garage': ['in', 'exact', 'range', 'gte'], # 'garage_type': ['exact'], # 'unit_floor': ['in', 'exact', 'range'], # 'unit_number': ['in', 'exact'], # 'publish_date': ['exact', 'range', 'icontains'], # 'sync_date': ['exact', 'range'], # 'region_id': ['in', 'exact'], # 'office__office_id': ['in', 'exact'], # 'office__name': ['in', 'exact','icontains'], # 'office__website': ['icontains'], # 'office__telephone': ['icontains'], # 'office__country': ['icontains'], # 'office__country_abbreviation': ['exact'], # 'office__state': ['icontains'], # 'office__state_abbreviation': ['exact'], # 'office__city': ['icontains'], # 'office__zone': ['icontains'], # 'office__neighborhood': ['icontains'], # 'office__complement': ['icontains'], # 'office__address': ['icontains'], # 'office__street_number': ['exact'], # 'office__postal_code': ['exact'], # 'office__total_properties': ['in', 'exact', 'range'], # 'agent__agent_id': ['exact','in','contains'], # 'agent__name': ['exact', 'icontains'], # 'agent__email': ['icontains'], # 'agent__creci': ['icontains','exact'], # 'agent__total_properties':['in', 'exact', 'range'], # 'usage__usage_id': ['exact'], # 'usage__usage_name_en': ['exact', 'icontains'], # 'usage__usage_name_pt': ['exact', 'icontains'], # 'type__type_id': ['exact'], # 'type__type_name_en': ['exact', 'icontains'], # 'type__type_name_pt': ['exact', 'icontains'], # 'location__property_location_id': ['in', 'exact'], # 'location__display_address': ['icontains'], # 'location__country': ['in', 'exact','icontains'], # 'location__country_abbreviation': ['in','icontains'], # 'location__state': ['in', 'icontains'], # 'location__zone': ['in', 'icontains'], # 'location__neighborhood': ['in', 'icontains'], # 'location__city': ['in', 'icontains'], # 'location__complement': ['icontains'], # 'location__address': ['in', 'icontains'], # 'location__street_number': ['icontains'], # 'location__postal_code': ['icontains'], # 'location__lat': ['exact'], # 'location__lng': ['exact'], # 'medias__url': ['icontains','exists'], # 'features__feature_name_en': ['in', 'icontains'], # 'features__feature_name_pt': ['in', 'icontains'], # } # def list(self, request): # return super().list(request) # def retrieve(self, request, pk=None): # return super().retrieve(request, pk) # @swagger_auto_schema(operation=None, auto_schema=None) # def create(self, request): # response = {'message': 'Method not allowed!'} # return Response(response, status=status.HTTP_405_METHOD_NOT_ALLOWED) # @swagger_auto_schema(operation=None, auto_schema=None) # def update(self, request, pk=None): # response = {'message': 'Method not allowed!'} # return Response(response, status=status.HTTP_405_METHOD_NOT_ALLOWED) # @swagger_auto_schema(operation=None, auto_schema=None) # def partial_update(self, request, pk=None): # response = {'message': 'Method not allowed!'} # return Response(response, status=status.HTTP_405_METHOD_NOT) # @swagger_auto_schema(operation=None, auto_schema=None) # def destroy(self, request, pk=None): # response = {'message': 'Method not allowed!'} # return Response(response, status=status.HTTP_405_METHOD_NOT_ALLOWED) ############################################################################################################ class PropertiesNestedViewSet(CustomViewSet): """ A viewset for handling Property objects. Methods Allowed: retrieve -- Retrieve a specific Property object by ID. list -- List all Property objects. Methods Not Allowed: create -- Create a new Property object. update -- Update an existing Property object. destroy -- Delete an existing Property object. """ model = PropertiesNested unique_identifier_field = 'property_id' queryset = model.objects.using('mongodb').all() serializer_class = PropertiesNestedSerializer name = "Properties" description = "Properties Endpoint" filter_backends = [filters.SearchFilter, filters.OrderingFilter] search_fields = ['title', 'description', 'office.name', 'agent.name', 'usage.usage_name_en', 'usage.usage_name_pt', 'type.type_name_en', 'type.type_name_pt', 'location.display_address', 'location.country', 'location.state', 'location.zone', 'location.neighborhood', 'location.address', 'location.postal_code', 'medias.url', 'features.feature_name_en', 'features.feature_name_pt'] ordering_fields = '__all__' ordering = ['property_id'] pagination_class = SetPagination allowed_methods = ['GET', 'POST', 'HEAD', 'OPTIONS', 'TRACE'] exclude_fields = [] filter_fields = { 'property_id': ['in', 'exact','in_num'], 'listing_id': ['in', 'exact'], 'title': ['icontains'], 'transaction_type': ['icontains'], 'detail_view_url': ['icontains'], 'description': ['icontains'], 'list_price': ['range','exact'], 'list_price_currency': ['exact'], 'rental_price' : ['range','exact'], 'rental_price_currency': ['exact'], 'rental_price_period': ['icontains'], 'property_administration_fee' : ['range','exact'], 'property_administration_fee_period': ['icontains'], 'yearly_tax': ['range', 'exact'], 'yearly_tax_currency': ['exact'], 'living_area': ['range', 'exact'], 'living_area_unit': ['exact'], 'year_built' : ['in', 'exact'], 'bedrooms': ['in', 'exact', 'range','gte','lte','lt','precise'], 'bathrooms': ['in','exact', 'range','gte'], 'garage': ['in', 'exact', 'range', 'gte'], 'garage_type': ['exact'], 'unit_floor': ['in', 'exact', 'range'], 'unit_number': ['in', 'exact'], 'publish_date': ['exact', 'range', 'icontains'], 'sync_date': ['exact', 'range'], 'region_id': ['in', 'exact'], 'office__office_id': ['in', 'exact'], 'office__name': ['in', 'exact','icontains'], 'office__website': ['icontains'], 'office__telephone': ['icontains'], 'office__country': ['icontains'], 'office__country_abbreviation': ['exact'], 'office__state': ['icontains'], 'office__state_abbreviation': ['exact'], 'office__city': ['icontains'], 'office__neighborhood': ['icontains'], 'office__complement': ['icontains'], 'office__address': ['icontains'], 'office__street_number': ['exact'], 'office__postal_code': ['exact'], 'office__total_properties': ['in', 'exact', 'range'], 'agent__agent_id': ['exact','in','contains'], 'agent__name': ['exact', 'icontains'], 'agent__email': ['icontains'], 'agent__creci': ['icontains','exact'], 'agent__total_properties':['in', 'exact', 'range'], 'usage__usage_id': ['exact'], 'usage__usage_name_en': ['exact', 'icontains'], 'usage__usage_name_pt': ['exact', 'icontains'], 'type__type_id': ['exact'], 'type__type_name_en': ['exact', 'icontains'], 'type__type_name_pt': ['exact', 'icontains'], 'location__property_location_id': ['in', 'exact'], 'location__display_address': ['icontains'], 'location__country': ['in', 'exact','icontains'], 'location__country_abbreviation': ['in','icontains'], 'location__state': ['in', 'icontains'], 'location__zone': ['in', 'icontains'], 'location__neighborhood': ['in', 'icontains'], 'location__city': ['in', 'icontains'], 'location__complement': ['icontains'], 'location__address': ['in', 'icontains'], 'location__street_number': ['icontains'], 'location__postal_code': ['icontains'], 'location__lat': ['exact'], 'location__lng': ['exact'], 'medias__url': ['icontains','exists'], 'features__feature_name_en': ['in', 'icontains'], 'features__feature_name_pt': ['in', 'icontains'], } def list(self, request): return super().list(request) def retrieve(self, request, pk=None): return super().retrieve(request, pk) @swagger_auto_schema(operation=None, auto_schema=None) def create(self, request): response = {'message': 'Method not allowed!'} return Response(response, status=status.HTTP_405_METHOD_NOT_ALLOWED) @swagger_auto_schema(operation=None, auto_schema=None) def update(self, request, pk=None): response = {'message': 'Method not allowed!'} return Response(response, status=status.HTTP_405_METHOD_NOT_ALLOWED) @swagger_auto_schema(operation=None, auto_schema=None) def partial_update(self, request, pk=None): response = {'message': 'Method not allowed!'} return Response(response, status=status.HTTP_405_METHOD_NOT) @swagger_auto_schema(operation=None, auto_schema=None) def destroy(self, request, pk=None): response = {'message': 'Method not allowed!'} return Response(response, status=status.HTTP_405_METHOD_NOT_ALLOWED) # class PropertySimpleViewSet(viewsets.ReadOnlyModelViewSet): # """ # A viewset for handling Property objects. # Methods Allowed: # retrieve -- Retrieve a specific Property object by ID. # list -- List all Property objects. # Methods Not Allowed: # create -- Create a new Property object. # update -- Update an existing Property object. # destroy -- Delete an existing Property object. # """ # # if not settings.DEBUG: # # authentication_classes = [TokenAuthentication] # # authentication_classes = [JWTAuthentication] # queryset = Properties.objects.using('mongodb').all() # serializer_class = PropertySimpleSerializer # name = "Properties" # description = "Properties Endpoint" # filter_backends = [django_filters.DjangoFilterBackend, filters.SearchFilter, filters.OrderingFilter] # filterset_class = PropertyFilter # search_fields = ['listing_id', 'title', 'property_type', 'description', 'list_price', 'rental_price', 'living_area', 'bedrooms', 'bathrooms', # 'year_built', 'garage', 'garage_type', 'unit_floor', 'unit_number', 'office__name', 'office__website', # 'office__telephone','agent__name', 'agent__email', 'usage__usage_name_en', 'usage__usage_name_pt', # 'type__type_name_en', 'type__type_name_pt', 'location__display_address', # 'location__country', 'location__country_abbreviation', 'location__state', # 'location__zone', 'location__neighborhood', 'location__complement', 'location__address', # 'location__postal_code', 'location__lat', 'location__lng',] # ordering_fields = '__all__' # ordering = ['property_id'] # pagination_class = SetPagination # allowed_methods = ['GET', 'HEAD', 'OPTIONS', 'TRACE'] # permission_classes = [IsAuthenticatedOrReadOnly] # def get_permissions(self): # """ # Instantiates and returns the list of permissions that this view requires. # """ # if self.action in ['list', 'retrieve','summary','simple_list','simple_retrieve', 'complete_list']: # permission_classes = [IsAuthenticatedOrReadOnly] # else: # permission_classes = [IsAdminUser] # return [permission() for permission in permission_classes] # property_id_param = openapi.Parameter('id', openapi.IN_PATH, description="Property register ID", type=openapi.TYPE_INTEGER, required=True) # manual_parameters = [property_id_param] # @swagger_auto_schema(operation_description=""" # Retrieve a specific Property object by ID. # Arguments: # request -- The HTTP request object. # pk -- The ID of the Property object to retrieve. # Returns: # A Response object containing the retrieved Property object as JSON. # """, manual_parameters=manual_parameters,\ # responses={ # 200: 'Property fields listed', # 400: 'Data not Found', }, \ # operation_id='retrieve_property_simple', operation_summary='Retrieve a given property from database using the property Id.', tags=['Properties']) # def retrieve(self, request, pk=None): # """ # Retrieve a specific Property object by ID. # Arguments: # request -- The HTTP request object. # pk -- The ID of the Property object to retrieve. # Returns: # A Response object containing the retrieved Property object as JSON. # """ # try: # start_time = time.time() # serializer = self.serializer_class(self.queryset.get(pk=pk)) # try: # filtered_data = filter_data_dict(serializer) # filtered_data['response_time'] = time.time() - start_time # return Response(filtered_data, status=status.HTTP_200_OK) # except Exception as e: # return Response({'message': f'Error trying to retrieve item {pk} => {e}'}, status=status.HTTP_400_BAD_REQUEST) # except Exception as e: # return Response({'message': f'Item {pk} not found => {e}'}, status=status.HTTP_404_NOT_FOUND) # @swagger_auto_schema(operation_description=""" # List all Property objects. # Arguments: # request -- The HTTP request object. # Returns: # A Response object containing all the Property objects as JSON. # Authentication: # If the user is not authenticated, return a 401 Unauthorized response # """, # responses={ # 200: 'Properties listed', # 400: 'Error listing', }, \ # operation_id='list_properties_simple', operation_summary='List all the obtained properties', tags=['Properties']) # def list(self, request): # try: # start_time = time.time() # queryset = self.filter_queryset(self.get_queryset()) # paginator = SetPagination() # page = paginator.paginate_queryset(queryset, request) # try: # serializer = self.get_serializer(page, many=True) # try: # # Get min and max values for list_price, rental_price and living_area # from django.db.models import Min, Max # min_list_price = self.queryset.aggregate(Min('list_price'))['list_price__min'] # max_list_price = self.queryset.aggregate(Max('list_price'))['list_price__max'] # min_rental_price = self.queryset.aggregate(Min('rental_price'))['rental_price__min'] # max_rental_price = self.queryset.aggregate(Max('rental_price'))['rental_price__max'] # min_living_area = self.queryset.aggregate(Min('living_area'))['living_area__min'] # max_living_area = self.queryset.aggregate(Max('living_area'))['living_area__max'] # filtered_data = filter_data_dict(serializer) # response = { # 'count': self.queryset.count(), # 'num_pages': paginator.page.paginator.num_pages, # 'current_count': len(serializer.data), # 'next': paginator.get_next_link(), # 'previous': paginator.get_previous_link(), # 'min_list_price': min_list_price, # 'max_list_price': max_list_price, # 'min_rental_price': min_rental_price, # 'max_rental_price': max_rental_price, # 'min_living_area': min_living_area, # 'max_living_area': max_living_area, # '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 != ''} # return Response(response, 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_404_NOT_FOUND) # except Exception as e: # response = {'message': 'Error listing properties', 'Error': str(e)} # return Response(response, status=status.HTTP_400_BAD_REQUEST) # from rest_framework.views import APIView # from rest_framework.response import Response # from rest_framework import status # class PropertiesLocationAggregateView(APIView): # def get(self, request, *args, **kwargs): # # Aggregate the data # aggregated_data = aggregate_properties_by_location() # # Serialize the data # serializer = PropertiesNestedAggregateSerializer({'location_aggregate': aggregated_data}) # return Response(serializer.data, status=status.HTTP_200_OK) # class PropertiesLocationAggregateViewSet(CustomViewSet): # """ # A viewset for aggregate location data from Property objects. # Methods Allowed: # retrieve -- Retrieve a specific Property object by ID. # list -- List all Property objects. # Methods Not Allowed: # create -- Create a new Property object. # update -- Update an existing Property object. # destroy -- Delete an existing Property object. # """ # queryset = PropertiesNested.objects.using('mongodb').all() # serializer_class = PropertiesNestedSerializer() # name = "Aggregate Locations" # description = "Aggregate Locations Endpoint" # filter_backends = [filters.SearchFilter, filters.OrderingFilter] # search_fields = '__all__' # ordering_fields = '__all__' # ordering = '__all__' # pagination_class = SetPagination # allowed_methods = ['GET', 'HEAD', 'OPTIONS', 'TRACE'] # exclude_fields = [] # filter_fields = { # 'property_id': ['in', 'exact'], # 'listing_id': ['in', 'exact'], # 'title': ['icontains'], # 'transaction_type': ['icontains'], # 'detail_view_url': ['icontains'], # 'description': ['icontains'], # 'list_price': ['range','exact'], # 'list_price_currency': ['exact'], # 'rental_price' : ['range','exact'], # 'rental_price_currency': ['exact'], # 'rental_price_period': ['icontains'], # 'property_administration_fee' : ['range','exact'], # 'property_administration_fee_period': ['icontains'], # 'yearly_tax': ['range', 'exact'], # 'yearly_tax_currency': ['exact'], # 'living_area': ['range', 'exact'], # 'living_area_unit': ['exact'], # 'year_built' : ['in', 'exact'], # 'bedrooms': ['in', 'exact', 'range','gte'], # 'bathrooms': ['in','exact', 'range','gte'], # 'garage': ['in', 'exact', 'range', 'gte'], # 'garage_type': ['exact'], # 'unit_floor': ['in', 'exact', 'range'], # 'unit_number': ['in', 'exact'], # 'publish_date': ['exact', 'range', 'icontains'], # 'sync_date': ['exact', 'range'], # 'region_id': ['in', 'exact'], # 'office__office_id': ['in', 'exact'], # 'office__name': ['in', 'exact','icontains'], # 'office__website': ['icontains'], # 'office__telephone': ['icontains'], # 'office__country': ['icontains'], # 'office__country_abbreviation': ['exact'], # 'office__state': ['icontains'], # 'office__state_abbreviation': ['exact'], # 'office__city': ['icontains'], # 'office__zone': ['icontains'], # 'office__neighborhood': ['icontains'], # 'office__complement': ['icontains'], # 'office__address': ['icontains'], # 'office__street_number': ['exact'], # 'office__postal_code': ['exact'], # 'office__total_properties': ['in', 'exact', 'range'], # 'agent__agent_id': ['exact','in','contains'], # 'agent__name': ['exact', 'icontains'], # 'agent__email': ['icontains'], # 'agent__total_properties':['in', 'exact', 'range'], # 'usage__usage_id': ['exact'], # 'usage__usage_name_en': ['exact', 'icontains'], # 'usage__usage_name_pt': ['exact', 'icontains'], # 'type__type_id': ['exact'], # 'type__type_name_en': ['exact', 'icontains'], # 'type__type_name_pt': ['exact', 'icontains'], # 'location__property_location_id': ['in', 'exact'], # 'location__display_address': ['icontains'], # 'location__country': ['in', 'exact','icontains'], # 'location__country_abbreviation': ['in','icontains'], # 'location__state': ['in', 'icontains'], # 'location__zone': ['in', 'icontains'], # 'location__neighborhood': ['in', 'icontains'], # 'location__city': ['in', 'icontains'], # 'location__complement': ['icontains'], # 'location__address': ['in', 'icontains'], # 'location__street_number': ['icontains'], # 'location__postal_code': ['icontains'], # 'location__lat': ['exact'], # 'location__lng': ['exact'], # 'medias__url': ['icontains','exists'], # 'features__feature_name_en': ['in', 'icontains'], # 'features__feature_name_pt': ['in', 'icontains'], # } # @action(detail=False, methods=['get'], url_path='aggregate-locations', url_name='aggregate_locations') # def aggregate_locations(self, request, *args, **kwargs): # """ # Custom action to aggregate properties by location (state and city). # """ # aggregated_data = aggregate_properties_by_location() # serializer = PropertiesNestedAggregateSerializer({'location_aggregate': aggregated_data}) # return Response(serializer.data, status=status.HTTP_200_OK) # from panel.views import CustomViewSet as CustomForLancamento # from panel.models import LancamentoNested # class LancamentoNestedViewSet(CustomForLancamento): # """ # A viewset for handling Property objects. # Methods Allowed: # retrieve -- Retrieve a specific Property object by ID. # list -- List all Property objects. # Methods Not Allowed: # create -- Create a new Property object. # update -- Update an existing Property object. # destroy -- Delete an existing Property object. # """ # model = LancamentoNested # queryset = model.objects.using('mongodb').all() # serializer_class = LancamentoNestedSerializer # name = "Lançamentos" # description = "Lançamentos" # filter_backends = [filters.SearchFilter, filters.OrderingFilter] # ordering_fields = '__all__' # ordering = ['id'] # pagination_class = SetPagination # allowed_methods = ['GET', 'POST', 'HEAD', 'OPTIONS', 'TRACE'] # exclude_fields = [] # search_fields = '__all__' # def get_permissions(self): # if self.action in ['list', 'retrieve']: # return [AllowAny()] # return super().get_permissions() # def get_authentication_classes(self): # if self.action in ['list', 'retrieve']: # return [] # No authentication required # return super().get_authentication_classes() # def list(self, request): # return super().list(request) # def retrieve(self, request, pk=None): # return super().retrieve(request, pk) # @swagger_auto_schema(operation=None, auto_schema=None) # def create(self, request): # response = {'message': 'Method not allowed!'} # return Response(response, status=status.HTTP_405_METHOD_NOT_ALLOWED) # @swagger_auto_schema(operation=None, auto_schema=None) # def update(self, request, pk=None): # response = {'message': 'Method not allowed!'} # return Response(response, status=status.HTTP_405_METHOD_NOT_ALLOWED) # @swagger_auto_schema(operation=None, auto_schema=None) # def partial_update(self, request, pk=None): # response = {'message': 'Method not allowed!'} # return Response(response, status=status.HTTP_405_METHOD_NOT) # @swagger_auto_schema(operation=None, auto_schema=None) # def destroy(self, request, pk=None): # response = {'message': 'Method not allowed!'} # return Response(response, status=status.HTTP_405_METHOD_NOT_ALLOWED) # from panel.models import LancamentoUnidadeNested # class LancamentoUnidadeNestedViewSet(CustomForLancamento): # """ # A viewset for handling Property objects. # Methods Allowed: # retrieve -- Retrieve a specific Property object by ID. # list -- List all Property objects. # Methods Not Allowed: # create -- Create a new Property object. # update -- Update an existing Property object. # destroy -- Delete an existing Property object. # """ # model = LancamentoUnidadeNested # queryset = model.objects.using('mongodb').all() # serializer_class = LancamentoUnidadeNestedSerializer # name = "Lançamentos" # description = "Lançamentos" # filter_backends = [filters.SearchFilter, filters.OrderingFilter] # ordering_fields = '__all__' # ordering = ['id'] # pagination_class = SetPagination # allowed_methods = ['GET', 'POST', 'HEAD', 'OPTIONS', 'TRACE'] # exclude_fields = [] # search_fields = '__all__' # def get_permissions(self): # if self.action in ['list', 'retrieve']: # return [AllowAny()] # return super().get_permissions() # def get_authentication_classes(self): # if self.action in ['list', 'retrieve']: # return [] # No authentication required # return super().get_authentication_classes() # def list(self, request): # return super().list(request) # def retrieve(self, request, pk=None): # return super().retrieve(request, pk) # @swagger_auto_schema(operation=None, auto_schema=None) # def create(self, request): # response = {'message': 'Method not allowed!'} # return Response(response, status=status.HTTP_405_METHOD_NOT_ALLOWED) # @swagger_auto_schema(operation=None, auto_schema=None) # def update(self, request, pk=None): # response = {'message': 'Method not allowed!'} # return Response(response, status=status.HTTP_405_METHOD_NOT_ALLOWED) # @swagger_auto_schema(operation=None, auto_schema=None) # def partial_update(self, request, pk=None): # response = {'message': 'Method not allowed!'} # return Response(response, status=status.HTTP_405_METHOD_NOT) # @swagger_auto_schema(operation=None, auto_schema=None) # def destroy(self, request, pk=None): # response = {'message': 'Method not allowed!'} # return Response(response, status=status.HTTP_405_METHOD_NOT_ALLOWED) class NeighborhoodFilter(django_filters.FilterSet): class Meta: model = Neighborhood fields = { 'neighborhood_id': ['in', 'exact'], 'neighborhood_name': ['icontains'], 'city_id': ['in', 'exact'], 'city_name': ['icontains'], 'state_id': ['in', 'exact'], 'country_id': ['in', 'exact'], 'country_abbreviation': ['exact'], 'country_name': ['icontains'], 'state_abbreviation': ['exact'], 'state_name': ['icontains'], } class NeighborhoodViewSet(viewsets.ReadOnlyModelViewSet): name = "Neighborhood" description = "Neighborhood Endpoint" queryset = Neighborhood.objects.using('mongodb').all() serializer_class = NeighborhoodSerializer filterset_class = NeighborhoodFilter filter_backends = [django_filters.DjangoFilterBackend, filters.SearchFilter, filters.OrderingFilter] search_fields='__all__' ordering_fields = '__all__' ordering = ['neighborhood_name'] pagination_class = SetPagination #allowed_methods = ['GET', 'HEAD', 'OPTIONS', 'TRACE'] def get_permissions(self): """ Instantiates and returns the list of permissions that this view requires. """ if self.action in ['list', 'retrieve']: permission_classes = [IsAuthenticatedOrReadOnly] else: permission_classes = [IsAdminUser] return [permission() for permission in permission_classes] class CityFilter(django_filters.FilterSet): class Meta: model = City fields = { 'city_id': ['in', 'exact'], 'city_name': ['icontains' ], 'state_id': ['in', 'exact'], 'country_id': ['in', 'exact'], 'country_abbreviation': ['exact'], 'country_name': ['icontains'], 'state_abbreviation': ['exact'], 'state_name': ['icontains'], } class CityViewSet(viewsets.ReadOnlyModelViewSet): name = "City" description = "City Endpoint" queryset = City.objects.using('mongodb').all() serializer_class = CitySerializer filterset_class = CityFilter filter_backends = [django_filters.DjangoFilterBackend, filters.SearchFilter, filters.OrderingFilter] search_fields='__all__' ordering_fields = '__all__' ordering = ['city_name'] pagination_class = SetPagination #allowed_methods = ['GET', 'HEAD', 'OPTIONS', 'TRACE'] def get_permissions(self): """ Instantiates and returns the list of permissions that this view requires. """ if self.action in ['list', 'retrieve']: permission_classes = [IsAuthenticatedOrReadOnly] else: permission_classes = [IsAdminUser] return [permission() for permission in permission_classes] class StateFilter(django_filters.FilterSet): class Meta: model = State fields = { 'state_id': ['in', 'exact'], 'state_name': ['icontains'], 'state_abbreviation': ['exact'], 'country_id': ['in', 'exact'], 'country_abbreviation': ['exact'], 'country_name': ['icontains'], } class StateViewSet(viewsets.ReadOnlyModelViewSet): name = "State" description = "State Endpoint" queryset = State.objects.using('mongodb').all() filterset_class = StateFilter serializer_class = StateSerializer filter_backends = [django_filters.DjangoFilterBackend, filters.SearchFilter, filters.OrderingFilter] search_fields='__all__' ordering_fields = '__all__' ordering = ['state_name'] pagination_class = SetPagination #allowed_methods = ['GET', 'HEAD', 'OPTIONS', 'TRACE'] def get_permissions(self): """ Instantiates and returns the list of permissions that this view requires. """ if self.action in ['list', 'retrieve']: permission_classes = [IsAuthenticatedOrReadOnly] else: permission_classes = [IsAdminUser] return [permission() for permission in permission_classes] ####################################################################### ############################ Total Views ############################## ####################################################################### from rest_framework.views import APIView from rest_framework.permissions import AllowAny class TotalCountsAPIView(APIView): """ An API view to return the total counts of various models. """ permission_classes = [AllowAny] def get(self, request, *args, **kwargs): try: start_time = time.time() total_agents = Agents.objects.count() total_offices = Offices.objects.count() total_locations = Location.objects.count() total_features = Features.objects.count() total_usages = Usages.objects.count() total_types = Types.objects.count() total_properties = PropertiesNested.objects.count() #total_medias = PropertiesMedia.objects.count() data = { 'results': [ { 'total_agents': total_agents, 'total_offices': total_offices, 'total_locations': total_locations, 'total_features': total_features, 'total_usages': total_usages, 'total_types': total_types, 'total_properties': total_properties, #'total_medias': total_medias, 'response_time': time.time() - start_time } ] } return Response(data, status=status.HTTP_200_OK) except Exception as e: return Response({'message': 'Error retrieving totals', 'Error': str(e)}, status=status.HTTP_400_BAD_REQUEST) @swagger_auto_schema(operation=None, auto_schema=None) def post(self, request, *args, **kwargs): response = {'message': 'Method not allowed!'} return Response(response, status=status.HTTP_405_METHOD_NOT_ALLOWED) @swagger_auto_schema(operation=None, auto_schema=None) def put(self, request, *args, **kwargs): response = {'message': 'Method not allowed!'} return Response(response, status=status.HTTP_405_METHOD_NOT_ALLOWED) @swagger_auto_schema(operation=None, auto_schema=None) def patch(self, request, *args, **kwargs): response = {'message': 'Method not allowed!'} return Response(response, status=status.HTTP_405_METHOD_NOT_ALLOWED) @swagger_auto_schema(operation=None, auto_schema=None) def delete(self, request, *args, **kwargs): response = {'message': 'Method not allowed!'} return Response(response, status=status.HTTP_405_METHOD_NOT_ALLOWED)