9faefd06-ed6e-4208-913d-c6157c5d73b5

Committed 30 May 2024 02:12PM UTC coverage: 71.08%. First build

Build # 9faefd06-ed6e-4208-913d-c6157c5d73b5

Build Type

push

circleci

Committed by

GeoWill

Commit Message

Allow swapping EE backends based on USE_LOCAL_PARQUET_ELECTIONS

Run Details

39 of 57 new or added lines in 5 files covered. (68.42%)

3763 of 5294 relevant lines covered (71.08%)

0.71 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

82.86

/polling_stations/apps/api/postcode.py

from data_finder.helpers import (
    PostcodeError,
    RoutingHelper,
    geocode,
    get_council,
)
from data_finder.helpers.every_election import (
    EmptyEveryElectionWrapper,
)
from data_finder.views import LogLookUpMixin, polling_station_current
from django.core.exceptions import ObjectDoesNotExist
from drf_spectacular.types import OpenApiTypes
from drf_spectacular.utils import OpenApiExample, OpenApiParameter, extend_schema
from pollingstations.models import CustomFinder
from rest_framework.permissions import IsAuthenticatedOrReadOnly
from rest_framework.response import Response
from rest_framework.viewsets import ViewSet
from uk_geo_utils.geocoders import MultipleCodesException
from uk_geo_utils.helpers import AddressSorter, Postcode

from .address import PostcodeResponseSerializer, get_bug_report_url
from .mixins import parse_qs_to_python


class PostcodeViewSet(ViewSet, LogLookUpMixin):
    permission_classes = [IsAuthenticatedOrReadOnly]
    http_method_names = ["get", "post", "head", "options"]
    lookup_field = "postcode"
    serializer_class = PostcodeResponseSerializer

    def generate_addresses(self, routing_helper):
        if routing_helper.route_type == "multiple_addresses":
            sorter = AddressSorter(routing_helper.addresses)
            return sorter.natural_sort()
        return []

    def generate_polling_station(self, routing_helper):
        if routing_helper.route_type == "single_address":
            return routing_helper.addresses[0].polling_station_with_elections()
        return None

    def generate_custom_finder(self, geocoder, postcode):
        try:
            finder = CustomFinder.objects.get_custom_finder(geocoder, postcode)
        except MultipleCodesException:
            finder = None
        if finder and finder.base_url:
            if finder.can_pass_postcode:
                return finder.base_url + finder.encoded_postcode
            return finder.base_url
        return None

    def generate_advance_voting_station(self, routing_helper):
        if routing_helper.route_type == "single_address":
            return routing_helper.addresses[0].uprntocouncil.advance_voting_station
        return None

    def get_ee_wrapper(self, postcode, rh, query_params):
        if rh.route_type == "multiple_addresses":
            return EmptyEveryElectionWrapper()
        if rh.elections_response:
            return rh.elections_backend.ee_wrapper(rh.elections_response)
        kwargs = {}
        query_params = parse_qs_to_python(query_params)
        if include_current := query_params.get("include_current", False):
            kwargs["include_current"] = any(include_current)
        return rh.elections_backend.ee_wrapper(postcode, **kwargs)

    @extend_schema(
        parameters=[
            OpenApiParameter(
                name="postcode",
                type=OpenApiTypes.STR,
                location=OpenApiParameter.PATH,
                description="A Valid UK postcode",
                examples=[
                    OpenApiExample(
                        "Example 1", summary="Buckingham Palace", value="SW1A 1AA"
                    ),
                ],
            )
        ]
    )
    def retrieve(self, request, postcode=None, format=None, geocoder=geocode, log=True):
        postcode = Postcode(postcode)
        ret = {}

        rh = RoutingHelper(postcode)

        # attempt to attach point and gss_codes
        try:
            loc = geocoder(postcode)
            location = loc.centroid
        except PostcodeError as e:
            return Response({"detail": e.args[0]}, status=400)

        ret["postcode_location"] = location

        # council object
        if rh.councils or not loc:
            # We can't assign this postcode to exactly one council
            council = None
        else:
            try:
                council = get_council(loc)
            except ObjectDoesNotExist:
                return Response({"detail": "Internal server error"}, 500)
        ret["council"] = council

        ret["addresses"] = self.generate_addresses(rh)

        ret["polling_station_known"] = False
        ret["polling_station"] = None

        ee = self.get_ee_wrapper(postcode, rh, request.query_params)
        has_election = ee.has_election()
        if has_election:
            # get polling station if there is an election in this area
            ret["polling_station_known"] = False
            ret["polling_station"] = self.generate_polling_station(rh)
            if ret["polling_station"]:
                if polling_station_current(ret["polling_station"]):
                    ret["polling_station_known"] = True
                else:
                    ret["polling_station"] = None
            if ret["polling_station"] and not ret["council"]:
                ret["council"] = ret["polling_station"].council

        # get custom finder (if no polling station)
        ret["custom_finder"] = None
        if not ret["polling_station_known"] and loc:
            ret["custom_finder"] = self.generate_custom_finder(loc, postcode)

        # get advance voting station
        ret["advance_voting_station"] = self.generate_advance_voting_station(rh)

        ret["metadata"] = ee.get_metadata()

        if request.query_params.get("all_future_ballots", None):
            ret["ballots"] = ee.get_all_ballots()
        else:
            ret["ballots"] = ee.get_ballots_for_next_date()

        # create log entry
        log_data = {}
        log_data["we_know_where_you_should_vote"] = ret["polling_station_known"]
        log_data["location"] = location
        log_data["council"] = council
        log_data["brand"] = "api"
        log_data["language"] = ""
        log_data["api_user"] = request.user
        log_data["has_election"] = has_election
        if log and not ret["addresses"]:
            self.log_postcode(postcode, log_data, "api")
            # don't log 'address select' hits

        ret["report_problem_url"] = get_bug_report_url(
            request, ret["polling_station_known"]
        )

        serializer = PostcodeResponseSerializer(
            ret, read_only=True, context={"request": request}
        )
        return Response(serializer.data)

1	from data_finder.helpers import (	1✔
2	PostcodeError,
3	RoutingHelper,
4	geocode,
5	get_council,
6	)
7	from data_finder.helpers.every_election import (	1✔
8	EmptyEveryElectionWrapper,
9	)
10	from data_finder.views import LogLookUpMixin, polling_station_current	1✔
11	from django.core.exceptions import ObjectDoesNotExist	1✔
12	from drf_spectacular.types import OpenApiTypes	1✔
13	from drf_spectacular.utils import OpenApiExample, OpenApiParameter, extend_schema	1✔
14	from pollingstations.models import CustomFinder	1✔
15	from rest_framework.permissions import IsAuthenticatedOrReadOnly	1✔
16	from rest_framework.response import Response	1✔
17	from rest_framework.viewsets import ViewSet	1✔
18	from uk_geo_utils.geocoders import MultipleCodesException	1✔
19	from uk_geo_utils.helpers import AddressSorter, Postcode	1✔
20
21	from .address import PostcodeResponseSerializer, get_bug_report_url	1✔
22	from .mixins import parse_qs_to_python	1✔
23
24
25	class PostcodeViewSet(ViewSet, LogLookUpMixin):	1✔
26	permission_classes = [IsAuthenticatedOrReadOnly]	1✔
27	http_method_names = ["get", "post", "head", "options"]	1✔
28	lookup_field = "postcode"	1✔
29	serializer_class = PostcodeResponseSerializer	1✔
30
31	def generate_addresses(self, routing_helper):	1✔
32	if routing_helper.route_type == "multiple_addresses":	1✔
33	sorter = AddressSorter(routing_helper.addresses)	1✔
34	return sorter.natural_sort()	1✔
35	return []	1✔
36
37	def generate_polling_station(self, routing_helper):	1✔
38	if routing_helper.route_type == "single_address":	1✔
39	return routing_helper.addresses[0].polling_station_with_elections()	1✔
40	return None	1✔
41
42	def generate_custom_finder(self, geocoder, postcode):	1✔
43	try:	1✔
44	finder = CustomFinder.objects.get_custom_finder(geocoder, postcode)	1✔
45	except MultipleCodesException:	×
46	finder = None	×
47	if finder and finder.base_url:	1✔
48	if finder.can_pass_postcode:	×
49	return finder.base_url + finder.encoded_postcode	×
50	return finder.base_url	×
51	return None	1✔
52
53	def generate_advance_voting_station(self, routing_helper):	1✔
54	if routing_helper.route_type == "single_address":	1✔
55	return routing_helper.addresses[0].uprntocouncil.advance_voting_station	1✔
56	return None	1✔
57
58	def get_ee_wrapper(self, postcode, rh, query_params):	1✔
59	if rh.route_type == "multiple_addresses":	×
60	return EmptyEveryElectionWrapper()	×
61	if rh.elections_response:	×
NEW 62	return rh.elections_backend.ee_wrapper(rh.elections_response)	×
63	kwargs = {}	×
64	query_params = parse_qs_to_python(query_params)	×
65	if include_current := query_params.get("include_current", False):	×
66	kwargs["include_current"] = any(include_current)	×
NEW 67	return rh.elections_backend.ee_wrapper(postcode, **kwargs)	×
68
69	@extend_schema(	1✔
70	parameters=[
71	OpenApiParameter(
72	name="postcode",
73	type=OpenApiTypes.STR,
74	location=OpenApiParameter.PATH,
75	description="A Valid UK postcode",
76	examples=[
77	OpenApiExample(
78	"Example 1", summary="Buckingham Palace", value="SW1A 1AA"
79	),
80	],
81	)
82	]
83	)
84	def retrieve(self, request, postcode=None, format=None, geocoder=geocode, log=True):	1✔
85	postcode = Postcode(postcode)	1✔
86	ret = {}	1✔
87
88	rh = RoutingHelper(postcode)	1✔
89
90	# attempt to attach point and gss_codes
91	try:	1✔
92	loc = geocoder(postcode)	1✔
93	location = loc.centroid	1✔
94	except PostcodeError as e:	1✔
95	return Response({"detail": e.args[0]}, status=400)	1✔
96
97	ret["postcode_location"] = location	1✔
98
99	# council object
100	if rh.councils or not loc:	1✔
101	# We can't assign this postcode to exactly one council
102	council = None	×
103	else:
104	try:	1✔
105	council = get_council(loc)	1✔
106	except ObjectDoesNotExist:	1✔
107	return Response({"detail": "Internal server error"}, 500)	1✔
108	ret["council"] = council	1✔
109
110	ret["addresses"] = self.generate_addresses(rh)	1✔
111
112	ret["polling_station_known"] = False	1✔
113	ret["polling_station"] = None	1✔
114
115	ee = self.get_ee_wrapper(postcode, rh, request.query_params)	1✔
116	has_election = ee.has_election()	1✔
117	if has_election:	1✔
118	# get polling station if there is an election in this area
119	ret["polling_station_known"] = False	1✔
120	ret["polling_station"] = self.generate_polling_station(rh)	1✔
121	if ret["polling_station"]:	1✔
122	if polling_station_current(ret["polling_station"]):	1✔
123	ret["polling_station_known"] = True	1✔
124	else:
125	ret["polling_station"] = None	1✔
126	if ret["polling_station"] and not ret["council"]:	1✔
127	ret["council"] = ret["polling_station"].council	×
128
129	# get custom finder (if no polling station)
130	ret["custom_finder"] = None	1✔
131	if not ret["polling_station_known"] and loc:	1✔
132	ret["custom_finder"] = self.generate_custom_finder(loc, postcode)	1✔
133
134	# get advance voting station
135	ret["advance_voting_station"] = self.generate_advance_voting_station(rh)	1✔
136
137	ret["metadata"] = ee.get_metadata()	1✔
138
139	if request.query_params.get("all_future_ballots", None):	1✔
140	ret["ballots"] = ee.get_all_ballots()	×
141	else:
142	ret["ballots"] = ee.get_ballots_for_next_date()	1✔
143
144	# create log entry
145	log_data = {}	1✔
146	log_data["we_know_where_you_should_vote"] = ret["polling_station_known"]	1✔
147	log_data["location"] = location	1✔
148	log_data["council"] = council	1✔
149	log_data["brand"] = "api"	1✔
150	log_data["language"] = ""	1✔
151	log_data["api_user"] = request.user	1✔
152	log_data["has_election"] = has_election	1✔
153	if log and not ret["addresses"]:	1✔
154	self.log_postcode(postcode, log_data, "api")	×
155	# don't log 'address select' hits
156
157	ret["report_problem_url"] = get_bug_report_url(	1✔
158	request, ret["polling_station_known"]
159	)
160
161	serializer = PostcodeResponseSerializer(	1✔
162	ret, read_only=True, context={"request": request}
163	)
164	return Response(serializer.data)	1✔

DemocracyClub / UK-Polling-Stations / 9faefd06-ed6e-4208-913d-c6157c5d73b5

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous