Add basic filtering to retrieve items via the API

Can use:

  -  /api/items?serial_number=123456
  -  /api/items?location_id=1

app/api/main.py

@@ -89,7 +89,8 @@ def login():
 @jwt_required
 def get_items():
     # TODO: add pagination
-    items = Item.query.order_by(Item._created)
+    query = utils.get_query(Item.query, request.args)
+    items = query.order_by(Item._created)
     data = [item.to_dict() for item in items]
     return jsonify(data)
 

app/utils.py

@@ -12,6 +12,7 @@ This module implements utility functions.
 import base64
 import datetime
 import io
+import sqlalchemy as sa
 
 
 class InventoryError(Exception):
@@ -103,3 +104,19 @@ def get_choices(iterable, allow_blank=False, allow_null=False):
         choices.append(('null', 'not set'))
     choices.extend([(val, val) for val in iterable])
     return choices
+
+
+def get_query(query, args):
+    """Retrieve the query from the arguments
+
+    :param query: sqlalchemy base query
+    :param MultiDict args: args from a request
+    :returns: query filtered by the arguments
+    """
+    if args:
+        kwargs = args.to_dict()
+        try:
+            query = query.filter_by(**kwargs)
+        except (sa.exc.InvalidRequestError, AttributeError) as e:
+            raise InventoryError('Invalid query arguments', status_code=422)
+    return query

tests/functional/test_api.py

@@ -357,3 +357,40 @@ def test_patch_item_parent(client, session, user_token):
     assert response.json['manufacturer'] is None
     response = get(client, f'/api/items/{item3.ics_id}', token=user_token)
     assert response.json['manufacturer'] == 'Dell'
+
+
+def test_get_items(client, session, readonly_token):
+    # Create some items
+    session.add(models.Manufacturer(name='Dell'))
+    session.add(models.Status(name='Stock'))
+    session.add(models.Status(name='In service'))
+    session.add(models.Location(name='ESS'))
+    session.add(models.Location(name='ICS lab'))
+    item1 = models.Item(serial_number='123456', ics_id='AAA001', location='ESS', status='In service',
+                        manufacturer='Dell')
+    item2 = models.Item(serial_number='234567', ics_id='AAA002', status='Stock')
+    item3 = models.Item(serial_number='345678', ics_id='AAA003', status='Stock', manufacturer='Dell')
+    for item in (item1, item2, item3):
+        session.add(item)
+    session.commit()
+
+    response = get(client, '/api/items', token=readonly_token)
+    assert response.status_code == 200
+    assert len(response.json) == 3
+    check_items(response, (item1.to_dict(), item2.to_dict(), item3.to_dict()))
+
+    # test filtering
+    response = get(client, '/api/items?serial_number=234567', token=readonly_token)
+    assert response.status_code == 200
+    assert len(response.json) == 1
+    check_items(response, (item2.to_dict(),))
+    # filtering on location_id works but not location (might want to change that)
+    response = get(client, f'/api/items?location_id={item1.location_id}', token=readonly_token)
+    assert response.status_code == 200
+    assert len(response.json) == 1
+    check_items(response, (item1.to_dict(),))
+    response = get(client, '/api/items?location=ESS', token=readonly_token)
+    check_response_message(response, 'Invalid query arguments', 422)
+    # using an unknown key raises a 422
+    response = get(client, '/api/items?foo=bar', token=readonly_token)
+    check_response_message(response, 'Invalid query arguments', 422)