20582800505

Committed 29 Dec 2025 09:11PM UTC coverage: 98.177% (-0.6%) from 98.783%

Build # 20582800505

Build Type

Pull #278

github

Committed by

web-flow

Commit Message

Merge a1fc4a3ed into 783f855e7

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

Run Details

552 of 572 branches covered (96.5%)

Branch coverage included in aggregate %.

85 of 91 new or added lines in 4 files covered. (93.41%)

3 existing lines in 2 files now uncovered.

1225 of 1238 relevant lines covered (98.95%)

0.99 hits per line

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

98.0

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

Problematy / goodmap / 20582800505

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