Enable getting only a partial AirTable table by view and/or formula.

Author: pcorpet

airtable/airtable.py

@@ -63,7 +63,9 @@ def __request(self, method, url, params=None, payload=None):
                 'error': dict(code=r.status_code, message=message)
             }
 
-    def get(self, table_name, record_id=None, limit=0, offset=None):
+    def get(
+            self, table_name, record_id=None, limit=0, offset=None,
+            filter_by_formula=None, view=None):
         params = {}
         if check_string(record_id):
             url = os.path.join(table_name, record_id)
@@ -73,9 +75,14 @@ def get(self, table_name, record_id=None, limit=0, offset=None):
                 params.update({'pageSize': limit})
             if offset and check_string(offset):
                 params.update({'offset': offset})
+            if filter_by_formula is not None:
+                params.update({'filterByFormula': filter_by_formula})
+            if view is not None:
+                params.update({'view': view})
         return self.__request('GET', url, params)
 
-    def iterate(self, table_name, batch_size=0):
+    def iterate(
+            self, table_name, batch_size=0, filter_by_formula=None, view=None):
         """Iterate over all records of a table.
 
         Args:
@@ -84,13 +91,23 @@ def iterate(self, table_name, batch_size=0):
                 (0) is using the default of the API which is (as of 2016-09)
                 100. Note that the API does not allow more than that (but
                 allow for less).
+            filter_by_formula: a formula used to filter records. The formula
+                will be evaluated for each record, and if the result is not 0,
+                false, "", NaN, [], or #Error! the record will be included in
+                the response. If combined with view, only records in that view
+                which satisfy the formula will be returned.
+            view: the name or ID of a view in the table. If set, only the
+                records in that view will be returned. The records will be
+                sorted according to the order of the view.
         Yields:
             A dict for each record containing at least three fields: "id",
             "createdTime" and "fields".
         """
         offset = None
         while True:
-            response = self.get(table_name, limit=batch_size, offset=offset)
+            response = self.get(
+                table_name, limit=batch_size, offset=offset,
+                filter_by_formula=filter_by_formula, view=view)
             for record in response.pop('records'):
                 yield record
             if 'offset' in response:

test_airtable.py

@@ -128,6 +128,32 @@ def test_get_not_found(self, mock_request):
         r = self.airtable.get(FAKE_TABLE_NAME, '123')
         self.assertEqual(r['error']['code'], 404)
 
+    @mock.patch.object(requests, 'request')
+    def test_get_view(self, mock_request):
+        mock_response = mock.MagicMock()
+        mock_response.status_code = 200
+        mock_response.json.return_value = {'records': []}
+        mock_request.return_value = mock_response
+
+        r = self.airtable.get(FAKE_TABLE_NAME, view='My view')
+        mock_request.assert_called_once_with(
+            'GET', 'https://api.airtable.com/v0/app12345/TableName',
+            data=None, headers={'Authorization': 'Bearer fake_api_key'},
+            params={'view': 'My view'})
+
+    @mock.patch.object(requests, 'request')
+    def test_get_filter_by_formula(self, mock_request):
+        mock_response = mock.MagicMock()
+        mock_response.status_code = 200
+        mock_response.json.return_value = {'records': []}
+        mock_request.return_value = mock_response
+
+        r = self.airtable.get(FAKE_TABLE_NAME, filter_by_formula="{title} = ''")
+        mock_request.assert_called_once_with(
+            'GET', 'https://api.airtable.com/v0/app12345/TableName',
+            data=None, headers={'Authorization': 'Bearer fake_api_key'},
+            params={'filterByFormula': "{title} = ''"})
+
     def test_invalid_get(self):
         with self.assertRaises(airtable.IsNotString):
             self.airtable.get(FAKE_TABLE_NAME, 123)