Set limit in sqlalchemy backend between 1 and 10000

CHANGES.md

@@ -11,6 +11,7 @@
 ### Fixed
 
 * Pin FastAPI to 0.67 to avoid issues with rendering OpenAPI documentation ([#246](https://github.com/stac-utils/stac-fastapi/pull/246))
+* Restrict `limit` parameter in sqlalchemy backend to between 1 and 10,000. ([#251](https://github.com/stac-utils/stac-fastapi/pull/251))
 
 ## [2.1.0]
 

stac_fastapi/sqlalchemy/stac_fastapi/sqlalchemy/types/search.py

@@ -11,7 +11,7 @@
 from typing import Any, Callable, Dict, List, Optional, Set, Union
 
 import sqlalchemy as sa
-from pydantic import Field, ValidationError, root_validator
+from pydantic import Field, ValidationError, conint, root_validator
 from pydantic.error_wrappers import ErrorWrapper
 from stac_pydantic.api import Search
 from stac_pydantic.api.extensions.fields import FieldsExtension as FieldsBase
@@ -145,6 +145,7 @@ class SQLAlchemySTACSearch(Search):
     # Override query extension with supported operators
     query: Optional[Dict[Queryables, Dict[Operator, Any]]]
     token: Optional[str] = None
+    limit: Optional[conint(ge=0, le=10000)] = 10
 
     @root_validator(pre=True)
     def validate_query_fields(cls, values: Dict) -> Dict:

stac_fastapi/sqlalchemy/tests/api/test_api.py

@@ -91,6 +91,39 @@ def test_app_query_extension(load_test_data, app_client, postgres_transactions):
     assert len(resp_json["features"]) == 0
 
 
+def test_app_query_extension_limit_lt0(
+    load_test_data, app_client, postgres_transactions
+):
+    item = load_test_data("test_item.json")
+    postgres_transactions.create_item(item, request=MockStarletteRequest)
+
+    params = {"limit": -1}
+    resp = app_client.post("/search", json=params)
+    assert resp.status_code == 400
+
+
+def test_app_query_extension_limit_gt10000(
+    load_test_data, app_client, postgres_transactions
+):
+    item = load_test_data("test_item.json")
+    postgres_transactions.create_item(item, request=MockStarletteRequest)
+
+    params = {"limit": 10001}
+    resp = app_client.post("/search", json=params)
+    assert resp.status_code == 400
+
+
+def test_app_query_extension_limit_10000(
+    load_test_data, app_client, postgres_transactions
+):
+    item = load_test_data("test_item.json")
+    postgres_transactions.create_item(item, request=MockStarletteRequest)
+
+    params = {"limit": 10000}
+    resp = app_client.post("/search", json=params)
+    assert resp.status_code == 200
+
+
 def test_app_sort_extension(load_test_data, app_client, postgres_transactions):
     first_item = load_test_data("test_item.json")
     item_date = datetime.strptime(