1bd9fda2-91d4-4587-a1b1-22dace2973b6

Committed 27 Mar 2025 11:17AM UTC coverage: 73.958% (+0.6%) from 73.378%

Build # 1bd9fda2-91d4-4587-a1b1-22dace2973b6

Build Type

push

circleci

Committed by

chris48s

Commit Message

one retry on connect error

Run Details

4098 of 5541 relevant lines covered (73.96%)

0.74 hits per line

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

94.79

/polling_stations/apps/api/postcode.py

from data_finder.helpers import (
    PostcodeError,
    RoutingHelper,
    geocode,
    get_council,
)
from data_finder.helpers.every_election import EEFetcher, EEWrapper, EmptyEEWrapper
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 rest_framework.exceptions import APIException
from rest_framework.permissions import IsAuthenticatedOrReadOnly
from rest_framework.response import Response
from rest_framework.viewsets import ViewSet
from uk_geo_utils.helpers import AddressSorter, Postcode

from .address import PostcodeResponseSerializer, get_bug_report_url
from .councils import tmp_fix_parl_24_scotland_details
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_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):
        query_params = parse_qs_to_python(query_params)
        include_current = any(query_params.get("include_current", []))

        if rh.route_type == "multiple_addresses":
            return EmptyEEWrapper()

        if rh.elections_response:
            if not rh.elections_response["request_success"]:
                raise APIException("failed to get list of ballots")
            return EEWrapper(
                rh.elections_response["ballots"],
                request_success=rh.elections_response["request_success"],
                include_current=include_current,
            )

        ee_response = EEFetcher(postcode=postcode).fetch()
        if not ee_response["request_success"]:
            raise APIException("failed to get list of ballots")
        return EEWrapper(**ee_response, include_current=include_current)

    @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:
                raise APIException("council does not exist")
        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:
            ret["council"] = tmp_fix_parl_24_scotland_details(ret["council"], ee)

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

DemocracyClub / UK-Polling-Stations / 1bd9fda2-91d4-4587-a1b1-22dace2973b6

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