17080930063

Committed 19 Aug 2025 08:23PM UTC coverage: 98.778% (-0.3%) from 99.125%

Build # 17080930063

Build Type

Pull #239

github

Committed by

web-flow

Commit Message

Merge 379fc1721 into eb5d05074

Pull Request Pull Request #239: Helps rebase

Coverage Stats

344 of 350 branches covered (98.29%)

Branch coverage included in aggregate %.

21 of 22 new or added lines in 1 file covered. (95.45%)

707 of 714 relevant lines covered (99.02%)

0.99 hits per line

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

98.21

/goodmap/core_api.py

import importlib.metadata
import uuid

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.formatter import prepare_pin


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 paginate_results(items, raw_params, sort_by_default=None):
    """
    Apply pagination and sorting to a list of items.

    Args:
        items: The list of items to paginate
        raw_params: The query parameters dictionary

    Returns:
        Tuple of (paginated_items, pagination_metadata)
    """
    # Extract pagination parameters
    try:
        page = int(raw_params.pop("page", ["1"])[0])
    except ValueError:
        page = 1

    per_page_raw = raw_params.pop("per_page", [None])[0]
    if per_page_raw is None:
        per_page = 20
    elif per_page_raw == "all":
        per_page = None
    else:
        try:
            per_page = max(1, int(per_page_raw))
        except ValueError:
            per_page = 20

    sort_by = raw_params.pop("sort_by", [None])[0] or sort_by_default
    sort_order = raw_params.pop("sort_order", ["asc"])[0].lower()

    def get_sort_key(item):
        if not sort_by:
            return None

        value = None
        if isinstance(item, dict):
            value = item.get(sort_by)
        else:
            value = getattr(item, sort_by, None)

        return (value is not None, value)

    if sort_by:
        reverse = sort_order == "desc"
        items.sort(key=get_sort_key, reverse=reverse)

    # Apply pagination
    total = len(items)
    if per_page:
        start = (page - 1) * per_page
        end = start + per_page
        page_items = items[start:end]
        total_pages = (total + per_page - 1) // per_page
    else:
        page_items = items
        total_pages = 1
        page = 1
        per_page = total

    return page_items, {
        "total": total,
        "page": page,
        "per_page": per_page,
        "total_pages": total_pages,
    }


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 ValueError 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
            """
            query_params = request.args.to_dict(flat=False)
            all_locations = database.get_locations(query_params)
            return jsonify([x.basic_info() for x in all_locations])

    @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)

            # TODO getting visible_data and meta_data should be taken from db methods
            #    e.g. db.get_visible_data() and db.get_meta_data()
            #    visible_data and meta_data should be models
            all_data = database.get_data()
            visible_data = all_data["visible_data"]
            meta_data = all_data["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"""
            all_data = database.get_data()
            categories = make_tuple_translation(all_data["categories"].keys())

            if not feature_flags.get("CATEGORIES_HELP", False):
                return jsonify(categories)
            else:
                categories_help = all_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"""
            all_data = database.get_data()
            local_data = make_tuple_translation(all_data["categories"][category_type])

            categories_options_help = get_or_none(
                all_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.
            """
            # Raw query params from request
            raw_params = request.args.to_dict(flat=False)
            all_locations = database.get_locations(raw_params)
            page_items, pagination = paginate_results(
                all_locations, raw_params, sort_by_default="name"
            )
            items = [x.model_dump() for x in page_items]
            return jsonify({"items": items, **pagination})

        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 ValueError as e:
                return make_response(jsonify({"message": f"Invalid location data: {e}"}), 400)
            except Exception as e:
                return make_response(jsonify({"message": f"Error creating location: {e}"}), 400)
            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 ValueError as e:
                return make_response(jsonify({"message": f"Invalid location data: {e}"}), 400)
            except Exception as e:
                return make_response(jsonify({"message": f"Error updating location: {e}"}), 400)
            return jsonify(location.model_dump())

        def delete(self, location_id):
            """
            Deletes a single location
            """
            try:
                database.delete_location(location_id)
            except ValueError 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}"}), 400)
            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.
            """
            raw_params = request.args.to_dict(flat=False)
            suggestions = database.get_suggestions(raw_params)
            page_items, pagination = paginate_results(suggestions, raw_params)
            return jsonify({"items": page_items, **pagination})

    @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 ValueError as e:
                return make_response(jsonify({"message": f"{e}"}), 400)
            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.
            """
            raw_params = request.args.to_dict(flat=False)
            reports = database.get_reports(raw_params)
            page_items, pagination = paginate_results(reports, raw_params)
            return jsonify({"items": page_items, **pagination})

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

Problematy / goodmap / 17080930063

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