TDL-13967: Add ProfitAndLoss report stream

tap_quickbooks/schemas/profit_loss_report.json

@@ -0,0 +1,28 @@
+{
+    "type": [
+      "null",
+      "object"
+    ],
+    "properties": {
+      "ReportDate": {
+        "type": [
+          "null",
+          "string"
+        ],
+        "format": "date-time"
+      },
+      "AccountingMethod": {
+        "type": [
+          "null",
+          "string"
+        ]
+      },
+      "Details": {
+        "type": [
+          "null",
+          "object"
+        ],
+        "properties": {}
+      }
+    }
+}

tap_quickbooks/streams.py

@@ -1,6 +1,11 @@
 import tap_quickbooks.query_builder as query_builder
 
 import singer
+from singer import utils
+from singer.utils import strptime_to_utc, strftime
+from datetime import timedelta
+
+DATE_WINDOW_SIZE = 30
 
 class Stream:
     endpoint = '/v3/company/{realm_id}/query'
@@ -193,6 +198,134 @@ class Vendors(Stream):
     table_name = 'Vendor'
     additional_where = "Active IN (true, false)"
 
+class ReportStream(Stream):
+    parsed_metadata = {
+        'dates': [],
+        'data': []
+    }
+    key_properties = []
+    replication_method = 'INCREMENTAL'
+    # replication keys is ReportDate, manually created from data
+    replication_keys = ['ReportDate']
+
+    def sync(self):
+
+        start_dttm_str = singer.get_bookmark(self.state, self.stream_name, 'LastUpdatedTime', self.config.get('start_date'))
+        start_dttm = strptime_to_utc(start_dttm_str)
+
+        end_dttm = start_dttm + timedelta(days=DATE_WINDOW_SIZE)
+        now_dttm = utils.now()
+
+        if end_dttm > now_dttm:
+            end_dttm = now_dttm
+        params = {
+            'summarize_column_by': 'Days'
+        }
+
+        while start_dttm < now_dttm:
+            self.parsed_metadata = {
+                'dates': [],
+                'data': []
+            }
+
+            start_tm_str = strftime(start_dttm)[0:10]
+            end_tm_str = strftime(end_dttm)[0:10]
+
+            params["start_date"] = start_tm_str
+            params["end_date"] = end_tm_str
+
+            resp = self.client.get(self.endpoint, params=params)
+            self.parse_report_metadata(resp)
+
+            reports = self.day_wise_reports()
+            if reports:
+                for report in reports:
+                    yield report
+                self.state = singer.write_bookmark(self.state, self.stream_name, 'LastUpdatedTime', strptime_to_utc(report.get('ReportDate')).isoformat())
+                singer.write_state(self.state)
+
+            start_dttm = end_dttm
+            end_dttm = start_dttm + timedelta(days=DATE_WINDOW_SIZE)
+
+            if end_dttm > now_dttm:
+                end_dttm = now_dttm
+
+        singer.write_state(self.state)
+
+    def parse_report_metadata(self, pileOfRows):
+        '''
+            Restructure data from report response on daily basis and update self.parsed_metadata dictionary
+            {
+                "dates": ["2021-07-01", "2021-07-02", "2021-07-03"],
+                "data": [ {
+                    "name": "Total Income",
+                    "values": ["4.00", "4.00", "4.00", "12.00"]
+                }, {
+                    "name": "Gross Profit",
+                    "values": ["4.00", "4.00", "4.00", "12.00"]
+                }, {
+                    "name": "Total Expenses",
+                    "values": ["1.00", "1.00", "1.00", "3.00"]
+                }, {
+                    "name": "Net Income",
+                    "values": ["3.00", "3.00", "3.00", "9.00"]
+                }]
+            }
+        '''
+
+        if isinstance(pileOfRows, list):
+            for x in pileOfRows:
+                self.parse_report_metadata(x)
+
+        else:
+
+            if 'Rows' in pileOfRows.keys():
+                self.parse_report_metadata(pileOfRows['Rows'])
+
+            if 'Row' in pileOfRows.keys():
+                self.parse_report_metadata(pileOfRows['Row'])
+
+            if 'Summary' in pileOfRows.keys():
+                self.parse_report_metadata(pileOfRows['Summary'])
+
+            if 'ColData' in pileOfRows.keys():
+                d = dict()
+                d['name'] = pileOfRows['ColData'][0]['value']
+                vals = []
+                for x in pileOfRows['ColData'][1:]:
+                    vals.append(x['value'])
+                d['values'] = vals
+                self.parsed_metadata['data'].append(d)
+
+            if 'Columns' in pileOfRows.keys():
+                self.parse_report_metadata(pileOfRows['Columns'])
+
+            if 'Column' in pileOfRows.keys():
+                for x in pileOfRows['Column']:
+                    if 'MetaData' in x.keys():
+                        for md in x['MetaData']:
+                            if md['Name'] in ['StartDate']:
+                                self.parsed_metadata['dates'].append(md['Value'])
+
+    def day_wise_reports(self):
+        '''
+            Return record for every day formed using output of parse_report_metadata
+        '''
+        for i, date in enumerate(self.parsed_metadata['dates']):
+            report = dict()
+            report['ReportDate'] = date
+            report['AccountingMethod'] = 'Accrual'
+            report['Details'] = {}
+
+            for data in self.parsed_metadata['data']:
+                report['Details'][data['name']] = data['values'][i]
+
+            yield report
+
+
+class ProfitAndLossReport(ReportStream):
+    stream_name = 'profit_loss_report'
+    endpoint = '/v3/company/{realm_id}/reports/ProfitAndLoss'
 
 STREAM_OBJECTS = {
     "accounts": Accounts,
@@ -222,5 +355,6 @@ class Vendors(Stream):
     "time_activities": TimeActivities,
     "transfers": Transfers,
     "vendor_credits": VendorCredits,
-    "vendors": Vendors
+    "vendors": Vendors,
+    "profit_loss_report": ProfitAndLossReport
 }

tests/base.py

@@ -1,5 +1,6 @@
 import os
 import unittest
+import time
 from datetime import datetime as dt
 from datetime import timedelta
 
@@ -23,6 +24,12 @@ class TestQuickbooksBase(unittest.TestCase):
     INCREMENTAL = "INCREMENTAL"
     FULL = "FULL_TABLE"
     START_DATE_FORMAT = "%Y-%m-%dT00:00:00Z" # %H:%M:%SZ
+    DATETIME_FMT = {
+        "%Y-%m-%dT%H:%M:%SZ",
+        "%Y-%m-%d %H:%M:%S",
+        "%Y-%m-%dT%H:%M:%S.000000Z",
+        "%Y-%m-%dT%H:%M:%S%z"
+    }
 
     def setUp(self):
         missing_envs = [x for x in [
@@ -89,18 +96,25 @@ def expected_check_streams():
             "transfers",
             "vendor_credits",
             "vendors",
+            "profit_loss_report"
         }
 
     def expected_metadata(self):
         """The expected streams and metadata about the streams"""
 
         mdata = {}
         for stream in self.expected_check_streams():
-            mdata[stream] = {
-                self.PRIMARY_KEYS: {'Id'},
-                self.REPLICATION_METHOD: self.INCREMENTAL,
-                self.REPLICATION_KEYS: {'MetaData'},
-            }
+            if self.is_report_stream(stream):
+                mdata[stream] = {
+                    self.REPLICATION_METHOD: self.INCREMENTAL,
+                    self.REPLICATION_KEYS: {'ReportDate'},
+                }
+            else:
+                mdata[stream] = {
+                    self.PRIMARY_KEYS: {'Id'},
+                    self.REPLICATION_METHOD: self.INCREMENTAL,
+                    self.REPLICATION_KEYS: {'MetaData'},
+                }
 
         return mdata
 
@@ -222,3 +236,14 @@ def minimum_record_count_by_stream(self):
         record_counts["vendors"] = 26
 
         return record_counts
+
+    def dt_to_ts(self, dtime):
+        for date_format in self.DATETIME_FMT:
+            try:
+                date_stripped = int(time.mktime(dt.strptime(dtime, date_format).timetuple()))
+                return date_stripped
+            except ValueError:
+                continue
+
+    def is_report_stream(self, stream):
+        return stream in ["profit_loss_report"]

tests/test_quickbooks_bookmarks.py

@@ -138,20 +138,29 @@ def test_run(self):
                 # Verify the second sync records respect the previous (simulated) bookmark value
                 simulated_bookmark_value = new_state['bookmarks'][stream][sub_level_replication_key]
                 for message in second_sync_messages:
-                    replication_key_value = message.get('data').get(top_level_replication_key).get(sub_level_replication_key)
-                    self.assertGreaterEqual(replication_key_value, simulated_bookmark_value,
+                    if self.is_report_stream(stream):
+                        replication_key_value = message.get('data').get('ReportDate')
+                    else:
+                        replication_key_value = message.get('data').get(top_level_replication_key).get(sub_level_replication_key)
+                    self.assertGreaterEqual(self.dt_to_ts(replication_key_value), self.dt_to_ts(simulated_bookmark_value),
                                             msg="Second sync records do not repect the previous bookmark.")
 
                 # Verify the first sync bookmark value is the max replication key value for a given stream
                 for message in first_sync_messages:
-                    replication_key_value = message.get('data').get(top_level_replication_key).get(sub_level_replication_key)
-                    self.assertLessEqual(replication_key_value, first_bookmark_value_utc,
+                    if self.is_report_stream(stream):
+                        replication_key_value = message.get('data').get('ReportDate')
+                    else:
+                        replication_key_value = message.get('data').get(top_level_replication_key).get(sub_level_replication_key)
+                    self.assertLessEqual(self.dt_to_ts(replication_key_value), self.dt_to_ts(first_bookmark_value_utc),
                                          msg="First sync bookmark was set incorrectly, a record with a greater rep key value was synced")
 
                 # Verify the second sync bookmark value is the max replication key value for a given stream
                 for message in second_sync_messages:
-                    replication_key_value = message.get('data').get(top_level_replication_key).get(sub_level_replication_key)
-                    self.assertLessEqual(replication_key_value, second_bookmark_value_utc,
+                    if self.is_report_stream(stream):
+                        replication_key_value = message.get('data').get('ReportDate')
+                    else:
+                        replication_key_value = message.get('data').get(top_level_replication_key).get(sub_level_replication_key)
+                    self.assertLessEqual(self.dt_to_ts(replication_key_value), self.dt_to_ts(second_bookmark_value_utc),
                                          msg="Second sync bookmark was set incorrectly, a record with a greater rep key value was synced")
 
                 # Verify the number of records in the 2nd sync is less then the first

tests/test_quickbooks_pagination.py

@@ -67,18 +67,24 @@ def test_run(self):
 
                 sync_messages = sync_records.get(stream, {'messages': []}).get('messages')
 
-                primary_key = self.expected_primary_keys().get(stream).pop()
-
                 # Verify the sync meets or exceeds the default record count
                 self.assertLessEqual(expected_count, record_count)
 
                 # Verify the number or records exceeds the max_results (api limit)
-                pagination_threshold = int(self.get_properties().get(page_size_key))
+                if self.is_report_stream(stream):
+                    #Tap is making API call in 30 days window for reports stream
+                    pagination_threshold = 30
+                else:
+                    pagination_threshold = int(self.get_properties().get(page_size_key))
                 self.assertGreater(record_count, pagination_threshold,
                                    msg="Record count not large enough to gaurantee pagination.")
 
-                # Verify we did not duplicate any records across pages
-                records_pks_set = {message.get('data').get(primary_key) for message in sync_messages}
-                records_pks_list = [message.get('data').get(primary_key) for message in sync_messages]
-                self.assertCountEqual(records_pks_set, records_pks_list,
-                                      msg="We have duplicate records for {}".format(stream))
+                # ProfitAndLoss report stream doesn't have any primary key
+                if not self.is_report_stream(stream):
+                    primary_key = self.expected_primary_keys().get(stream).pop()
+
+                    # Verify we did not duplicate any records across pages
+                    records_pks_set = {message.get('data').get(primary_key) for message in sync_messages}
+                    records_pks_list = [message.get('data').get(primary_key) for message in sync_messages]
+                    self.assertCountEqual(records_pks_set, records_pks_list,
+                                        msg="We have duplicate records for {}".format(stream))

tests/test_quickbooks_start_date.py

@@ -89,7 +89,9 @@ def test_run(self):
 
                 # start dates
                 start_date_1 = self.get_properties()['start_date']
+                start_date_1_epoch = self.dt_to_ts(start_date_1)
                 start_date_2 = self.get_properties(original=False)['start_date']
+                start_date_2_epoch = self.dt_to_ts(start_date_2)
 
                 # Verify by stream that our first sync meets or exceeds the default record count
                 self.assertLessEqual(expected_first_sync_count, first_sync_count)
@@ -99,10 +101,16 @@ def test_run(self):
 
                 # Verify by stream that all records have a rep key that is equal to or greater than that sync's start_date
                 for message in first_sync_messages:
-                    rep_key_value = message.get('data').get('MetaData').get('LastUpdatedTime')
-                    self.assertGreaterEqual(rep_key_value, start_date_1,
+                    if self.is_report_stream(stream):
+                        rep_key_value = message.get('data').get('ReportDate')
+                    else :
+                        rep_key_value = message.get('data').get('MetaData').get('LastUpdatedTime')
+                    self.assertGreaterEqual(self.dt_to_ts(rep_key_value), start_date_1_epoch,
                                             msg="A record was replicated with a replication key value prior to the start date")
                 for message in second_sync_messages:
-                    rep_key_value = message.get('data').get('MetaData').get('LastUpdatedTime')
-                    self.assertGreaterEqual(rep_key_value, start_date_2,
+                    if self.is_report_stream(stream):
+                        rep_key_value = message.get('data').get('ReportDate')
+                    else :
+                        rep_key_value = message.get('data').get('MetaData').get('LastUpdatedTime')
+                    self.assertGreaterEqual(self.dt_to_ts(rep_key_value), start_date_2_epoch,
                                             msg="A record was replicated with a replication key value prior to the start date")