Address reviews

- create schema map in controller for each request
- put back test for invalid order param with 200
- unit test for get_schema_fields_map helper

Author: harminius

server/mergin/sync/public_api_v2_controller.py

@@ -437,8 +437,9 @@ def list_workspace_projects(workspace_id, page, per_page, order_params=None, q=N
         projects = projects.filter(Project.name.ilike(f"%{q}%"))
 
     if order_params:
+        schema_map = get_schema_fields_map(ProjectSchemaV2)
         order_by_params = parse_order_params(
-            Project, order_params, field_map=ProjectSchemaV2.field_map
+            Project, order_params, field_map=schema_map
         )
         projects = projects.order_by(*order_by_params)
 

server/mergin/sync/schemas_v2.py

@@ -11,7 +11,6 @@
     Project,
     ProjectVersion,
 )
-from ..utils import get_schema_fields_map
 
 
 class ProjectSchema(ma.SQLAlchemyAutoSchema):
@@ -47,6 +46,3 @@ class Meta:
             "workspace",
             "role",
         )
-
-
-ProjectSchema.field_map = get_schema_fields_map(ProjectSchema)

server/mergin/tests/test_public_api_v2.py

@@ -651,6 +651,10 @@ def test_list_workspace_projects(client):
     resp_data = json.loads(response.data)
     assert resp_data["projects"][0]["name"] == project_name
 
+    # invalid order param
+    response = client.get(url + f"?page=1&per_page=10&order_params=invalid DESC")
+    assert response.status_code == 200
+
     # no permissions to workspace
     user2 = add_user("user", "password")
     login(client, user2.username, "password")

server/mergin/tests/test_utils.py

@@ -7,14 +7,15 @@
 import json
 import pytest
 from flask import url_for, current_app
+from marshmallow import Schema, fields
 from sqlalchemy import desc
 import os
 from unittest.mock import patch
 from pathvalidate import sanitize_filename
 from pygeodiff import GeoDiff
 from pathlib import PureWindowsPath
 
-from ..utils import save_diagnostic_log_file
+from ..utils import save_diagnostic_log_file, get_schema_fields_map
 
 from ..sync.utils import (
     is_reserved_word,
@@ -297,3 +298,27 @@ def test_save_diagnostic_log_file(client, app):
         with open(saved_file_path, "r") as f:
             content = f.read()
             assert content == body.decode("utf-8")
+
+
+def test_get_schema_fields_map():
+    """Test that schema map correctly resolves DB attributes, keeps all fields, and ignores virtual fields."""
+
+    # dummy schema for testing
+    class TestSchema(Schema):
+        # standard field -> map 'name': 'name'
+        name = fields.String()
+        # aliased field -> map 'size': 'disk_usage
+        size = fields.Integer(attribute="disk_usage")
+        # virtual fields -> skip
+        version = fields.Function(lambda obj: "v1")
+        role = fields.Method("get_role")
+        # excluded field - set to None in schema inheritance -> skip
+        hidden_field = None
+
+    schema_map = get_schema_fields_map(TestSchema)
+
+    expected_map = {
+        "name": "name",
+        "size": "disk_usage",
+    }
+    assert schema_map == expected_map

server/mergin/utils.py

@@ -8,14 +8,13 @@
 from datetime import datetime, timedelta, timezone
 from enum import Enum
 import os
-from flask import current_app, abort
+from flask import current_app
 from flask_sqlalchemy import Model
-from marshmallow import Schema
+from marshmallow import Schema, fields
 from pathvalidate import sanitize_filename
 from sqlalchemy import Column, JSON
 from sqlalchemy.sql.elements import UnaryExpression
-from typing import Optional
-
+from typing import Optional, Type
 
 OrderParam = namedtuple("OrderParam", "name direction")
 
@@ -151,14 +150,23 @@ def save_diagnostic_log_file(app: str, username: str, body: bytes) -> str:
     return file_name
 
 
-def get_schema_fields_map(schema: Schema) -> dict:
+def get_schema_fields_map(schema: Type[Schema]) -> dict:
     """
     Creates a mapping of schema field names to corresponding DB columns.
     This allows sorting by the API field name (e.g. 'size') while
     actually sorting by the database column (e.g. 'disk_usage').
     """
     mapping = {}
     for name, field in schema._declared_fields.items():
-        if field and field.attribute:
+        # some fields could have been overridden with None to be excluded
+        if not field:
+            continue
+        # skip virtual fields as DB cannot sort by them
+        if isinstance(field, (fields.Function, fields.Method)):
+            continue
+        if field.attribute:
             mapping[name] = field.attribute
+        # keep the map complete
+        else:
+            mapping[name] = name
     return mapping