21304342528

Committed 23 Jan 2026 11:15PM UTC coverage: 98.935%. First build

Build # 21304342528

Build Type

Pull #324

github

Committed by

web-flow

Commit Message

Merge 3e21f0ed2 into 0778a150c

Pull Request Pull Request #324: feat: images support introduced

Run Details

12 of 19 new or added lines in 2 files covered. (63.16%)

1486 of 1502 relevant lines covered (98.93%)

0.99 hits per line

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

96.82

/goodmap/core_api.py

import importlib.metadata
import logging
import uuid

import deprecation
import numpy
import pysupercluster
from flask import Blueprint, jsonify, make_response, request
from flask_babel import gettext
from platzky.attachment import AttachmentProtocol
from platzky.config import LanguagesMapping
from spectree import Response, SpecTree

from goodmap.api_models import (
    CSRFTokenResponse,
    ErrorResponse,
    LocationReportRequest,
    LocationReportResponse,
    SuccessResponse,
    VersionResponse,
)
from goodmap.clustering import (
    map_clustering_data_to_proper_lazy_loading_object,
    match_clusters_uuids,
)
from goodmap.exceptions import LocationValidationError
from goodmap.formatter import prepare_pin
from goodmap.json_security import (
    MAX_JSON_DEPTH_LOCATION,
    JSONDepthError,
    JSONSizeError,
    safe_json_loads,
)

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

# Error message constants
ERROR_INVALID_REQUEST_DATA = "Invalid request data"
ERROR_INVALID_LOCATION_DATA = "Invalid location data"
ERROR_LOCATION_NOT_FOUND = "Location not found"

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):
    """
    Shared helper to fetch locations from database based on request arguments.

    Args:
        database: Database instance
        request_args: Request arguments (flask.request.args)

    Returns:
        List of locations as basic_info dicts
    """
    query_params = request_args.to_dict(flat=False)
    all_locations = database.get_locations(query_params)
    return [x.basic_info() for x in all_locations]


def core_pages(
    database,
    languages: LanguagesMapping,
    notifier_function,
    csrf_generator,
    location_model,
    photo_attachment_class: type[AttachmentProtocol],
    feature_flags={},
) -> Blueprint:
    core_api_blueprint = Blueprint("api", __name__, url_prefix="/api")

    # Initialize Spectree for API documentation and validation
    # Use simple naming strategy without hashes for cleaner schema names
    from typing import Any, Type

    def _clean_model_name(model: Type[Any]) -> str:
        return model.__name__

    spec = SpecTree(
        "flask",
        title="Goodmap API",
        version="0.1",
        path="doc",
        annotations=True,
        naming_strategy=_clean_model_name,  # Use clean model names without hash
    )

    @core_api_blueprint.route("/suggest-new-point", methods=["POST"])
    @spec.validate(resp=Response(HTTP_200=SuccessResponse, HTTP_400=ErrorResponse))
    def suggest_new_point():
        """Suggest new location for review.

        Accepts location data either as JSON or multipart/form-data.
        All fields are validated using Pydantic location model.
        """
        import json as json_lib

        try:
            # Initialize photo attachment (only populated for multipart/form-data)
            photo_attachment = None

            # Handle both multipart/form-data (with file uploads) and JSON
            if request.content_type and request.content_type.startswith("multipart/form-data"):
                # Parse form data dynamically
                suggested_location = {}

                for key in request.form:
                    value = request.form[key]
                    # Try to parse as JSON for complex types (arrays, objects, position)
                    try:
                        # SECURITY: Use safe_json_loads with strict depth limit
                        # MAX_JSON_DEPTH_LOCATION=1: arrays/objects of primitives only
                        suggested_location[key] = safe_json_loads(
                            value, max_depth=MAX_JSON_DEPTH_LOCATION
                        )
                    except (JSONDepthError, JSONSizeError) as e:
                        # Log security event and return 400
                        logger.warning(
                            f"JSON parsing blocked for security: {e}",
                            extra={"field": key, "value_size": len(value)},
                        )
                        return make_response(
                            jsonify(
                                {
                                    "message": (
                                        "Invalid request: JSON payload too complex or too large"
                                    ),
                                    "error": str(e),
                                }
                            ),
                            400,
                        )
                    except ValueError:  # JSONDecodeError inherits from ValueError
                        # If not JSON, use as-is (simple string values)
                        suggested_location[key] = value

                # Extract and validate photo attachment if present
                photo_file = request.files.get("photo")
                if photo_file and photo_file.filename:
                    photo_content = photo_file.read()
                    photo_mime = photo_file.content_type or "application/octet-stream"

                    # Validate using configured Attachment class
                    try:
                        photo_attachment = photo_attachment_class(
                            photo_file.filename, photo_content, photo_mime
                        )
                    except ValueError as e:
                        logger.warning("Rejected photo: %s", e)
                        return make_response(jsonify({"message": str(e)}), 400)
            else:
                # Parse JSON data with security checks (depth/size protection)
                raw_data = request.get_data(as_text=True)
                if not raw_data:
                    logger.warning("Empty JSON body in suggest endpoint")
                    return make_response(jsonify({"message": ERROR_INVALID_REQUEST_DATA}), 400)
                try:
                    suggested_location = safe_json_loads(
                        raw_data, max_depth=MAX_JSON_DEPTH_LOCATION
                    )
                except (JSONDepthError, JSONSizeError) as e:
                    logger.warning(
                        f"JSON parsing blocked for security: {e}",
                        extra={"value_size": len(raw_data)},
                    )
                    return make_response(
                        jsonify(
                            {
                                "message": (
                                    "Invalid request: JSON payload too complex or too large"
                                ),
                                "error": str(e),
                            }
                        ),
                        400,
                    )
                except ValueError:
                    logger.warning("Invalid JSON in suggest endpoint")
                    return make_response(jsonify({"message": ERROR_INVALID_REQUEST_DATA}), 400)
                if suggested_location is None:
                    logger.warning("Null JSON value in suggest endpoint")
                    return make_response(jsonify({"message": ERROR_INVALID_REQUEST_DATA}), 400)

            suggested_location.update({"uuid": str(uuid.uuid4())})
            location = location_model.model_validate(suggested_location)
            database.add_suggestion(location.model_dump())
            message = gettext("A new location has been suggested with details")
            notifier_message = f"{message}: {json_lib.dumps(suggested_location, indent=2)}"
            attachments = [photo_attachment] if photo_attachment else None
            notifier_function(notifier_message, attachments=attachments)
        except LocationValidationError as e:
            # NOTE: validation_errors includes input values from the location model fields:
            # - Core fields: position (lat/long), uuid, remark
            # - Dynamic fields: categories and obligatory_fields configured per deployment
            # These are geographic/categorical data, NOT PII (no email, phone, names of people).
            # Safe to log for debugging. If PII fields are ever added to the location model,
            # strip 'input' from validation_errors before logging.
            logger.warning(
                "Location validation failed in suggest endpoint: %s",
                e.validation_errors,
                extra={"errors": e.validation_errors},
            )
            return make_response(jsonify({"message": ERROR_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_blueprint.route("/report-location", methods=["POST"])
    @spec.validate(
        json=LocationReportRequest,
        resp=Response(HTTP_200=LocationReportResponse, HTTP_400=ErrorResponse),
    )
    def report_location():
        """Report a problem with a location.

        Allows users to report issues with existing locations,
        such as incorrect information or closed establishments.
        """
        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 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_blueprint.route("/locations", methods=["GET"])
    @spec.validate()
    def get_locations():
        """Get list of locations with basic info.

        Returns locations filtered by query parameters,
        showing only uuid, position, and remark flag.
        """
        locations = get_locations_from_request(database, request.args)
        return jsonify(locations)

    @core_api_blueprint.route("/locations-clustered", methods=["GET"])
    @spec.validate(resp=Response(HTTP_400=ErrorResponse))
    def get_locations_clustered():
        """Get clustered locations for map display.

        Returns locations grouped into clusters based on zoom level,
        optimized for rendering on interactive maps.
        """
        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)
            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_blueprint.route("/location/<location_id>", methods=["GET"])
    @spec.validate(resp=Response(HTTP_404=ErrorResponse))
    def get_location(location_id):
        """Get detailed information for a single location.

        Returns full location data including all custom fields,
        formatted for display in the location details view.
        """
        location = database.get_location(location_id)
        if location is None:
            logger.info(ERROR_LOCATION_NOT_FOUND, extra={"uuid": location_id})
            return make_response(jsonify({"message": ERROR_LOCATION_NOT_FOUND}), 404)

        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_blueprint.route("/version", methods=["GET"])
    @spec.validate(resp=Response(HTTP_200=VersionResponse))
    def get_version():
        """Get backend version information.

        Returns the current version of the Goodmap backend.
        """
        version_info = {"backend": importlib.metadata.version("goodmap")}
        return jsonify(version_info)

    @core_api_blueprint.route("/generate-csrf-token", methods=["GET"])
    @spec.validate(resp=Response(HTTP_200=CSRFTokenResponse))
    @deprecation.deprecated(
        deprecated_in="1.1.8",
        details="This endpoint for explicit CSRF token generation is deprecated. "
        "CSRF protection remains active in the application.",
    )
    def generate_csrf_token():
        """Generate CSRF token (DEPRECATED).

        This endpoint is deprecated and maintained only for backward compatibility.
        CSRF protection remains active in the application.
        """
        csrf_token = csrf_generator()
        return {"csrf_token": csrf_token}

    @core_api_blueprint.route("/categories", methods=["GET"])
    @spec.validate()
    def get_categories():
        """Get all available location categories.

        Returns list of categories with optional help text
        if CATEGORIES_HELP feature flag is enabled.
        """
        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.get("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_blueprint.route("/categories-full", methods=["GET"])
    @spec.validate()
    def get_categories_full():
        """Get all categories with their subcategory options in a single request.

        Returns combined category data to reduce API calls for filter panel loading.
        This endpoint eliminates the need for multiple sequential requests.
        """
        categories_data = database.get_category_data()
        result = []

        categories_options_help = categories_data.get("categories_options_help", {})

        for key, options in categories_data["categories"].items():
            category_entry = {
                "key": key,
                "name": gettext(key),
                "options": make_tuple_translation(options),
            }

            if feature_flags.get("CATEGORIES_HELP", False):
                option_help_list = categories_options_help.get(key, [])
                proper_options_help = []
                for option in option_help_list:
                    proper_options_help.append(
                        {option: gettext(f"categories_options_help_{option}")}
                    )
                category_entry["options_help"] = proper_options_help

            result.append(category_entry)

        response = {"categories": result}

        if feature_flags.get("CATEGORIES_HELP", False):
            categories_help = categories_data.get("categories_help", [])
            proper_categories_help = []
            for option in categories_help:
                proper_categories_help.append({option: gettext(f"categories_help_{option}")})
            response["categories_help"] = proper_categories_help

        return jsonify(response)

    @core_api_blueprint.route("/languages", methods=["GET"])
    @spec.validate()
    def get_languages():
        """Get all available interface languages.

        Returns list of supported languages for the application.
        """
        return jsonify(languages)

    @core_api_blueprint.route("/category/<category_type>", methods=["GET"])
    @spec.validate()
    def get_category_types(category_type):
        """Get all available options for a specific category.

        Returns list of category options with optional help text
        if CATEGORIES_HELP feature flag is enabled.
        """
        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,
                }
            )

    # Register Spectree with blueprint after all routes are defined
    spec.register(core_api_blueprint)

    @core_api_blueprint.route("/doc")
    def api_doc_index():
        """Return links to available API documentation formats."""
        html = """<!DOCTYPE html>
<html><head><title>API Documentation</title></head>
<body>
<h1>API Documentation</h1>
<ul>
<li><a href="/api/doc/swagger/">Swagger UI</a></li>
<li><a href="/api/doc/redoc/">ReDoc</a></li>
<li><a href="/api/doc/openapi.json">OpenAPI JSON</a></li>
</ul>
</body></html>"""
        return html, 200, {"Content-Type": "text/html"}

    return core_api_blueprint

1	import importlib.metadata	1✔
2	import logging	1✔
3	import uuid	1✔
4
5	import deprecation	1✔
6	import numpy	1✔
7	import pysupercluster	1✔
8	from flask import Blueprint, jsonify, make_response, request	1✔
9	from flask_babel import gettext	1✔
10	from platzky.attachment import AttachmentProtocol	1✔
11	from platzky.config import LanguagesMapping	1✔
12	from spectree import Response, SpecTree	1✔
13
14	from goodmap.api_models import (	1✔
15	CSRFTokenResponse,
16	ErrorResponse,
17	LocationReportRequest,
18	LocationReportResponse,
19	SuccessResponse,
20	VersionResponse,
21	)
22	from goodmap.clustering import (	1✔
23	map_clustering_data_to_proper_lazy_loading_object,
24	match_clusters_uuids,
25	)
26	from goodmap.exceptions import LocationValidationError	1✔
27	from goodmap.formatter import prepare_pin	1✔
28	from goodmap.json_security import (	1✔
29	MAX_JSON_DEPTH_LOCATION,
30	JSONDepthError,
31	JSONSizeError,
32	safe_json_loads,
33	)
34
35	# SuperCluster configuration constants
36	MIN_ZOOM = 0	1✔
37	MAX_ZOOM = 16	1✔
38	CLUSTER_RADIUS = 200	1✔
39	CLUSTER_EXTENT = 512	1✔
40
41	# Error message constants
42	ERROR_INVALID_REQUEST_DATA = "Invalid request data"	1✔
43	ERROR_INVALID_LOCATION_DATA = "Invalid location data"	1✔
44	ERROR_LOCATION_NOT_FOUND = "Location not found"	1✔
45
46	logger = logging.getLogger(__name__)	1✔
47
48
49	def make_tuple_translation(keys_to_translate):	1✔
50	return [(x, gettext(x)) for x in keys_to_translate]	1✔
51
52
53	def get_or_none(data, *keys):	1✔
54	for key in keys:	1✔
55	if isinstance(data, dict):	1✔
56	data = data.get(key)	1✔
57	else:
58	return None	1✔
59	return data	1✔
60
61
62	def get_locations_from_request(database, request_args):	1✔
63	"""
64	Shared helper to fetch locations from database based on request arguments.
65
66	Args:
67	database: Database instance
68	request_args: Request arguments (flask.request.args)
69
70	Returns:
71	List of locations as basic_info dicts
72	"""
73	query_params = request_args.to_dict(flat=False)	1✔
74	all_locations = database.get_locations(query_params)	1✔
75	return [x.basic_info() for x in all_locations]	1✔
76
77
78	def core_pages(	1✔
79	database,
80	languages: LanguagesMapping,
81	notifier_function,
82	csrf_generator,
83	location_model,
84	photo_attachment_class: type[AttachmentProtocol],
85	feature_flags={},
86	) -> Blueprint:
87	core_api_blueprint = Blueprint("api", __name__, url_prefix="/api")	1✔
88
89	# Initialize Spectree for API documentation and validation
90	# Use simple naming strategy without hashes for cleaner schema names
91	from typing import Any, Type	1✔
92
93	def _clean_model_name(model: Type[Any]) -> str:	1✔
94	return model.__name__	1✔
95
96	spec = SpecTree(	1✔
97	"flask",
98	title="Goodmap API",
99	version="0.1",
100	path="doc",
101	annotations=True,
102	naming_strategy=_clean_model_name, # Use clean model names without hash
103	)
104
105	@core_api_blueprint.route("/suggest-new-point", methods=["POST"])	1✔
106	@spec.validate(resp=Response(HTTP_200=SuccessResponse, HTTP_400=ErrorResponse))	1✔
107	def suggest_new_point():	1✔
108	"""Suggest new location for review.
109
110	Accepts location data either as JSON or multipart/form-data.
111	All fields are validated using Pydantic location model.
112	"""
113	import json as json_lib	1✔
114
115	try:	1✔
116	# Initialize photo attachment (only populated for multipart/form-data)
117	photo_attachment = None	1✔
118
119	# Handle both multipart/form-data (with file uploads) and JSON
120	if request.content_type and request.content_type.startswith("multipart/form-data"):	1✔
121	# Parse form data dynamically
122	suggested_location = {}	1✔
123
124	for key in request.form:	1✔
125	value = request.form[key]	1✔
126	# Try to parse as JSON for complex types (arrays, objects, position)
127	try:	1✔
128	# SECURITY: Use safe_json_loads with strict depth limit
129	# MAX_JSON_DEPTH_LOCATION=1: arrays/objects of primitives only
130	suggested_location[key] = safe_json_loads(	1✔
131	value, max_depth=MAX_JSON_DEPTH_LOCATION
132	)
133	except (JSONDepthError, JSONSizeError) as e:	1✔
134	# Log security event and return 400
135	logger.warning(	1✔
136	f"JSON parsing blocked for security: {e}",
137	extra={"field": key, "value_size": len(value)},
138	)
139	return make_response(	1✔
140	jsonify(
141	{
142	"message": (
143	"Invalid request: JSON payload too complex or too large"
144	),
145	"error": str(e),
146	}
147	),
148	400,
149	)
150	except ValueError: # JSONDecodeError inherits from ValueError	1✔
151	# If not JSON, use as-is (simple string values)
152	suggested_location[key] = value	1✔
153
154	# Extract and validate photo attachment if present
155	photo_file = request.files.get("photo")	1✔
156	if photo_file and photo_file.filename:	1✔
NEW 157	photo_content = photo_file.read()	×
NEW 158	photo_mime = photo_file.content_type or "application/octet-stream"	×
159
160	# Validate using configured Attachment class
NEW 161	try:	×
NEW 162	photo_attachment = photo_attachment_class(	×
163	photo_file.filename, photo_content, photo_mime
164	)
NEW 165	except ValueError as e:	×
NEW 166	logger.warning("Rejected photo: %s", e)	×
NEW 167	return make_response(jsonify({"message": str(e)}), 400)	×
168	else:
169	# Parse JSON data with security checks (depth/size protection)
170	raw_data = request.get_data(as_text=True)	1✔
171	if not raw_data:	1✔
172	logger.warning("Empty JSON body in suggest endpoint")	1✔
173	return make_response(jsonify({"message": ERROR_INVALID_REQUEST_DATA}), 400)	1✔
174	try:	1✔
175	suggested_location = safe_json_loads(	1✔
176	raw_data, max_depth=MAX_JSON_DEPTH_LOCATION
177	)
178	except (JSONDepthError, JSONSizeError) as e:	1✔
179	logger.warning(	1✔
180	f"JSON parsing blocked for security: {e}",
181	extra={"value_size": len(raw_data)},
182	)
183	return make_response(	1✔
184	jsonify(
185	{
186	"message": (
187	"Invalid request: JSON payload too complex or too large"
188	),
189	"error": str(e),
190	}
191	),
192	400,
193	)
194	except ValueError:	1✔
195	logger.warning("Invalid JSON in suggest endpoint")	1✔
196	return make_response(jsonify({"message": ERROR_INVALID_REQUEST_DATA}), 400)	1✔
197	if suggested_location is None:	1✔
198	logger.warning("Null JSON value in suggest endpoint")	1✔
199	return make_response(jsonify({"message": ERROR_INVALID_REQUEST_DATA}), 400)	1✔
200
201	suggested_location.update({"uuid": str(uuid.uuid4())})	1✔
202	location = location_model.model_validate(suggested_location)	1✔
203	database.add_suggestion(location.model_dump())	1✔
204	message = gettext("A new location has been suggested with details")	1✔
205	notifier_message = f"{message}: {json_lib.dumps(suggested_location, indent=2)}"	1✔
206	attachments = [photo_attachment] if photo_attachment else None	1✔
207	notifier_function(notifier_message, attachments=attachments)	1✔
208	except LocationValidationError as e:	1✔
209	# NOTE: validation_errors includes input values from the location model fields:
210	# - Core fields: position (lat/long), uuid, remark
211	# - Dynamic fields: categories and obligatory_fields configured per deployment
212	# These are geographic/categorical data, NOT PII (no email, phone, names of people).
213	# Safe to log for debugging. If PII fields are ever added to the location model,
214	# strip 'input' from validation_errors before logging.
215	logger.warning(	1✔
216	"Location validation failed in suggest endpoint: %s",
217	e.validation_errors,
218	extra={"errors": e.validation_errors},
219	)
220	return make_response(jsonify({"message": ERROR_INVALID_LOCATION_DATA}), 400)	1✔
221	except Exception:	1✔
222	logger.error("Error in suggest location endpoint", exc_info=True)	1✔
223	return make_response(	1✔
224	jsonify({"message": "An error occurred while processing your suggestion"}), 500
225	)
226	return make_response(jsonify({"message": "Location suggested"}), 200)	1✔
227
228	@core_api_blueprint.route("/report-location", methods=["POST"])	1✔
229	@spec.validate(	1✔
230	json=LocationReportRequest,
231	resp=Response(HTTP_200=LocationReportResponse, HTTP_400=ErrorResponse),
232	)
233	def report_location():	1✔
234	"""Report a problem with a location.
235
236	Allows users to report issues with existing locations,
237	such as incorrect information or closed establishments.
238	"""
239	try:	1✔
240	location_report = request.get_json()	1✔
241	report = {	1✔
242	"uuid": str(uuid.uuid4()),
243	"location_id": location_report["id"],
244	"description": location_report["description"],
245	"status": "pending",
246	"priority": "medium",
247	}
248	database.add_report(report)	1✔
249	message = (	1✔
250	f"A location has been reported: '{location_report['id']}' "
251	f"with problem: {location_report['description']}"
252	)
253	notifier_function(message)	1✔
254	except Exception:	1✔
255	logger.error("Error in report location endpoint", exc_info=True)	1✔
256	error_message = gettext("Error sending notification")	1✔
257	return make_response(jsonify({"message": error_message}), 500)	1✔
258	return make_response(jsonify({"message": gettext("Location reported")}), 200)	1✔
259
260	@core_api_blueprint.route("/locations", methods=["GET"])	1✔
261	@spec.validate()	1✔
262	def get_locations():	1✔
263	"""Get list of locations with basic info.
264
265	Returns locations filtered by query parameters,
266	showing only uuid, position, and remark flag.
267	"""
268	locations = get_locations_from_request(database, request.args)	1✔
269	return jsonify(locations)	1✔
270
271	@core_api_blueprint.route("/locations-clustered", methods=["GET"])	1✔
272	@spec.validate(resp=Response(HTTP_400=ErrorResponse))	1✔
273	def get_locations_clustered():	1✔
274	"""Get clustered locations for map display.
275
276	Returns locations grouped into clusters based on zoom level,
277	optimized for rendering on interactive maps.
278	"""
279	try:	1✔
280	query_params = request.args.to_dict(flat=False)	1✔
281	zoom = int(query_params.get("zoom", [7])[0])	1✔
282
283	# Validate zoom level (aligned with SuperCluster min_zoom/max_zoom)
284	if not MIN_ZOOM <= zoom <= MAX_ZOOM:	1✔
285	return make_response(	1✔
286	jsonify({"message": f"Zoom must be between {MIN_ZOOM} and {MAX_ZOOM}"}),
287	400,
288	)
289
290	points = get_locations_from_request(database, request.args)	1✔
291	if not points:	1✔
292	return jsonify([])	1✔
293
294	points_numpy = numpy.array(	1✔
295	[(point["position"][0], point["position"][1]) for point in points]
296	)
297
298	index = pysupercluster.SuperCluster(	1✔
299	points_numpy,
300	min_zoom=MIN_ZOOM,
301	max_zoom=MAX_ZOOM,
302	radius=CLUSTER_RADIUS,
303	extent=CLUSTER_EXTENT,
304	)
305
306	clusters = index.getClusters(	1✔
307	top_left=(-180.0, 90.0),
308	bottom_right=(180.0, -90.0),
309	zoom=zoom,
310	)
311	clusters = match_clusters_uuids(points, clusters)	1✔
312
313	return jsonify(map_clustering_data_to_proper_lazy_loading_object(clusters))	1✔
314	except ValueError as e:	1✔
315	logger.warning("Invalid parameter in clustering request: %s", e)	1✔
316	return make_response(jsonify({"message": "Invalid parameters provided"}), 400)	1✔
317	except Exception as e:	1✔
318	logger.error("Clustering operation failed: %s", e, exc_info=True)	1✔
319	return make_response(jsonify({"message": "An error occurred during clustering"}), 500)	1✔
320
321	@core_api_blueprint.route("/location/<location_id>", methods=["GET"])	1✔
322	@spec.validate(resp=Response(HTTP_404=ErrorResponse))	1✔
323	def get_location(location_id):	1✔
324	"""Get detailed information for a single location.
325
326	Returns full location data including all custom fields,
327	formatted for display in the location details view.
328	"""
329	location = database.get_location(location_id)	1✔
330	if location is None:	1✔
331	logger.info(ERROR_LOCATION_NOT_FOUND, extra={"uuid": location_id})	1✔
332	return make_response(jsonify({"message": ERROR_LOCATION_NOT_FOUND}), 404)	1✔
333
334	visible_data = database.get_visible_data()	1✔
335	meta_data = database.get_meta_data()	1✔
336
337	formatted_data = prepare_pin(location.model_dump(), visible_data, meta_data)	1✔
338	return jsonify(formatted_data)	1✔
339
340	@core_api_blueprint.route("/version", methods=["GET"])	1✔
341	@spec.validate(resp=Response(HTTP_200=VersionResponse))	1✔
342	def get_version():	1✔
343	"""Get backend version information.
344
345	Returns the current version of the Goodmap backend.
346	"""
347	version_info = {"backend": importlib.metadata.version("goodmap")}	1✔
348	return jsonify(version_info)	1✔
349
350	@core_api_blueprint.route("/generate-csrf-token", methods=["GET"])	1✔
351	@spec.validate(resp=Response(HTTP_200=CSRFTokenResponse))	1✔
352	@deprecation.deprecated(	1✔
353	deprecated_in="1.1.8",
354	details="This endpoint for explicit CSRF token generation is deprecated. "
355	"CSRF protection remains active in the application.",
356	)
357	def generate_csrf_token():	1✔
358	"""Generate CSRF token (DEPRECATED).
359
360	This endpoint is deprecated and maintained only for backward compatibility.
361	CSRF protection remains active in the application.
362	"""
363	csrf_token = csrf_generator()	1✔
364	return {"csrf_token": csrf_token}	1✔
365
366	@core_api_blueprint.route("/categories", methods=["GET"])	1✔
367	@spec.validate()	1✔
368	def get_categories():	1✔
369	"""Get all available location categories.
370
371	Returns list of categories with optional help text
372	if CATEGORIES_HELP feature flag is enabled.
373	"""
374	raw_categories = database.get_categories()	1✔
375	categories = make_tuple_translation(raw_categories)	1✔
376
377	if not feature_flags.get("CATEGORIES_HELP", False):	1✔
378	return jsonify(categories)	1✔
379	else:
380	category_data = database.get_category_data()	1✔
381	categories_help = category_data.get("categories_help")	1✔
382	proper_categories_help = []	1✔
383	if categories_help is not None:	1✔
384	for option in categories_help:	1✔
385	proper_categories_help.append({option: gettext(f"categories_help_{option}")})	1✔
386
387	return jsonify({"categories": categories, "categories_help": proper_categories_help})	1✔
388
389	@core_api_blueprint.route("/categories-full", methods=["GET"])	1✔
390	@spec.validate()	1✔
391	def get_categories_full():	1✔
392	"""Get all categories with their subcategory options in a single request.
393
394	Returns combined category data to reduce API calls for filter panel loading.
395	This endpoint eliminates the need for multiple sequential requests.
396	"""
397	categories_data = database.get_category_data()	1✔
398	result = []	1✔
399
400	categories_options_help = categories_data.get("categories_options_help", {})	1✔
401
402	for key, options in categories_data["categories"].items():	1✔
403	category_entry = {	1✔
404	"key": key,
405	"name": gettext(key),
406	"options": make_tuple_translation(options),
407	}
408
409	if feature_flags.get("CATEGORIES_HELP", False):	1✔
410	option_help_list = categories_options_help.get(key, [])	1✔
411	proper_options_help = []	1✔
412	for option in option_help_list:	1✔
413	proper_options_help.append(	1✔
414	{option: gettext(f"categories_options_help_{option}")}
415	)
416	category_entry["options_help"] = proper_options_help	1✔
417
418	result.append(category_entry)	1✔
419
420	response = {"categories": result}	1✔
421
422	if feature_flags.get("CATEGORIES_HELP", False):	1✔
423	categories_help = categories_data.get("categories_help", [])	1✔
424	proper_categories_help = []	1✔
425	for option in categories_help:	1✔
426	proper_categories_help.append({option: gettext(f"categories_help_{option}")})	1✔
427	response["categories_help"] = proper_categories_help	1✔
428
429	return jsonify(response)	1✔
430
431	@core_api_blueprint.route("/languages", methods=["GET"])	1✔
432	@spec.validate()	1✔
433	def get_languages():	1✔
434	"""Get all available interface languages.
435
436	Returns list of supported languages for the application.
437	"""
438	return jsonify(languages)	1✔
439
440	@core_api_blueprint.route("/category/<category_type>", methods=["GET"])	1✔
441	@spec.validate()	1✔
442	def get_category_types(category_type):	1✔
443	"""Get all available options for a specific category.
444
445	Returns list of category options with optional help text
446	if CATEGORIES_HELP feature flag is enabled.
447	"""
448	category_data = database.get_category_data(category_type)	1✔
449	local_data = make_tuple_translation(category_data["categories"][category_type])	1✔
450
451	categories_options_help = get_or_none(	1✔
452	category_data, "categories_options_help", category_type
453	)
454	proper_categories_options_help = []	1✔
455	if categories_options_help is not None:	1✔
456	for option in categories_options_help:	1✔
457	proper_categories_options_help.append(	1✔
458	{option: gettext(f"categories_options_help_{option}")}
459	)
460	if not feature_flags.get("CATEGORIES_HELP", False):	1✔
461	return jsonify(local_data)	1✔
462	else:
463	return jsonify(	1✔
464	{
465	"categories_options": local_data,
466	"categories_options_help": proper_categories_options_help,
467	}
468	)
469
470	# Register Spectree with blueprint after all routes are defined
471	spec.register(core_api_blueprint)	1✔
472
473	@core_api_blueprint.route("/doc")	1✔
474	def api_doc_index():	1✔
475	"""Return links to available API documentation formats."""
476	html = """<!DOCTYPE html>	1✔
477	<html><head><title>API Documentation</title></head>
478	<body>
479	<h1>API Documentation</h1>
480	<ul>
481	<li><a href="/api/doc/swagger/">Swagger UI</a></li>
482	<li><a href="/api/doc/redoc/">ReDoc</a></li>
483	<li><a href="/api/doc/openapi.json">OpenAPI JSON</a></li>
484	</ul>
485	</body></html>"""
486	return html, 200, {"Content-Type": "text/html"}	1✔
487
488	return core_api_blueprint	1✔

Problematy / goodmap / 21304342528

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