diff --git a/target_csv.py b/target_csv.py index 6ec5a80..ab76d15 100755 --- a/target_csv.py +++ b/target_csv.py @@ -13,6 +13,7 @@ from datetime import datetime import collections import pkg_resources +import pathlib from jsonschema.validators import Draft4Validator import singer @@ -96,6 +97,27 @@ def persist_messages(delimiter, quotechar, messages, destination_path, fixed_hea r = simplejson.dumps(flattened_record) writer.writerow(simplejson.loads(r)) + job_metrics_file_path = os.path.expanduser(os.path.join(destination_path, "job_metrics.json")) + + if not os.path.isfile(job_metrics_file_path): + pathlib.Path(job_metrics_file_path).touch() + + with open(job_metrics_file_path, 'r+') as job_metrics_file: + content = dict() + + try: + content = json.loads(job_metrics_file.read()) + except Exception: + pass + + if not content.get('recordCount'): + content['recordCount'] = dict() + + content['recordCount'][o['stream']] = content['recordCount'].get(o['stream'], 0) + 1 + + job_metrics_file.seek(0) + job_metrics_file.write(json.dumps(content)) + state = None elif message_type == 'STATE': logger.debug('Setting state to {}'.format(o['value']))