20583796061

Committed 29 Dec 2025 10:11PM UTC coverage: 96.449% (-2.3%) from 98.783%

Build # 20583796061

Build Type

Pull #278

github

Committed by

web-flow

Commit Message

Merge dacfe649f into 783f855e7

Pull Request Pull Request #278: chore: better errors handling

Run Details

559 of 590 branches covered (94.75%)

Branch coverage included in aggregate %.

139 of 167 new or added lines in 4 files covered. (83.23%)

1 existing line in 1 file now uncovered.

1261 of 1297 relevant lines covered (97.22%)

0.97 hits per line

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

94.32

/goodmap/core_api.py

import importlib.metadata
import logging
import uuid

import numpy
import pysupercluster
from flask import Blueprint, jsonify, make_response, request
from flask_babel import gettext
from flask_restx import Api, Resource, fields
from platzky.config import LanguagesMapping
from werkzeug.exceptions import BadRequest

from goodmap.clustering import (
    map_clustering_data_to_proper_lazy_loading_object,
    match_clusters_uuids,
)
from goodmap.exceptions import (
    LocationAlreadyExistsError,
    LocationNotFoundError,
    LocationValidationError,
    ReportNotFoundError,
)
from goodmap.formatter import prepare_pin

# SuperCluster configuration constants
MIN_ZOOM = 0
MAX_ZOOM = 16
CLUSTER_RADIUS = 200
CLUSTER_EXTENT = 512

logger = logging.getLogger(__name__)


def make_tuple_translation(keys_to_translate):
    return [(x, gettext(x)) for x in keys_to_translate]


def get_or_none(data, *keys):
    for key in keys:
        if isinstance(data, dict):
            data = data.get(key)
        else:
            return None
    return data


def get_locations_from_request(database, request_args, as_basic_info=False):
    """
    Shared helper to fetch locations from database based on request arguments.

    Args:
        database: Database instance
        request_args: Request arguments (flask.request.args)
        as_basic_info: If True, returns list of basic_info dicts, otherwise returns Location objects

    Returns:
        List of locations (either as objects or basic_info dicts)
    """
    query_params = request_args.to_dict(flat=False)
    all_locations = database.get_locations(query_params)

    if as_basic_info:
        return [x.basic_info() for x in all_locations]

    return all_locations


def core_pages(
    database,
    languages: LanguagesMapping,
    notifier_function,
    csrf_generator,
    location_model,
    feature_flags={},
) -> Blueprint:
    core_api_blueprint = Blueprint("api", __name__, url_prefix="/api")
    core_api = Api(core_api_blueprint, doc="/doc", version="0.1")

    location_report_model = core_api.model(
        "LocationReport",
        {
            "id": fields.String(required=True, description="Location ID"),
            "description": fields.String(required=True, description="Description of the problem"),
        },
    )

    # TODO get this from Location pydantic model
    suggested_location_model = core_api.model(
        "LocationSuggestion",
        {
            "name": fields.String(required=False, description="Organization name"),
            "position": fields.String(required=True, description="Location of the suggestion"),
            "photo": fields.String(required=False, description="Photo of the location"),
        },
    )

    @core_api.route("/suggest-new-point")
    class NewLocation(Resource):
        @core_api.expect(suggested_location_model)
        def post(self):
            """Suggest new location"""
            try:
                suggested_location = request.get_json()
                suggested_location.update({"uuid": str(uuid.uuid4())})
                location = location_model.model_validate(suggested_location)
                database.add_suggestion(location.model_dump())
                message = (
                    f"A new location has been suggested under uuid: '{location.uuid}' "
                    f"at position: {location.position}"
                )
                notifier_function(message)
            except BadRequest:
                logger.warning("Invalid JSON in suggest endpoint")
                return make_response(jsonify({"message": "Invalid request data"}), 400)
            except LocationValidationError as e:
                logger.warning(
                    "Location validation failed in suggest endpoint",
                    extra={"errors": e.validation_errors},
                )
                return make_response(jsonify({"message": "Invalid location data"}), 400)
            except Exception:
                logger.error("Error in suggest location endpoint", exc_info=True)
                return make_response(
                    jsonify({"message": "An error occurred while processing your suggestion"}), 500
                )
            return make_response(jsonify({"message": "Location suggested"}), 200)

    @core_api.route("/report-location")
    class ReportLocation(Resource):
        @core_api.expect(location_report_model)
        def post(self):
            """Report location"""
            try:
                location_report = request.get_json()
                report = {
                    "uuid": str(uuid.uuid4()),
                    "location_id": location_report["id"],
                    "description": location_report["description"],
                    "status": "pending",
                    "priority": "medium",
                }
                database.add_report(report)
                message = (
                    f"A location has been reported: '{location_report['id']}' "
                    f"with problem: {location_report['description']}"
                )
                notifier_function(message)
            except BadRequest:
                logger.warning("Invalid JSON in report location endpoint")
                return make_response(jsonify({"message": "Invalid request data"}), 400)
            except KeyError as e:
                logger.warning(
                    "Missing required field in report location", extra={"missing_field": str(e)}
                )
                error_message = gettext("Error reporting location")
                return make_response(jsonify({"message": error_message}), 400)
            except Exception:
                logger.error("Error in report location endpoint", exc_info=True)
                error_message = gettext("Error sending notification")
                return make_response(jsonify({"message": error_message}), 500)
            return make_response(jsonify({"message": gettext("Location reported")}), 200)

    @core_api.route("/locations")
    class GetLocations(Resource):
        def get(self):
            """
            Shows list of locations with uuid and position
            """
            locations = get_locations_from_request(database, request.args, as_basic_info=True)
            return jsonify(locations)

    @core_api.route("/locations-clustered")
    class GetLocationsClustered(Resource):
        def get(self):
            """
            Shows list of locations with uuid, position and clusters
            """
            try:
                query_params = request.args.to_dict(flat=False)
                zoom = int(query_params.get("zoom", [7])[0])

                # Validate zoom level (aligned with SuperCluster min_zoom/max_zoom)
                if not MIN_ZOOM <= zoom <= MAX_ZOOM:
                    return make_response(
                        jsonify({"message": f"Zoom must be between {MIN_ZOOM} and {MAX_ZOOM}"}),
                        400,
                    )

                points = get_locations_from_request(database, request.args, as_basic_info=True)
                if not points:
                    return jsonify([])

                points_numpy = numpy.array(
                    [(point["position"][0], point["position"][1]) for point in points]
                )

                index = pysupercluster.SuperCluster(
                    points_numpy,
                    min_zoom=MIN_ZOOM,
                    max_zoom=MAX_ZOOM,
                    radius=CLUSTER_RADIUS,
                    extent=CLUSTER_EXTENT,
                )

                clusters = index.getClusters(
                    top_left=(-180.0, 90.0),
                    bottom_right=(180.0, -90.0),
                    zoom=zoom,
                )
                clusters = match_clusters_uuids(points, clusters)

                return jsonify(map_clustering_data_to_proper_lazy_loading_object(clusters))
            except ValueError as e:
                logger.warning("Invalid parameter in clustering request: %s", e)
                return make_response(jsonify({"message": "Invalid parameters provided"}), 400)
            except Exception as e:
                logger.error("Clustering operation failed: %s", e, exc_info=True)
                return make_response(
                    jsonify({"message": "An error occurred during clustering"}), 500
                )

    @core_api.route("/location/<location_id>")
    class GetLocation(Resource):
        def get(self, location_id):
            """
            Shows a single location with all data
            """
            location = database.get_location(location_id)
            visible_data = database.get_visible_data()
            meta_data = database.get_meta_data()

            formatted_data = prepare_pin(location.model_dump(), visible_data, meta_data)
            return jsonify(formatted_data)

    @core_api.route("/version")
    class Version(Resource):
        def get(self):
            """Shows backend version"""
            version_info = {"backend": importlib.metadata.version("goodmap")}
            return jsonify(version_info)

    @core_api.route("/categories")
    class Categories(Resource):
        def get(self):
            """Shows all available categories"""
            raw_categories = database.get_categories()
            categories = make_tuple_translation(raw_categories)

            if not feature_flags.get("CATEGORIES_HELP", False):
                return jsonify(categories)
            else:
                category_data = database.get_category_data()
                categories_help = category_data["categories_help"]
                proper_categories_help = []
                if categories_help is not None:
                    for option in categories_help:
                        proper_categories_help.append(
                            {option: gettext(f"categories_help_{option}")}
                        )

            return jsonify({"categories": categories, "categories_help": proper_categories_help})

    @core_api.route("/languages")
    class Languages(Resource):
        def get(self):
            """Shows all available languages"""
            return jsonify(languages)

    @core_api.route("/category/<category_type>")
    class CategoryTypes(Resource):
        def get(self, category_type):
            """Shows all available types in category"""
            category_data = database.get_category_data(category_type)
            local_data = make_tuple_translation(category_data["categories"][category_type])

            categories_options_help = get_or_none(
                category_data, "categories_options_help", category_type
            )
            proper_categories_options_help = []
            if categories_options_help is not None:
                for option in categories_options_help:
                    proper_categories_options_help.append(
                        {option: gettext(f"categories_options_help_{option}")}
                    )
            if not feature_flags.get("CATEGORIES_HELP", False):
                return jsonify(local_data)
            else:
                return jsonify(
                    {
                        "categories_options": local_data,
                        "categories_options_help": proper_categories_options_help,
                    }
                )

    @core_api.route("/generate-csrf-token")
    class CsrfToken(Resource):
        def get(self):
            csrf_token = csrf_generator()
            return {"csrf_token": csrf_token}

    @core_api.route("/admin/locations")
    class AdminManageLocations(Resource):
        def get(self):
            """
            Shows full list of locations, with optional server-side pagination, sorting,
            and filtering.
            """
            query_params = request.args.to_dict(flat=False)
            if "sort_by" not in query_params:
                query_params["sort_by"] = ["name"]
            result = database.get_locations_paginated(query_params)
            return jsonify(result)

        def post(self):
            """
            Creates a new location
            """
            location_data = request.get_json()
            try:
                location_data.update({"uuid": str(uuid.uuid4())})
                location = location_model.model_validate(location_data)
                database.add_location(location.model_dump())
            except LocationValidationError as e:
                logger.warning(
                    "Location validation failed",
                    extra={"uuid": e.uuid, "errors": e.validation_errors},
                )
                return make_response(jsonify({"message": "Invalid location data"}), 400)
            except LocationAlreadyExistsError as e:
                logger.warning("Attempted to create duplicate location", extra={"uuid": e.uuid})
                return make_response(jsonify({"message": "Location already exists"}), 409)
            except Exception:
                logger.error("Error creating location", exc_info=True)
                return make_response(jsonify({"message": "An internal error occurred"}), 500)
            return jsonify(location.model_dump())

    @core_api.route("/admin/locations/<location_id>")
    class AdminManageLocation(Resource):
        def put(self, location_id):
            """
            Updates a single location
            """
            location_data = request.get_json()
            try:
                location_data.update({"uuid": location_id})
                location = location_model.model_validate(location_data)
                database.update_location(location_id, location.model_dump())
            except LocationValidationError as e:
                logger.warning(
                    "Location validation failed",
                    extra={"uuid": e.uuid, "errors": e.validation_errors},
                )
                return make_response(jsonify({"message": "Invalid location data"}), 400)
            except LocationNotFoundError as e:
                logger.info("Location not found for update", extra={"uuid": e.uuid})
                return make_response(jsonify({"message": "Location not found"}), 404)
            except Exception:
                logger.error("Error updating location", exc_info=True)
                return make_response(jsonify({"message": "An internal error occurred"}), 500)
            return jsonify(location.model_dump())

        def delete(self, location_id):
            """
            Deletes a single location
            """
            try:
                database.delete_location(location_id)
            except LocationNotFoundError as e:
                logger.info("Location not found for deletion", extra={"uuid": e.uuid})
                return make_response(jsonify({"message": "Location not found"}), 404)
            except Exception:
                logger.error("Error deleting location", exc_info=True)
                return make_response(jsonify({"message": "An internal error occurred"}), 500)
            return "", 204

    @core_api.route("/admin/suggestions")
    class AdminManageSuggestions(Resource):
        def get(self):
            """
            List location suggestions, with optional server-side pagination, sorting,
            and filtering by status.
            """
            query_params = request.args.to_dict(flat=False)
            result = database.get_suggestions_paginated(query_params)
            return jsonify(result)

    @core_api.route("/admin/suggestions/<suggestion_id>")
    class AdminManageSuggestion(Resource):
        def put(self, suggestion_id):
            """
            Accept or reject a location suggestion
            """
            try:
                data = request.get_json()
                status = data.get("status")
                if status not in ("accepted", "rejected"):
                    return make_response(jsonify({"message": "Invalid status"}), 400)
                suggestion = database.get_suggestion(suggestion_id)
                if not suggestion:
                    return make_response(jsonify({"message": "Suggestion not found"}), 404)
                if suggestion.get("status") != "pending":
                    return make_response(jsonify({"message": "Suggestion already processed"}), 400)
                if status == "accepted":
                    suggestion_data = {k: v for k, v in suggestion.items() if k != "status"}
                    database.add_location(suggestion_data)
                database.update_suggestion(suggestion_id, status)
            except LocationValidationError as e:
                logger.warning(
                    "Location validation failed in suggestion",
                    extra={"uuid": e.uuid, "errors": e.validation_errors},
                )
                return make_response(jsonify({"message": "Invalid location data"}), 400)
            except LocationAlreadyExistsError as e:
                logger.warning(
                    "Attempted to create duplicate location from suggestion", extra={"uuid": e.uuid}
                )
                return make_response(jsonify({"message": "Location already exists"}), 409)
            except Exception:
                logger.error("Error processing suggestion", exc_info=True)
                return make_response(jsonify({"message": "An internal error occurred"}), 500)
            return jsonify(database.get_suggestion(suggestion_id))

    @core_api.route("/admin/reports")
    class AdminManageReports(Resource):
        def get(self):
            """
            List location reports, with optional server-side pagination, sorting,
            and filtering by status/priority.
            """
            query_params = request.args.to_dict(flat=False)
            result = database.get_reports_paginated(query_params)
            return jsonify(result)

    @core_api.route("/admin/reports/<report_id>")
    class AdminManageReport(Resource):
        def put(self, report_id):
            """
            Update a report's status and/or priority
            """
            try:
                data = request.get_json()
                status = data.get("status")
                priority = data.get("priority")
                valid_status = ("resolved", "rejected")
                valid_priority = ("critical", "high", "medium", "low")
                if status and status not in valid_status:
                    return make_response(jsonify({"message": "Invalid status"}), 400)
                if priority and priority not in valid_priority:
                    return make_response(jsonify({"message": "Invalid priority"}), 400)
                report = database.get_report(report_id)
                if not report:
                    return make_response(jsonify({"message": "Report not found"}), 404)
                database.update_report(report_id, status=status, priority=priority)
            except BadRequest:
                logger.warning("Invalid JSON in report update endpoint")
                return make_response(jsonify({"message": "Invalid request data"}), 400)
            except ReportNotFoundError as e:
                logger.info("Report not found for update", extra={"uuid": e.uuid})
                return make_response(jsonify({"message": "Report not found"}), 404)
            except Exception:
                logger.error("Error updating report", exc_info=True)
                return make_response(jsonify({"message": "An internal error occurred"}), 500)
            return jsonify(database.get_report(report_id))

    return core_api_blueprint

1	import importlib.metadata	1✔
2	import logging	1✔
3	import uuid	1✔
4
5	import numpy	1✔
6	import pysupercluster	1✔
7	from flask import Blueprint, jsonify, make_response, request	1✔
8	from flask_babel import gettext	1✔
9	from flask_restx import Api, Resource, fields	1✔
10	from platzky.config import LanguagesMapping	1✔
11	from werkzeug.exceptions import BadRequest	1✔
12
13	from goodmap.clustering import (	1✔
14	map_clustering_data_to_proper_lazy_loading_object,
15	match_clusters_uuids,
16	)
17	from goodmap.exceptions import (	1✔
18	LocationAlreadyExistsError,
19	LocationNotFoundError,
20	LocationValidationError,
21	ReportNotFoundError,
22	)
23	from goodmap.formatter import prepare_pin	1✔
24
25	# SuperCluster configuration constants
26	MIN_ZOOM = 0	1✔
27	MAX_ZOOM = 16	1✔
28	CLUSTER_RADIUS = 200	1✔
29	CLUSTER_EXTENT = 512	1✔
30
31	logger = logging.getLogger(__name__)	1✔
32
33
34	def make_tuple_translation(keys_to_translate):	1✔
35	return [(x, gettext(x)) for x in keys_to_translate]	1✔
36
37
38	def get_or_none(data, *keys):	1✔
39	for key in keys:	1✔
40	if isinstance(data, dict):	1✔
41	data = data.get(key)	1✔
42	else:
43	return None	1✔
44	return data	1✔
45
46
47	def get_locations_from_request(database, request_args, as_basic_info=False):	1✔
48	"""
49	Shared helper to fetch locations from database based on request arguments.
50
51	Args:
52	database: Database instance
53	request_args: Request arguments (flask.request.args)
54	as_basic_info: If True, returns list of basic_info dicts, otherwise returns Location objects
55
56	Returns:
57	List of locations (either as objects or basic_info dicts)
58	"""
59	query_params = request_args.to_dict(flat=False)	1✔
60	all_locations = database.get_locations(query_params)	1✔
61
62	if as_basic_info:	1✔
63	return [x.basic_info() for x in all_locations]	1✔
64
65	return all_locations	1✔
66
67
68	def core_pages(	1✔
69	database,
70	languages: LanguagesMapping,
71	notifier_function,
72	csrf_generator,
73	location_model,
74	feature_flags={},
75	) -> Blueprint:
76	core_api_blueprint = Blueprint("api", __name__, url_prefix="/api")	1✔
77	core_api = Api(core_api_blueprint, doc="/doc", version="0.1")	1✔
78
79	location_report_model = core_api.model(	1✔
80	"LocationReport",
81	{
82	"id": fields.String(required=True, description="Location ID"),
83	"description": fields.String(required=True, description="Description of the problem"),
84	},
85	)
86
87	# TODO get this from Location pydantic model
88	suggested_location_model = core_api.model(	1✔
89	"LocationSuggestion",
90	{
91	"name": fields.String(required=False, description="Organization name"),
92	"position": fields.String(required=True, description="Location of the suggestion"),
93	"photo": fields.String(required=False, description="Photo of the location"),
94	},
95	)
96
97	@core_api.route("/suggest-new-point")	1✔
98	class NewLocation(Resource):	1✔
99	@core_api.expect(suggested_location_model)	1✔
100	def post(self):	1✔
101	"""Suggest new location"""
102	try:	1✔
103	suggested_location = request.get_json()	1✔
104	suggested_location.update({"uuid": str(uuid.uuid4())})	1✔
105	location = location_model.model_validate(suggested_location)	1✔
106	database.add_suggestion(location.model_dump())	1✔
107	message = (	1✔
108	f"A new location has been suggested under uuid: '{location.uuid}' "
109	f"at position: {location.position}"
110	)
111	notifier_function(message)	1✔
112	except BadRequest:	1✔
113	logger.warning("Invalid JSON in suggest endpoint")	1✔
114	return make_response(jsonify({"message": "Invalid request data"}), 400)	1✔
115	except LocationValidationError as e:	1✔
116	logger.warning(	1✔
117	"Location validation failed in suggest endpoint",
118	extra={"errors": e.validation_errors},
119	)
120	return make_response(jsonify({"message": "Invalid location data"}), 400)	1✔
NEW 121	except Exception:	×
NEW 122	logger.error("Error in suggest location endpoint", exc_info=True)	×
NEW 123	return make_response(	×
124	jsonify({"message": "An error occurred while processing your suggestion"}), 500
125	)
126	return make_response(jsonify({"message": "Location suggested"}), 200)	1✔
127
128	@core_api.route("/report-location")	1✔
129	class ReportLocation(Resource):	1✔
130	@core_api.expect(location_report_model)	1✔
131	def post(self):	1✔
132	"""Report location"""
133	try:	1✔
134	location_report = request.get_json()	1✔
135	report = {	1✔
136	"uuid": str(uuid.uuid4()),
137	"location_id": location_report["id"],
138	"description": location_report["description"],
139	"status": "pending",
140	"priority": "medium",
141	}
142	database.add_report(report)	1✔
143	message = (	1✔
144	f"A location has been reported: '{location_report['id']}' "
145	f"with problem: {location_report['description']}"
146	)
147	notifier_function(message)	1✔
148	except BadRequest:	1✔
149	logger.warning("Invalid JSON in report location endpoint")	1✔
150	return make_response(jsonify({"message": "Invalid request data"}), 400)	1✔
151	except KeyError as e:	1✔
152	logger.warning(	1✔
153	"Missing required field in report location", extra={"missing_field": str(e)}
154	)
155	error_message = gettext("Error reporting location")	1✔
156	return make_response(jsonify({"message": error_message}), 400)	1✔
NEW 157	except Exception:	×
NEW 158	logger.error("Error in report location endpoint", exc_info=True)	×
UNCOV 159	error_message = gettext("Error sending notification")	×
NEW 160	return make_response(jsonify({"message": error_message}), 500)	×
161	return make_response(jsonify({"message": gettext("Location reported")}), 200)	1✔
162
163	@core_api.route("/locations")	1✔
164	class GetLocations(Resource):	1✔
165	def get(self):	1✔
166	"""
167	Shows list of locations with uuid and position
168	"""
169	locations = get_locations_from_request(database, request.args, as_basic_info=True)	1✔
170	return jsonify(locations)	1✔
171
172	@core_api.route("/locations-clustered")	1✔
173	class GetLocationsClustered(Resource):	1✔
174	def get(self):	1✔
175	"""
176	Shows list of locations with uuid, position and clusters
177	"""
178	try:	1✔
179	query_params = request.args.to_dict(flat=False)	1✔
180	zoom = int(query_params.get("zoom", [7])[0])	1✔
181
182	# Validate zoom level (aligned with SuperCluster min_zoom/max_zoom)
183	if not MIN_ZOOM <= zoom <= MAX_ZOOM:	1✔
184	return make_response(	1✔
185	jsonify({"message": f"Zoom must be between {MIN_ZOOM} and {MAX_ZOOM}"}),
186	400,
187	)
188
189	points = get_locations_from_request(database, request.args, as_basic_info=True)	1✔
190	if not points:	1✔
191	return jsonify([])	1✔
192
193	points_numpy = numpy.array(	1✔
194	[(point["position"][0], point["position"][1]) for point in points]
195	)
196
197	index = pysupercluster.SuperCluster(	1✔
198	points_numpy,
199	min_zoom=MIN_ZOOM,
200	max_zoom=MAX_ZOOM,
201	radius=CLUSTER_RADIUS,
202	extent=CLUSTER_EXTENT,
203	)
204
205	clusters = index.getClusters(	1✔
206	top_left=(-180.0, 90.0),
207	bottom_right=(180.0, -90.0),
208	zoom=zoom,
209	)
210	clusters = match_clusters_uuids(points, clusters)	1✔
211
212	return jsonify(map_clustering_data_to_proper_lazy_loading_object(clusters))	1✔
213	except ValueError as e:	1✔
214	logger.warning("Invalid parameter in clustering request: %s", e)	1✔
215	return make_response(jsonify({"message": "Invalid parameters provided"}), 400)	1✔
216	except Exception as e:	1✔
217	logger.error("Clustering operation failed: %s", e, exc_info=True)	1✔
218	return make_response(	1✔
219	jsonify({"message": "An error occurred during clustering"}), 500
220	)
221
222	@core_api.route("/location/<location_id>")	1✔
223	class GetLocation(Resource):	1✔
224	def get(self, location_id):	1✔
225	"""
226	Shows a single location with all data
227	"""
228	location = database.get_location(location_id)	1✔
229	visible_data = database.get_visible_data()	1✔
230	meta_data = database.get_meta_data()	1✔
231
232	formatted_data = prepare_pin(location.model_dump(), visible_data, meta_data)	1✔
233	return jsonify(formatted_data)	1✔
234
235	@core_api.route("/version")	1✔
236	class Version(Resource):	1✔
237	def get(self):	1✔
238	"""Shows backend version"""
239	version_info = {"backend": importlib.metadata.version("goodmap")}	1✔
240	return jsonify(version_info)	1✔
241
242	@core_api.route("/categories")	1✔
243	class Categories(Resource):	1✔
244	def get(self):	1✔
245	"""Shows all available categories"""
246	raw_categories = database.get_categories()	1✔
247	categories = make_tuple_translation(raw_categories)	1✔
248
249	if not feature_flags.get("CATEGORIES_HELP", False):	1✔
250	return jsonify(categories)	1✔
251	else:
252	category_data = database.get_category_data()	1✔
253	categories_help = category_data["categories_help"]	1✔
254	proper_categories_help = []	1✔
255	if categories_help is not None:	1✔
256	for option in categories_help:	1✔
257	proper_categories_help.append(	1✔
258	{option: gettext(f"categories_help_{option}")}
259	)
260
261	return jsonify({"categories": categories, "categories_help": proper_categories_help})	1✔
262
263	@core_api.route("/languages")	1✔
264	class Languages(Resource):	1✔
265	def get(self):	1✔
266	"""Shows all available languages"""
267	return jsonify(languages)	1✔
268
269	@core_api.route("/category/<category_type>")	1✔
270	class CategoryTypes(Resource):	1✔
271	def get(self, category_type):	1✔
272	"""Shows all available types in category"""
273	category_data = database.get_category_data(category_type)	1✔
274	local_data = make_tuple_translation(category_data["categories"][category_type])	1✔
275
276	categories_options_help = get_or_none(	1✔
277	category_data, "categories_options_help", category_type
278	)
279	proper_categories_options_help = []	1✔
280	if categories_options_help is not None:	1✔
281	for option in categories_options_help:	1✔
282	proper_categories_options_help.append(	1✔
283	{option: gettext(f"categories_options_help_{option}")}
284	)
285	if not feature_flags.get("CATEGORIES_HELP", False):	1!
286	return jsonify(local_data)	1✔
287	else:
288	return jsonify(	1✔
289	{
290	"categories_options": local_data,
291	"categories_options_help": proper_categories_options_help,
292	}
293	)
294
295	@core_api.route("/generate-csrf-token")	1✔
296	class CsrfToken(Resource):	1✔
297	def get(self):	1✔
298	csrf_token = csrf_generator()	1✔
299	return {"csrf_token": csrf_token}	1✔
300
301	@core_api.route("/admin/locations")	1✔
302	class AdminManageLocations(Resource):	1✔
303	def get(self):	1✔
304	"""
305	Shows full list of locations, with optional server-side pagination, sorting,
306	and filtering.
307	"""
308	query_params = request.args.to_dict(flat=False)	1✔
309	if "sort_by" not in query_params:	1✔
310	query_params["sort_by"] = ["name"]	1✔
311	result = database.get_locations_paginated(query_params)	1✔
312	return jsonify(result)	1✔
313
314	def post(self):	1✔
315	"""
316	Creates a new location
317	"""
318	location_data = request.get_json()	1✔
319	try:	1✔
320	location_data.update({"uuid": str(uuid.uuid4())})	1✔
321	location = location_model.model_validate(location_data)	1✔
322	database.add_location(location.model_dump())	1✔
323	except LocationValidationError as e:	1✔
324	logger.warning(	1✔
325	"Location validation failed",
326	extra={"uuid": e.uuid, "errors": e.validation_errors},
327	)
328	return make_response(jsonify({"message": "Invalid location data"}), 400)	1✔
329	except LocationAlreadyExistsError as e:	1✔
NEW 330	logger.warning("Attempted to create duplicate location", extra={"uuid": e.uuid})	×
NEW 331	return make_response(jsonify({"message": "Location already exists"}), 409)	×
332	except Exception:	1✔
333	logger.error("Error creating location", exc_info=True)	1✔
334	return make_response(jsonify({"message": "An internal error occurred"}), 500)	1✔
335	return jsonify(location.model_dump())	1✔
336
337	@core_api.route("/admin/locations/<location_id>")	1✔
338	class AdminManageLocation(Resource):	1✔
339	def put(self, location_id):	1✔
340	"""
341	Updates a single location
342	"""
343	location_data = request.get_json()	1✔
344	try:	1✔
345	location_data.update({"uuid": location_id})	1✔
346	location = location_model.model_validate(location_data)	1✔
347	database.update_location(location_id, location.model_dump())	1✔
348	except LocationValidationError as e:	1✔
349	logger.warning(	1✔
350	"Location validation failed",
351	extra={"uuid": e.uuid, "errors": e.validation_errors},
352	)
353	return make_response(jsonify({"message": "Invalid location data"}), 400)	1✔
354	except LocationNotFoundError as e:	1✔
355	logger.info("Location not found for update", extra={"uuid": e.uuid})	1✔
356	return make_response(jsonify({"message": "Location not found"}), 404)	1✔
357	except Exception:	1✔
358	logger.error("Error updating location", exc_info=True)	1✔
359	return make_response(jsonify({"message": "An internal error occurred"}), 500)	1✔
360	return jsonify(location.model_dump())	1✔
361
362	def delete(self, location_id):	1✔
363	"""
364	Deletes a single location
365	"""
366	try:	1✔
367	database.delete_location(location_id)	1✔
368	except LocationNotFoundError as e:	1✔
369	logger.info("Location not found for deletion", extra={"uuid": e.uuid})	1✔
370	return make_response(jsonify({"message": "Location not found"}), 404)	1✔
371	except Exception:	1✔
372	logger.error("Error deleting location", exc_info=True)	1✔
373	return make_response(jsonify({"message": "An internal error occurred"}), 500)	1✔
374	return "", 204	1✔
375
376	@core_api.route("/admin/suggestions")	1✔
377	class AdminManageSuggestions(Resource):	1✔
378	def get(self):	1✔
379	"""
380	List location suggestions, with optional server-side pagination, sorting,
381	and filtering by status.
382	"""
383	query_params = request.args.to_dict(flat=False)	1✔
384	result = database.get_suggestions_paginated(query_params)	1✔
385	return jsonify(result)	1✔
386
387	@core_api.route("/admin/suggestions/<suggestion_id>")	1✔
388	class AdminManageSuggestion(Resource):	1✔
389	def put(self, suggestion_id):	1✔
390	"""
391	Accept or reject a location suggestion
392	"""
393	try:	1✔
394	data = request.get_json()	1✔
395	status = data.get("status")	1✔
396	if status not in ("accepted", "rejected"):	1✔
397	return make_response(jsonify({"message": "Invalid status"}), 400)	1✔
398	suggestion = database.get_suggestion(suggestion_id)	1✔
399	if not suggestion:	1✔
400	return make_response(jsonify({"message": "Suggestion not found"}), 404)	1✔
401	if suggestion.get("status") != "pending":	1✔
402	return make_response(jsonify({"message": "Suggestion already processed"}), 400)	1✔
403	if status == "accepted":	1✔
404	suggestion_data = {k: v for k, v in suggestion.items() if k != "status"}	1✔
405	database.add_location(suggestion_data)	1✔
406	database.update_suggestion(suggestion_id, status)	1✔
407	except LocationValidationError as e:	1✔
NEW 408	logger.warning(	×
409	"Location validation failed in suggestion",
410	extra={"uuid": e.uuid, "errors": e.validation_errors},
411	)
NEW 412	return make_response(jsonify({"message": "Invalid location data"}), 400)	×
413	except LocationAlreadyExistsError as e:	1✔
NEW 414	logger.warning(	×
415	"Attempted to create duplicate location from suggestion", extra={"uuid": e.uuid}
416	)
NEW 417	return make_response(jsonify({"message": "Location already exists"}), 409)	×
418	except Exception:	1✔
419	logger.error("Error processing suggestion", exc_info=True)	1✔
420	return make_response(jsonify({"message": "An internal error occurred"}), 500)	1✔
421	return jsonify(database.get_suggestion(suggestion_id))	1✔
422
423	@core_api.route("/admin/reports")	1✔
424	class AdminManageReports(Resource):	1✔
425	def get(self):	1✔
426	"""
427	List location reports, with optional server-side pagination, sorting,
428	and filtering by status/priority.
429	"""
430	query_params = request.args.to_dict(flat=False)	1✔
431	result = database.get_reports_paginated(query_params)	1✔
432	return jsonify(result)	1✔
433
434	@core_api.route("/admin/reports/<report_id>")	1✔
435	class AdminManageReport(Resource):	1✔
436	def put(self, report_id):	1✔
437	"""
438	Update a report's status and/or priority
439	"""
440	try:	1✔
441	data = request.get_json()	1✔
442	status = data.get("status")	1✔
443	priority = data.get("priority")	1✔
444	valid_status = ("resolved", "rejected")	1✔
445	valid_priority = ("critical", "high", "medium", "low")	1✔
446	if status and status not in valid_status:	1✔
447	return make_response(jsonify({"message": "Invalid status"}), 400)	1✔
448	if priority and priority not in valid_priority:	1✔
449	return make_response(jsonify({"message": "Invalid priority"}), 400)	1✔
450	report = database.get_report(report_id)	1✔
451	if not report:	1✔
452	return make_response(jsonify({"message": "Report not found"}), 404)	1✔
453	database.update_report(report_id, status=status, priority=priority)	1✔
454	except BadRequest:	1✔
455	logger.warning("Invalid JSON in report update endpoint")	1✔
456	return make_response(jsonify({"message": "Invalid request data"}), 400)	1✔
457	except ReportNotFoundError as e:	1✔
NEW 458	logger.info("Report not found for update", extra={"uuid": e.uuid})	×
NEW 459	return make_response(jsonify({"message": "Report not found"}), 404)	×
460	except Exception:	1✔
461	logger.error("Error updating report", exc_info=True)	1✔
462	return make_response(jsonify({"message": "An internal error occurred"}), 500)	1✔
463	return jsonify(database.get_report(report_id))	1✔
464
465	return core_api_blueprint	1✔

Problematy / goodmap / 20583796061

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