21284115567

Committed 23 Jan 2026 11:10AM UTC coverage: 97.152%. First build

Build # 21284115567

Build Type

Pull #324

github

Committed by

web-flow

Commit Message

Merge 2403aa344 into 0778a150c

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

Run Details

27 of 62 new or added lines in 2 files covered. (43.55%)

1501 of 1545 relevant lines covered (97.15%)

0.97 hits per line

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

90.12

/goodmap/core_api.py

import importlib.metadata
import io
import logging
import uuid

import deprecation
import numpy
import pysupercluster
from PIL import Image
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"

# Photo compression settings
PHOTO_COMPRESSION_THRESHOLD = 2 * 1024 * 1024  # 2MB
PHOTO_MAX_DIMENSION = 1920  # Max width or height
PHOTO_JPEG_QUALITY = 85

logger = logging.getLogger(__name__)


def compress_photo(
    content: bytes, mime_type: str, filename: str
) -> tuple[bytes, str, str]:
    """Compress photo if it exceeds size threshold.

    Converts PNG to JPEG and resizes if needed to reduce file size.

    Args:
        content: Original image bytes
        mime_type: Original MIME type
        filename: Original filename

    Returns:
        Tuple of (compressed_content, new_mime_type, new_filename)
    """
    # Only compress if above threshold
    if len(content) <= PHOTO_COMPRESSION_THRESHOLD:
        return content, mime_type, filename

    try:
        img = Image.open(io.BytesIO(content))

        # Convert RGBA/P to RGB for JPEG (no transparency support)
        if img.mode in ("RGBA", "P"):
            img = img.convert("RGB")

        # Resize if larger than max dimension
        if max(img.size) > PHOTO_MAX_DIMENSION:
            img.thumbnail((PHOTO_MAX_DIMENSION, PHOTO_MAX_DIMENSION), Image.LANCZOS)

        # Save as JPEG
        output = io.BytesIO()
        img.save(output, format="JPEG", quality=PHOTO_JPEG_QUALITY, optimize=True)
        compressed = output.getvalue()

        # Update filename extension
        new_filename = filename.rsplit(".", 1)[0] + ".jpg"

        logger.info(
            "Compressed photo %s: %d bytes -> %d bytes",
            filename,
            len(content),
            len(compressed),
        )

        return compressed, "image/jpeg", new_filename

    except Exception as e:
        logger.warning("Failed to compress photo %s: %s", filename, e)
        return content, mime_type, filename


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

Problematy / goodmap / 21284115567

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