diff --git a/src/meshapi/permissions.py b/src/meshapi/permissions.py index a3aecb86..070c24e9 100644 --- a/src/meshapi/permissions.py +++ b/src/meshapi/permissions.py @@ -42,7 +42,10 @@ class HasPanoramaUpdatePermission(HasDjangoPermission): # Janky class LegacyMeshQueryPassword(permissions.BasePermission): def has_permission(self, request, view): - if "password" in request.query_params and request.query_params["password"] == os.environ.get("QUERY_PSK"): + if ( + request.headers["Authorization"] + and request.headers["Authorization"] == f"Bearer {os.environ.get('QUERY_PSK')}" + ): return True raise PermissionDenied("Authentication Failed.") diff --git a/src/meshapi/serializers/query_api.py b/src/meshapi/serializers/query_api.py index 806ecc63..5d42f27a 100644 --- a/src/meshapi/serializers/query_api.py +++ b/src/meshapi/serializers/query_api.py @@ -35,7 +35,7 @@ class Meta: name = serializers.CharField(source="member.name") - network_number = serializers.IntegerField(source="node.network_number") + network_number = serializers.IntegerField(source="node.network_number", allow_null=True) primary_email_address = serializers.CharField(source="member.primary_email_address") stripe_email_address = serializers.CharField(source="member.stripe_email_address") diff --git a/src/meshapi/tests/test_query_form.py b/src/meshapi/tests/test_query_form.py index 8c4dab12..a45f4b04 100644 --- a/src/meshapi/tests/test_query_form.py +++ b/src/meshapi/tests/test_query_form.py @@ -31,8 +31,9 @@ def setUp(self): def query(self, route, field, data): code = 200 password = os.environ.get("QUERY_PSK") - route = f"/api/v1/query/{route}/?{field}={data}&password={password}" - response = self.c.get(route) + route = f"/api/v1/query/{route}/?{field}={data}" + headers = {"Authorization": f"Bearer {password}"} + response = self.c.get(route, headers=headers) self.assertEqual( code, response.status_code, @@ -40,7 +41,7 @@ def query(self, route, field, data): ) resp_json = json.loads(response.content.decode("utf-8")) - self.assertEqual(len(resp_json), 1) + self.assertEqual(len(resp_json["results"]), 1) def test_query_address(self): self.query("buildings", "street_address", self.install.building.street_address) diff --git a/src/meshapi/views/lookups.py b/src/meshapi/views/lookups.py index 7f8d403f..becf2470 100644 --- a/src/meshapi/views/lookups.py +++ b/src/meshapi/views/lookups.py @@ -29,8 +29,8 @@ def get(self, request, *args, **kwargs): # If they gave us filters that aren't actually available, bail and return a bad status if invalid_filters: return Response({"detail": f"Invalid filters provided: {list(invalid_filters)}"}, 400) - else: - return super().get(request, *args, **kwargs) + + return super().get(request, *args, **kwargs) class MemberFilter(filters.FilterSet): diff --git a/src/meshapi/views/query_api.py b/src/meshapi/views/query_api.py index 0a66a0fe..19ae2edf 100644 --- a/src/meshapi/views/query_api.py +++ b/src/meshapi/views/query_api.py @@ -1,6 +1,8 @@ +import os from typing import Any, Dict, Optional from drf_spectacular.utils import extend_schema, extend_schema_view +from meshapi.views.lookups import FilterRequiredListAPIView, InstallFilter from rest_framework import generics from rest_framework.views import models @@ -8,6 +10,17 @@ from meshapi.docs import map_query_filters_to_param_annotations, query_form_password_param from meshapi.models import Building, Install, Member from meshapi.serializers.query_api import QueryFormSerializer +from django.db.models import Q +from django_filters import rest_framework as filters +from drf_spectacular.types import OpenApiTypes +from drf_spectacular.utils import OpenApiParameter, extend_schema, extend_schema_view +from rest_framework import generics +from rest_framework.response import Response + +from rest_framework.decorators import api_view, permission_classes + +from rest_framework import permissions +from meshapi.permissions import LegacyMeshQueryPassword """ Re-implements https://docs.nycmesh.net/installs/query/ @@ -18,128 +31,77 @@ """ -def filter_model_on_query_params( - query_params: Dict[str, Any], - model: type[models.Model], - permitted_filters: Dict[str, Optional[str]], - subsitutions: Dict[str, str] = None, -): - query_dict = {} - for k, v in query_params.items(): - if v: - if subsitutions and k in subsitutions: - query_dict[subsitutions[k]] = v - else: - query_dict[k] = v - - filter_args = {} - for k, v in query_dict.items(): - if k in permitted_filters.keys(): - if permitted_filters[k]: - filter_args[f"{k}__{permitted_filters[k]}"] = v - else: - filter_args[f"{k}"] = v - - return model.objects.filter(**filter_args) - - -BUILDING_FILTERS = { - "street_address": "icontains", - "zip_code": "iexact", - "city": "icontains", - "state": "icontains", - "bin": "iexact", -} - - -@extend_schema_view( - get=extend_schema( - tags=["Legacy Query Form"], - parameters=[query_form_password_param] + map_query_filters_to_param_annotations(BUILDING_FILTERS), - summary="Query & filter based on Building attributes. " - "Results are returned as flattened spreadsheet row style output", - auth=[], - ), -) -class QueryBuilding(generics.ListAPIView): - serializer_class = QueryFormSerializer - pagination_class = None - permission_classes = [permissions.LegacyMeshQueryPassword] - - def get_queryset(self): - buildings = filter_model_on_query_params( - self.request.query_params, - Building, - BUILDING_FILTERS, - ) - - output_qs = [] - for building in buildings: - output_qs.extend(building.installs.all()) - - return output_qs +class QueryMemberFilter(filters.FilterSet): + name = filters.CharFilter(field_name="member.name", lookup_expr="icontains") + email_address = filters.CharFilter(method="filter_on_all_emails") + phone_number = filters.CharFilter(field_name="member.phone_number", lookup_expr="icontains") + def filter_on_all_emails(self, queryset, name, value): + return queryset.filter( + Q(member__primary_email_address__icontains=value) + | Q(member__stripe_email_address__icontains=value) + | Q(member__additional_email_addresses__icontains=value) + ) -MEMBER_FILTERS = { - "email_address": None, -} + class Meta: + model = Install + fields = [] -@extend_schema_view( - get=extend_schema( - tags=["Legacy Query Form"], - parameters=[query_form_password_param] + map_query_filters_to_param_annotations(MEMBER_FILTERS), - summary="Query & filter based on Member attributes. " - "Results are returned as flattened spreadsheet row style output", - auth=[], - ), -) -class QueryMember(generics.ListAPIView): +class QueryMember(FilterRequiredListAPIView): + queryset = ( + Install.objects.all() + .prefetch_related("building") + .prefetch_related("node") + .prefetch_related("member") + .order_by("install_number") + ) serializer_class = QueryFormSerializer - pagination_class = None - permission_classes = [permissions.LegacyMeshQueryPassword] - - def get_queryset(self): - members = filter_model_on_query_params( - self.request.query_params, - Member, - MEMBER_FILTERS, - { - "email_address": "primary_email_address", - }, - ) + filterset_class = QueryMemberFilter + permission_classes = [LegacyMeshQueryPassword] - output_qs = [] - for member in members: - output_qs.extend(member.installs.all()) - return output_qs +class QueryInstallFilter(filters.FilterSet): + network_number = filters.NumberFilter(field_name="node__network_number", lookup_expr="exact") + class Meta: + model = Install + fields = ["install_number", "member", "building", "status"] -INSTALL_FILTERS = { - "node__network_number": None, - "install_number": None, -} - -@extend_schema_view( - get=extend_schema( - tags=["Legacy Query Form"], - parameters=[query_form_password_param] + map_query_filters_to_param_annotations(INSTALL_FILTERS), - summary="Query & filter based on Install attributes. " - "Results are returned as flattened spreadsheet row style output", - auth=[], - ), -) -class QueryInstall(generics.ListAPIView): +class QueryInstall(FilterRequiredListAPIView): + queryset = ( + Install.objects.all() + .prefetch_related("building") + .prefetch_related("node") + .prefetch_related("member") + .order_by("install_number") + ) serializer_class = QueryFormSerializer - pagination_class = None - permission_classes = [permissions.LegacyMeshQueryPassword] - - def get_queryset(self): - return filter_model_on_query_params( - self.request.query_params, - Install, - INSTALL_FILTERS, - {"network_number": "node__network_number"}, - ) + filterset_class = QueryInstallFilter + permission_classes = [LegacyMeshQueryPassword] + + +class QueryBuildingFilter(filters.FilterSet): + street_address = filters.CharFilter(field_name="building__street_address", lookup_expr="icontains") + city = filters.CharFilter(field_name="building__city", lookup_expr="iexact") + state = filters.CharFilter(field_name="building__state", lookup_expr="iexact") + zip_code = filters.CharFilter(field_name="building__zip_code", lookup_expr="iexact") + bin = filters.CharFilter(field_name="building__bin", lookup_expr="iexact") + + class Meta: + model = Install + fields = ["bin", "zip_code"] + + +class QueryBuilding(FilterRequiredListAPIView): + queryset = ( + Install.objects.all() + .prefetch_related("building") + .prefetch_related("node") + .prefetch_related("member") + .order_by("install_number") + ) + serializer_class = QueryFormSerializer + filterset_class = QueryBuildingFilter + permission_classes = [LegacyMeshQueryPassword]