filter_function parameter for ResourceFeeder

openprocurement_client/exceptions.py

@@ -119,6 +119,10 @@ class IdNotFound(ResourceError):
     pass
 
 
+class NotAFunction(TypeError):
+    pass
+
+
 http_exceptions_dict = {
     405: MethodNotAllowed,
     409: Conflict,

openprocurement_client/sync.py

@@ -2,6 +2,7 @@
 monkey.patch_all()
 
 import logging
+import types
 from .client import TendersClientSync
 from datetime import datetime
 from gevent import spawn, sleep, idle
@@ -10,7 +11,8 @@
 from openprocurement_client.exceptions import (
     RequestFailed,
     PreconditionFailed,
-    ResourceNotFound
+    ResourceNotFound,
+    NotAFunction
 )
 
 # Statuses
@@ -89,7 +91,7 @@ class ResourceFeeder(object):
 
     def __init__(self, host=DEFAULT_API_HOST, version=DEFAULT_API_VERSION,
                  key=DEFAULT_API_KEY, resource='tenders', extra_params=DEFAULT_API_EXTRA_PARAMS,
-                 retrievers_params=DEFAULT_RETRIEVERS_PARAMS, adaptive=False):
+                 retrievers_params=DEFAULT_RETRIEVERS_PARAMS, adaptive=False, filter_function=None):
         super(ResourceFeeder, self).__init__()
         self.host = host
         self.version = version
@@ -103,6 +105,19 @@ def __init__(self, host=DEFAULT_API_HOST, version=DEFAULT_API_VERSION,
         self.forward_info = {}
         self.backward_info = {}
 
+        if filter_function:
+            if isinstance(filter_function, (types.FunctionType,
+                                            types.LambdaType,
+                                            types.BuiltinFunctionType,
+                                            types.MethodType,
+                                            types.BuiltinMethodType,
+                                            types.UnboundMethodType
+                                            )):
+                self.filter_function = filter_function
+                self.handle_response_data = self.filtered_handle_response_data
+            else:
+                raise NotAFunction("supplied object is not a function")
+
     def init_api_clients(self):
         self.backward_params = {'descending': True, 'feed': 'changes'}
         self.backward_params.update(self.extra_params)
@@ -121,6 +136,11 @@ def handle_response_data(self, data):
             # self.idle()
             self.queue.put(tender)
 
+    def filtered_handle_response_data(self, data):
+        for tender in data:
+            if self.filter_function(tender):
+                self.queue.put(tender)
+
     def start_sync(self):
         # self.init_api_clients()
 

openprocurement_client/tests/tests_sync.py

@@ -3,9 +3,10 @@
 
 from openprocurement_client.client import TendersClientSync
 from openprocurement_client.exceptions import (
+    NotAFunction,
     RequestFailed,
     PreconditionFailed,
-    ResourceNotFound
+    ResourceNotFound,
 )
 from openprocurement_client.sync import (
     get_response,
@@ -26,6 +27,10 @@
 import unittest
 
 
+def filter_function(item):
+        return not set(item['id']).difference('0123456789')
+
+
 class AlmostAlwaysTrue(object):
     def __init__(self, total_iterations=1):
         self.total_iterations = total_iterations
@@ -193,6 +198,27 @@ def test_handle_response_data(self):
         self.assertIn('tender2', list(self.resource_feeder.queue.queue))
         self.assertNotIn('tender3', list(self.resource_feeder.queue.queue))
 
+    def test_filtered_handle_response_data(self):
+        tender1, tender2, tender3 = {'id': '18532'}, {'id': 'non_numeric_id'}, {'id': '9999'}
+        self.resource_feeder = ResourceFeeder(filter_function=filter_function)
+        self.resource_feeder.handle_response_data([tender1, tender2, tender3])
+        self.assertIn(tender1, list(self.resource_feeder.queue.queue))
+        self.assertIn(tender3, list(self.resource_feeder.queue.queue))
+        self.assertNotIn(tender2, list(self.resource_feeder.queue.queue))
+
+    def test_filtered_handle_response_data_lambda(self):
+        tender1, tender2, tender3 = {'id': '18532'}, {'id': 'non_numeric_id'}, {'id': '9999'}
+        self.resource_feeder = ResourceFeeder(filter_function=lambda x: not set(x['id']).difference('0123456789'))
+        self.resource_feeder.handle_response_data([tender1, tender2, tender3])
+        self.assertIn(tender1, list(self.resource_feeder.queue.queue))
+        self.assertIn(tender3, list(self.resource_feeder.queue.queue))
+        self.assertNotIn(tender2, list(self.resource_feeder.queue.queue))
+
+    def test_filtered_handle_response_data_not_a_function(self):
+        with self.assertRaises(NotAFunction) as e:
+            self.resource_feeder = ResourceFeeder(filter_function='12')
+        self.assertEqual(e.exception.message, "supplied object is not a function")
+
     @mock.patch('openprocurement_client.client.TendersClientSync.sync_tenders')
     @mock.patch('openprocurement_client.sync.spawn')
     def test_start_sync(self, mock_spawn, mock_sync_tenders):