From a15143e6a3e264fcd5f61341967d0190412b2edd Mon Sep 17 00:00:00 2001
From: Benjamin Bertrand <benjamin.bertrand@esss.se>
Date: Thu, 7 Sep 2017 15:23:40 +0200
Subject: [PATCH] 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 | 3 ++-
app/utils.py | 17 +++++++++++++++++
tests/functional/test_api.py | 37 ++++++++++++++++++++++++++++++++++++
3 files changed, 56 insertions(+), 1 deletion(-)
diff --git a/app/api/main.py b/app/api/main.py
index 316a011..e2c3f31 100644
--- a/app/api/main.py
+++ b/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)
diff --git a/app/utils.py b/app/utils.py
index 8365bcf..a369fbf 100644
--- a/app/utils.py
+++ b/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
diff --git a/tests/functional/test_api.py b/tests/functional/test_api.py
index 088eece..396dc55 100644
--- a/tests/functional/test_api.py
+++ b/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)
--
GitLab