• Home
  • Features
  • Pricing
  • Docs
  • Announcements
  • Sign In

Problematy / goodmap / 15472563239

05 Jun 2025 04:40PM UTC coverage: 99.125% (+1.1%) from 98.039%
15472563239

Pull #227

github

web-flow
Merge e0a801e92 into 34edc107c
Pull Request #227: Admin panel

331 of 334 branches covered (99.1%)

Branch coverage included in aggregate %.

393 of 398 new or added lines in 3 files covered. (98.74%)

688 of 694 relevant lines covered (99.14%)

0.99 hits per line

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

99.33
/goodmap/core_api.py
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 paginate_results(items, raw_params, sort_by_default=None):
1✔
17
    """
18
    Apply pagination and sorting to a list of items.
19

20
    Args:
21
        items: The list of items to paginate
22
        raw_params: The query parameters dictionary
23

24
    Returns:
25
        Tuple of (paginated_items, pagination_metadata)
26
    """
27
    # Extract pagination parameters
28
    try:
1✔
29
        page = int(raw_params.pop("page", ["1"])[0])
1✔
30
    except ValueError:
1✔
31
        page = 1
1✔
32

33
    per_page_raw = raw_params.pop("per_page", [None])[0]
1✔
34
    if per_page_raw is None:
1✔
35
        per_page = 20
1✔
36
    elif per_page_raw == "all":
1✔
37
        per_page = None
1✔
38
    else:
39
        try:
1✔
40
            per_page = max(1, int(per_page_raw))
1✔
41
        except ValueError:
1✔
42
            per_page = 20
1✔
43

44
    sort_by = raw_params.pop("sort_by", [None])[0] or sort_by_default
1✔
45
    sort_order = raw_params.pop("sort_order", ["asc"])[0].lower()
1✔
46

47
    def get_sort_key(item):
1✔
48
        if not sort_by:
1✔
NEW
49
            return None
×
50

51
        value = None
1✔
52
        if isinstance(item, dict):
1✔
53
            value = item.get(sort_by)
1✔
54
        else:
55
            value = getattr(item, sort_by, None)
1✔
56

57
        return (value is not None, value)
1✔
58

59
    if sort_by:
1✔
60
        reverse = sort_order == "desc"
1✔
61
        items.sort(key=get_sort_key, reverse=reverse)
1✔
62

63
    # Apply pagination
64
    total = len(items)
1✔
65
    if per_page:
1✔
66
        start = (page - 1) * per_page
1✔
67
        end = start + per_page
1✔
68
        page_items = items[start:end]
1✔
69
        total_pages = (total + per_page - 1) // per_page
1✔
70
    else:
71
        page_items = items
1✔
72
        total_pages = 1
1✔
73
        page = 1
1✔
74
        per_page = total
1✔
75

76
    return page_items, {
1✔
77
        "total": total,
78
        "page": page,
79
        "per_page": per_page,
80
        "total_pages": total_pages,
81
    }
82

83

84
def core_pages(
1✔
85
    database, languages: LanguagesMapping, notifier_function, csrf_generator, location_model
86
) -> Blueprint:
87
    core_api_blueprint = Blueprint("api", __name__, url_prefix="/api")
1✔
88
    core_api = Api(core_api_blueprint, doc="/doc", version="0.1")
1✔
89

90
    location_report_model = core_api.model(
1✔
91
        "LocationReport",
92
        {
93
            "id": fields.String(required=True, description="Location ID"),
94
            "description": fields.String(required=True, description="Description of the problem"),
95
        },
96
    )
97

98
    # TODO get this from Location pydantic model
99
    suggested_location_model = core_api.model(
1✔
100
        "LocationSuggestion",
101
        {
102
            "name": fields.String(required=False, description="Organization name"),
103
            "position": fields.String(required=True, description="Location of the suggestion"),
104
            "photo": fields.String(required=False, description="Photo of the location"),
105
        },
106
    )
107

108
    @core_api.route("/suggest-new-point")
1✔
109
    class NewLocation(Resource):
1✔
110
        @core_api.expect(suggested_location_model)
1✔
111
        def post(self):
1✔
112
            """Suggest new location"""
113
            try:
1✔
114
                suggested_location = request.get_json()
1✔
115
                suggested_location.update({"uuid": str(uuid.uuid4())})
1✔
116
                location = location_model.model_validate(suggested_location)
1✔
117
                database.add_suggestion(location.model_dump())
1✔
118
                message = (
1✔
119
                    f"A new location has been suggested under uuid: '{location.uuid}' "
120
                    f"at position: {location.position}"
121
                )
122
                notifier_function(message)
1✔
123
            except ValueError as e:
1✔
124
                return make_response(jsonify({"message": f"Invalid location data: {e}"}), 400)
1✔
125
            except Exception as e:
1✔
126
                return make_response(jsonify({"message": f"Error sending notification : {e}"}), 400)
1✔
127
            return make_response(jsonify({"message": "Location suggested"}), 200)
1✔
128

129
    @core_api.route("/report-location")
1✔
130
    class ReportLocation(Resource):
1✔
131
        @core_api.expect(location_report_model)
1✔
132
        def post(self):
1✔
133
            """Report location"""
134
            try:
1✔
135
                location_report = request.get_json()
1✔
136
                report = {
1✔
137
                    "uuid": str(uuid.uuid4()),
138
                    "location_id": location_report["id"],
139
                    "description": location_report["description"],
140
                    "status": "pending",
141
                    "priority": "medium",
142
                }
143
                database.add_report(report)
1✔
144
                message = (
1✔
145
                    f"A location has been reported: '{location_report['id']}' "
146
                    f"with problem: {location_report['description']}"
147
                )
148
                notifier_function(message)
1✔
149
            except KeyError as e:
1✔
150
                error_message = gettext("Error reporting location")
1✔
151
                return make_response(jsonify({"message": f"{error_message} : {e}"}), 400)
1✔
152
            except Exception as e:
1✔
153
                error_message = gettext("Error sending notification")
1✔
154
                return make_response(jsonify({"message": f"{error_message} : {e}"}), 400)
1✔
155
            return make_response(jsonify({"message": gettext("Location reported")}), 200)
1✔
156

157
    @core_api.route("/locations")
1✔
158
    class GetLocations(Resource):
1✔
159
        def get(self):
1✔
160
            """
161
            Shows list of locations with uuid and position
162
            """
163
            query_params = request.args.to_dict(flat=False)
1✔
164
            all_locations = database.get_locations(query_params)
1✔
165
            return jsonify([x.basic_info() for x in all_locations])
1✔
166

167
    @core_api.route("/location/<location_id>")
1✔
168
    class GetLocation(Resource):
1✔
169
        def get(self, location_id):
1✔
170
            """
171
            Shows a single location with all data
172
            """
173
            location = database.get_location(location_id)
1✔
174

175
            # TODO getting visible_data and meta_data should be taken from db methods
176
            #    e.g. db.get_visible_data() and db.get_meta_data()
177
            #    visible_data and meta_data should be models
178
            all_data = database.get_data()
1✔
179
            visible_data = all_data["visible_data"]
1✔
180
            meta_data = all_data["meta_data"]
1✔
181

182
            formatted_data = prepare_pin(location.model_dump(), visible_data, meta_data)
1✔
183
            return jsonify(formatted_data)
1✔
184

185
    @core_api.route("/version")
1✔
186
    class Version(Resource):
1✔
187
        def get(self):
1✔
188
            """Shows backend version"""
189
            version_info = {"backend": importlib.metadata.version("goodmap")}
1✔
190
            return jsonify(version_info)
1✔
191

192
    @core_api.route("/categories")
1✔
193
    class Categories(Resource):
1✔
194
        def get(self):
1✔
195
            """Shows all available categories"""
196
            all_data = database.get_data()
1✔
197
            categories = make_tuple_translation(all_data["categories"].keys())
1✔
198
            return jsonify(categories)
1✔
199

200
    @core_api.route("/languages")
1✔
201
    class Languages(Resource):
1✔
202
        def get(self):
1✔
203
            """Shows all available languages"""
204
            return jsonify(languages)
1✔
205

206
    @core_api.route("/category/<category_type>")
1✔
207
    class CategoryTypes(Resource):
1✔
208
        def get(self, category_type):
1✔
209
            """Shows all available types in category"""
210
            all_data = database.get_data()
1✔
211
            local_data = make_tuple_translation(all_data["categories"][category_type])
1✔
212
            return jsonify(local_data)
1✔
213

214
    @core_api.route("/generate-csrf-token")
1✔
215
    class CsrfToken(Resource):
1✔
216
        def get(self):
1✔
217
            csrf_token = csrf_generator()
1✔
218
            return {"csrf_token": csrf_token}
1✔
219

220
    @core_api.route("/admin/locations")
1✔
221
    class AdminManageLocations(Resource):
1✔
222
        def get(self):
1✔
223
            """
224
            Shows full list of locations, with optional server-side pagination, sorting,
225
            and filtering.
226
            """
227
            # Raw query params from request
228
            raw_params = request.args.to_dict(flat=False)
1✔
229
            all_locations = database.get_locations(raw_params)
1✔
230
            page_items, pagination = paginate_results(
1✔
231
                all_locations, raw_params, sort_by_default="name"
232
            )
233
            items = [x.model_dump() for x in page_items]
1✔
234
            return jsonify({"items": items, **pagination})
1✔
235

236
        def post(self):
1✔
237
            """
238
            Creates a new location
239
            """
240
            location_data = request.get_json()
1✔
241
            try:
1✔
242
                location_data.update({"uuid": str(uuid.uuid4())})
1✔
243
                location = location_model.model_validate(location_data)
1✔
244
                database.add_location(location.model_dump())
1✔
245
            except ValueError as e:
1✔
246
                return make_response(jsonify({"message": f"Invalid location data: {e}"}), 400)
1✔
247
            except Exception as e:
1✔
248
                return make_response(jsonify({"message": f"Error creating location: {e}"}), 400)
1✔
249
            return jsonify(location.model_dump())
1✔
250

251
    @core_api.route("/admin/locations/<location_id>")
1✔
252
    class AdminManageLocation(Resource):
1✔
253
        def put(self, location_id):
1✔
254
            """
255
            Updates a single location
256
            """
257
            location_data = request.get_json()
1✔
258
            try:
1✔
259
                location_data.update({"uuid": location_id})
1✔
260
                location = location_model.model_validate(location_data)
1✔
261
                database.update_location(location_id, location.model_dump())
1✔
262
            except ValueError as e:
1✔
263
                return make_response(jsonify({"message": f"Invalid location data: {e}"}), 400)
1✔
264
            except Exception as e:
1✔
265
                return make_response(jsonify({"message": f"Error updating location: {e}"}), 400)
1✔
266
            return jsonify(location.model_dump())
1✔
267

268
        def delete(self, location_id):
1✔
269
            """
270
            Deletes a single location
271
            """
272
            try:
1✔
273
                database.delete_location(location_id)
1✔
274
            except ValueError as e:
1✔
275
                return make_response(jsonify({"message": f"Location not found: {e}"}), 404)
1✔
276
            except Exception as e:
1✔
277
                return make_response(jsonify({"message": f"Error deleting location: {e}"}), 400)
1✔
278
            return "", 204
1✔
279

280
    @core_api.route("/admin/suggestions")
1✔
281
    class AdminManageSuggestions(Resource):
1✔
282
        def get(self):
1✔
283
            """
284
            List location suggestions, with optional server-side pagination, sorting,
285
            and filtering by status.
286
            """
287
            raw_params = request.args.to_dict(flat=False)
1✔
288
            suggestions = database.get_suggestions(raw_params)
1✔
289
            page_items, pagination = paginate_results(suggestions, raw_params)
1✔
290
            return jsonify({"items": page_items, **pagination})
1✔
291

292
    @core_api.route("/admin/suggestions/<suggestion_id>")
1✔
293
    class AdminManageSuggestion(Resource):
1✔
294
        def put(self, suggestion_id):
1✔
295
            """
296
            Accept or reject a location suggestion
297
            """
298
            try:
1✔
299
                data = request.get_json()
1✔
300
                status = data.get("status")
1✔
301
                if status not in ("accepted", "rejected"):
1✔
302
                    return make_response(jsonify({"message": f"Invalid status: {status}"}), 400)
1✔
303
                suggestion = database.get_suggestion(suggestion_id)
1✔
304
                if not suggestion:
1✔
305
                    return make_response(jsonify({"message": "Suggestion not found"}), 404)
1✔
306
                if suggestion.get("status") != "pending":
1✔
307
                    return make_response(jsonify({"message": "Suggestion already processed"}), 400)
1✔
308
                if status == "accepted":
1✔
309
                    suggestion_data = {k: v for k, v in suggestion.items() if k != "status"}
1✔
310
                    database.add_location(suggestion_data)
1✔
311
                database.update_suggestion(suggestion_id, status)
1✔
312
            except ValueError as e:
1✔
313
                return make_response(jsonify({"message": f"{e}"}), 400)
1✔
314
            return jsonify(database.get_suggestion(suggestion_id))
1✔
315

316
    @core_api.route("/admin/reports")
1✔
317
    class AdminManageReports(Resource):
1✔
318
        def get(self):
1✔
319
            """
320
            List location reports, with optional server-side pagination, sorting,
321
            and filtering by status/priority.
322
            """
323
            raw_params = request.args.to_dict(flat=False)
1✔
324
            reports = database.get_reports(raw_params)
1✔
325
            page_items, pagination = paginate_results(reports, raw_params)
1✔
326
            return jsonify({"items": page_items, **pagination})
1✔
327

328
    @core_api.route("/admin/reports/<report_id>")
1✔
329
    class AdminManageReport(Resource):
1✔
330
        def put(self, report_id):
1✔
331
            """
332
            Update a report's status and/or priority
333
            """
334
            try:
1✔
335
                data = request.get_json()
1✔
336
                status = data.get("status")
1✔
337
                priority = data.get("priority")
1✔
338
                valid_status = ("resolved", "rejected")
1✔
339
                valid_priority = ("critical", "high", "medium", "low")
1✔
340
                if status and status not in valid_status:
1✔
341
                    return make_response(jsonify({"message": f"Invalid status: {status}"}), 400)
1✔
342
                if priority and priority not in valid_priority:
1✔
343
                    return make_response(jsonify({"message": f"Invalid priority: {priority}"}), 400)
1✔
344
                report = database.get_report(report_id)
1✔
345
                if not report:
1✔
346
                    return make_response(jsonify({"message": "Report not found"}), 404)
1✔
347
                database.update_report(report_id, status=status, priority=priority)
1✔
348
            except ValueError as e:
1✔
349
                return make_response(jsonify({"message": f"{e}"}), 400)
1✔
350
            return jsonify(database.get_report(report_id))
1✔
351

352
    return core_api_blueprint
1✔
STATUS · Troubleshooting · Open an Issue · Sales · Support · CAREERS · ENTERPRISE · START FREE TRIAL · SCHEDULE DEMO
ANNOUNCEMENTS · TWITTER · TOS & SLA · Supported CI Services · What's a CI service? · Automated Testing

© 2026 Coveralls, Inc