Python DSL for setting up Flask app CDC

Project description

How to use

Install pip install avro-helper-devlibx

Quick example - V1

Debug what data I have

from devlibx_avro_helper.month_data_v1 import MonthDataAvroHelperV1

input = '''{"updated_at":1663665518937,"days":{"9-17":1,"9-4":1,"9-18":1,"9-5":1,"9-15":1,"9-6":1,"9-16":1}}'''
helper = MonthDataAvroHelperV1(input)
helper.dump_to_debug()

# Print the data
# -------------------------------------- Start: Data --------------------------------------------------
# Day Aggregations
# {'9-17': 1, '9-4': 1, '9-18': 1, '9-5': 1, '9-15': 1, '9-6': 1, '9-16': 1}
# -------------------------------------- End: Data ----------------------------------------------------

Get data for this month (from today to start of this month)

from devlibx_avro_helper.month_data_v1 import MonthDataAvroHelperV1


def test__get_current_month_numeric_aggregation_from_now_for_readme(self):
    # This is the data you will get from DB or some other place
    inputDataFromDB = '''
                  {"updated_at":1663665518937,"days":{"9-1":3,"9-2":1,"9-3":2, "9-4":1, "9-5":1}}
                  '''
    helper = MonthDataAvroHelperV1(inputDataFromDB)

    # Check with aggregate=True
    result = helper.get_current_month_numeric_aggregation_from_now()
    print(result)
    # >> 8

    self.assertEqual(8, result)

Get data for last N days (from today to last n days - including today)

from devlibx_avro_helper.month_data_v1 import MonthDataAvroHelperV1
from datetime import datetime


def test__get_last_n_days_numeric_aggregation_from_given_time(self):
    # This is the data you will get from DB or some other place
    inputDataFromDB = '''
                      {"updated_at":1663665518937,"days":{"9-1":3,"9-2":2,"9-3":7, "9-4":3, "9-5":11}}
                      '''
    helper = MonthDataAvroHelperV1(inputDataFromDB)

    # Check with aggregate=True
    result = helper.get_last_n_days_numeric_aggregation_from_given_time(datetime.now(), 4)
    print(result)
    # >> 23
    self.assertEqual(23, result)

Quick example

from devlibx_avro_helper.month_data import MonthDataAvroHelper

base64Str = "BgY3LTMCBjYtNgIGNy01BAAAAAI="
helper = MonthDataAvroHelper()
result = helper.process(base64Str)
print(result)

# Result
# {'days': {'8-16': 110, '8-17': 111, '8-14': 108, '8-15': 109, '8-18': 112, '8-19': 113, '8-30': 124, '8-31': 125, '8-12': 106, '8-13': 107, '8-10': 104, '8-11': 105, '9-1': 126, '9-2': 127, '9-3': 128, '8-27': 121, '9-4': 129, '8-6': 100, '8-28': 122, '8-7': 101, '8-25': 119, '8-8': 102, '8-26': 120, '8-9': 103, '8-29': 123, '8-20': 114, '8-23': 117, '8-24': 118, '8-21': 115, '8-22': 116}, 'entity_id': 'harish_1'}

Get data for this month

In this example we would have data in base 64 encoding. We will get aggregated data for this month

from devlibx_avro_helper.month_data import MonthDataAvroHelper
from datetime import datetime


def test_process_and_return_aggregation_for_month(self):
    base64Str = "AgoGNy0xAgY3LTICBjctMwIGNy00AgY3LTUKAAAAAAI="
    helper = MonthDataAvroHelper()
    result = helper.process(base64Str)
    print(result)
    # Output = {'days': {'7-1': 1, '7-2': 1, '7-3': 1, '7-4': 1, '7-5': 5}, 'days_str': None, 'entity_id': None, 'sub_entity_id': None, 'version': 1}

    date_time_str = '05/07/22 01:55:19'
    date_time_obj = datetime.strptime(date_time_str, '%d/%m/%y %H:%M:%S')

    # if you are looking for data for this month then use can use
    # helper.process_and_return_aggregation_for_this_month(base64Str)
    result = helper.process_and_return_aggregation_for_month(date_time_obj, base64Str)
    self.assertEqual(9, result, "result should be 9")
    # Output = 9

Get data for this week (last 7 days)

from devlibx_avro_helper.month_data import MonthDataAvroHelper
from datetime import datetime


def test_process_and_return_aggregation_for_week(self):
    # Test 1 - data from generateDataFor_test_parsing_Test_2
    base64Str = "Ag4INi0yOQIGNy0xAgY3LTICCDYtMzACBjctMwIGNy00AgY3LTUKAAAAAAI=="
    helper = MonthDataAvroHelper()
    result = helper.process(base64Str)
    print(result)
    # Output = {'days': {'6-29': 1, '7-1': 1, '7-2': 1, '6-30': 1, '7-3': 1, '7-4': 1, '7-5': 5}, 'days_str': None, 'entity_id': None, 'sub_entity_id': None, 'version': 1}

    date_time_str = '05/07/22 01:55:19'
    date_time_obj = datetime.strptime(date_time_str, '%d/%m/%y %H:%M:%S')

    # if you are looking for data for this month then use can use
    # helper.process_and_return_aggregation_for_this_week(base64Str)
    result = helper.process_and_return_aggregation_for_week(date_time_obj, base64Str)
    self.assertEqual(11, result, "result should be 9")
    # Output = 11

Get data for day

from devlibx_avro_helper.month_data import MonthDataAvroHelper
from datetime import datetime


def test_process_and_return_for_day(self):
    # Test 1 - data from generateDataFor_test_parsing_Test_2
    base64Str = "Ag4INi0yOQIGNy0xAgY3LTICCDYtMzACBjctMwIGNy00AgY3LTUKAAAAAAI=="
    helper = MonthDataAvroHelper()
    result = helper.process(base64Str)
    print(result)
    # Output = {'days': {'6-29': 1, '7-1': 1, '7-2': 1, '6-30': 1, '7-3': 1, '7-4': 1, '7-5': 5}, 'days_str': None, 'entity_id': None, 'sub_entity_id': None, 'version': 1}

    date_time_str = '05/07/22 01:55:19'
    date_time_obj = datetime.strptime(date_time_str, '%d/%m/%y %H:%M:%S')

    # if you are looking for data for this month then use can use
    # helper.process_and_return_aggregation_for_this_month(base64Str)
    result = helper.process_and_return_for_today(date_time_obj, base64Str)
    self.assertEqual(5, result, "result should be 9")
    # Output = 5

Get string data for this month

from devlibx_avro_helper.month_data import MonthDataAvroHelper
from datetime import datetime


def test_process_and_return_string_data_for_month(self):
    # Test 1 - data from generateDataFor_test_parsing_Test_2
    base64Str = "AAIOCDYtMjkCAgKamZmZmZnxPwAAAg5tLTE7bS0zAAAAAAAGNy0xAgYCMzMzMzMzC0AAAAIWbS0zO20tNDttLTYAAAAAAAY3LTICCALNzMzMzMwQQAAAAhZtLTQ7bS03O20tOAAAAAAACDYtMzACBAJmZmZmZmYCQAAAAhhtLTI7bS0xO20tMTEAAAAAAAY3LTMCCgJmZmZmZmYWQAAAAhZtLTM7bS02O20tOAAAAAAABjctNAIMAjMzMzMzMx9AAAACGG0tMjttLTc7bS0xMAAAAAAABjctNQIOAs3MzMzMzCFAAAACIG0tMjttLTU7bS03O20tMTAAAAAAAAAC"
    helper = MonthDataAvroHelper()
    result = helper.process(base64Str)
    print(result)
    # Output = {'ParentContainer': [], 'data': {'6-29': {'counter': 1, 'aggregate': 1.1, 'counter_secondary': None, 'aggregate_secondary': None, 'str': 'm-1;m-3', 'udf1': None, 'udf2': None, 'udf3': None, 'udf4': None, 'udf5': None}, '7-1': {'counter': 3, 'aggregate': 3.4, 'counter_secondary': None, 'aggregate_secondary': None, 'str': 'm-3;m-4;m-6', 'udf1': None, 'udf2': None, 'udf3': None, 'udf4': None, 'udf5': None}, '7-2': {'counter': 4, 'aggregate': 4.2, 'counter_secondary': None, 'aggregate_secondary': None, 'str': 'm-1;m-4;m-7;m-8', 'udf1': None, 'udf2': None, 'udf3': None, 'udf4': None, 'udf5': None}, '6-30': {'counter': 2, 'aggregate': 2.3, 'counter_secondary': None, 'aggregate_secondary': None, 'str': 'm-2;m-1', 'udf1': None, 'udf2': None, 'udf3': None, 'udf4': None, 'udf5': None}, '7-3': {'counter': 5, 'aggregate': 5.6, 'counter_secondary': None, 'aggregate_secondary': None, 'str': 'm-3;m-6;m-8', 'udf1': None, 'udf2': None, 'udf3': None, 'udf4': None, 'udf5': None}, '7-4': {'counter': 6, 'aggregate': 7.8, 'counter_secondary': None, 'aggregate_secondary': None, 'str': 'm-2;m-7;m-10', 'udf1': None, 'udf2': None, 'udf3': None, 'udf4': None, 'udf5': None}, '7-5': {'counter': 7, 'aggregate': 8.9, 'counter_secondary': None, 'aggregate_secondary': None, 'str': 'm-1;m-2;m-5;m-7;m-10', 'udf1': None, 'udf2': None, 'udf3': None, 'udf4': None, 'udf5': None}}, 'version': 1}

    date_time_str = '05/07/22 01:55:19'
    date_time_obj = datetime.strptime(date_time_str, '%d/%m/%y %H:%M:%S')

    # if you are looking for data for this month then use can use
    # helper.process_and_return_aggregation_for_this_month(base64Str)
    result = helper.process_and_return_string_data_for_month(date_time_obj, base64Str)
    self.assertEqual(8, result, "result should be 8")
    # Output = 8

    date_time_str = '30/06/22 01:55:19'
    date_time_obj = datetime.strptime(date_time_str, '%d/%m/%y %H:%M:%S')

    result = helper.process_and_return_string_data_for_month(date_time_obj, base64Str)
    self.assertEqual(4, result, "result should be 4")
    # Output = 8

Get string data for this week

from devlibx_avro_helper.month_data import MonthDataAvroHelper
from datetime import datetime


def test_process_and_return_string_data_for_week(self):
    # Test 1 - data from generateDataFor_test_parsing_Test_2
    base64Str = "AAIOCDYtMjkCAgKamZmZmZnxPwAAAg5tLTE7bS0zAAAAAAAGNy0xAgYCMzMzMzMzC0AAAAIWbS0zO20tNDttLTYAAAAAAAY3LTICCALNzMzMzMwQQAAAAhZtLTQ7bS03O20tOAAAAAAACDYtMzACBAJmZmZmZmYCQAAAAhhtLTI7bS0xO20tMTEAAAAAAAY3LTMCCgJmZmZmZmYWQAAAAhZtLTM7bS02O20tOAAAAAAABjctNAIMAjMzMzMzMx9AAAACGG0tMjttLTc7bS0xMAAAAAAABjctNQIOAs3MzMzMzCFAAAACIG0tMjttLTU7bS03O20tMTAAAAAAAAAC"
    helper = MonthDataAvroHelper()
    result = helper.process(base64Str)
    print(result)
    # Output = {'ParentContainer': [], 'data': {'6-29': {'counter': 1, 'aggregate': 1.1, 'counter_secondary': None, 'aggregate_secondary': None, 'str': 'm-1;m-3', 'udf1': None, 'udf2': None, 'udf3': None, 'udf4': None, 'udf5': None}, '7-1': {'counter': 3, 'aggregate': 3.4, 'counter_secondary': None, 'aggregate_secondary': None, 'str': 'm-3;m-4;m-6', 'udf1': None, 'udf2': None, 'udf3': None, 'udf4': None, 'udf5': None}, '7-2': {'counter': 4, 'aggregate': 4.2, 'counter_secondary': None, 'aggregate_secondary': None, 'str': 'm-1;m-4;m-7;m-8', 'udf1': None, 'udf2': None, 'udf3': None, 'udf4': None, 'udf5': None}, '6-30': {'counter': 2, 'aggregate': 2.3, 'counter_secondary': None, 'aggregate_secondary': None, 'str': 'm-2;m-1', 'udf1': None, 'udf2': None, 'udf3': None, 'udf4': None, 'udf5': None}, '7-3': {'counter': 5, 'aggregate': 5.6, 'counter_secondary': None, 'aggregate_secondary': None, 'str': 'm-3;m-6;m-8', 'udf1': None, 'udf2': None, 'udf3': None, 'udf4': None, 'udf5': None}, '7-4': {'counter': 6, 'aggregate': 7.8, 'counter_secondary': None, 'aggregate_secondary': None, 'str': 'm-2;m-7;m-10', 'udf1': None, 'udf2': None, 'udf3': None, 'udf4': None, 'udf5': None}, '7-5': {'counter': 7, 'aggregate': 8.9, 'counter_secondary': None, 'aggregate_secondary': None, 'str': 'm-1;m-2;m-5;m-7;m-10', 'udf1': None, 'udf2': None, 'udf3': None, 'udf4': None, 'udf5': None}}, 'version': 1}

    date_time_str = '05/07/22 01:55:19'
    date_time_obj = datetime.strptime(date_time_str, '%d/%m/%y %H:%M:%S')

    # if you are looking for data for this month then use can use
    # helper.process_and_return_aggregation_for_this_month(base64Str)
    result = helper.process_and_return_string_data_for_week(date_time_obj, base64Str)
    print(result)
    self.assertEqual(10, result, "result should be 10")
    # Output = 10

Get string data for the day

from devlibx_avro_helper.month_data import MonthDataAvroHelper
from datetime import datetime


def test_process_and_return_string_data_for_day(self):
    # Test 1 - data from generateDataFor_test_parsing_Test_2
    base64Str = "AAIOCDYtMjkCAgKamZmZmZnxPwAAAg5tLTE7bS0zAAAAAAAGNy0xAgYCMzMzMzMzC0AAAAIWbS0zO20tNDttLTYAAAAAAAY3LTICCALNzMzMzMwQQAAAAhZtLTQ7bS03O20tOAAAAAAACDYtMzACBAJmZmZmZmYCQAAAAhhtLTI7bS0xO20tMTEAAAAAAAY3LTMCCgJmZmZmZmYWQAAAAhZtLTM7bS02O20tOAAAAAAABjctNAIMAjMzMzMzMx9AAAACGG0tMjttLTc7bS0xMAAAAAAABjctNQIOAs3MzMzMzCFAAAACIG0tMjttLTU7bS03O20tMTAAAAAAAAAC"
    helper = MonthDataAvroHelper()
    result = helper.process(base64Str)
    print(result)
    # Output = {'ParentContainer': [], 'data': {'6-29': {'counter': 1, 'aggregate': 1.1, 'counter_secondary': None, 'aggregate_secondary': None, 'str': 'm-1;m-3', 'udf1': None, 'udf2': None, 'udf3': None, 'udf4': None, 'udf5': None}, '7-1': {'counter': 3, 'aggregate': 3.4, 'counter_secondary': None, 'aggregate_secondary': None, 'str': 'm-3;m-4;m-6', 'udf1': None, 'udf2': None, 'udf3': None, 'udf4': None, 'udf5': None}, '7-2': {'counter': 4, 'aggregate': 4.2, 'counter_secondary': None, 'aggregate_secondary': None, 'str': 'm-1;m-4;m-7;m-8', 'udf1': None, 'udf2': None, 'udf3': None, 'udf4': None, 'udf5': None}, '6-30': {'counter': 2, 'aggregate': 2.3, 'counter_secondary': None, 'aggregate_secondary': None, 'str': 'm-2;m-1', 'udf1': None, 'udf2': None, 'udf3': None, 'udf4': None, 'udf5': None}, '7-3': {'counter': 5, 'aggregate': 5.6, 'counter_secondary': None, 'aggregate_secondary': None, 'str': 'm-3;m-6;m-8', 'udf1': None, 'udf2': None, 'udf3': None, 'udf4': None, 'udf5': None}, '7-4': {'counter': 6, 'aggregate': 7.8, 'counter_secondary': None, 'aggregate_secondary': None, 'str': 'm-2;m-7;m-10', 'udf1': None, 'udf2': None, 'udf3': None, 'udf4': None, 'udf5': None}, '7-5': {'counter': 7, 'aggregate': 8.9, 'counter_secondary': None, 'aggregate_secondary': None, 'str': 'm-1;m-2;m-5;m-7;m-10', 'udf1': None, 'udf2': None, 'udf3': None, 'udf4': None, 'udf5': None}}, 'version': 1}

    date_time_str = '05/07/22 01:55:19'
    date_time_obj = datetime.strptime(date_time_str, '%d/%m/%y %H:%M:%S')

    # if you are looking for data for this month then use can use
    # helper.process_and_return_aggregation_for_this_month(base64Str)
    result = helper.process_and_return_string_data_for_day(date_time_obj, base64Str)
    self.assertEqual(4, result, "result should be 4")
    # Output = 4

Project details

Release history Release notifications | RSS feed

This version

0.2.1

Mar 31, 2023

0.2.0

Mar 31, 2023

Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

avro-helper-devlibx-v1-0.2.1.tar.gz (9.2 kB view details)

Uploaded Mar 31, 2023 Source

Built Distribution

avro_helper_devlibx_v1-0.2.1-py3-none-any.whl (10.1 kB view details)

Uploaded Mar 31, 2023 Python 3

File details

Details for the file avro-helper-devlibx-v1-0.2.1.tar.gz.

File metadata

Download URL: avro-helper-devlibx-v1-0.2.1.tar.gz
Upload date: Mar 31, 2023
Size: 9.2 kB
Tags: Source
Uploaded using Trusted Publishing? No
Uploaded via: twine/4.0.1 CPython/3.11.2

File hashes

Hashes for avro-helper-devlibx-v1-0.2.1.tar.gz
Algorithm	Hash digest
SHA256	`b2b272b0a5f5323694347e80184c0a040c894c3ee5fa398525fd85896adb4916`
MD5	`6f4804068f2b281a1cdec4bbea8faf1e`
BLAKE2b-256	`9b225ccae786f05b195391286d0c8c7827cf56327c03eb515a7687657cd1608e`

See more details on using hashes here.

File details

Details for the file avro_helper_devlibx_v1-0.2.1-py3-none-any.whl.

File metadata

Download URL: avro_helper_devlibx_v1-0.2.1-py3-none-any.whl
Upload date: Mar 31, 2023
Size: 10.1 kB
Tags: Python 3
Uploaded using Trusted Publishing? No
Uploaded via: twine/4.0.1 CPython/3.11.2

File hashes

Hashes for avro_helper_devlibx_v1-0.2.1-py3-none-any.whl
Algorithm	Hash digest
SHA256	`3a7e3e5679e6445c82a5b5bbc9ee3b75aa7ae83832a327d7f48d5a0c9ae6bc96`
MD5	`d332249a7d79b1240ce49a0e66841817`
BLAKE2b-256	`fb16dbff46e5d2732a5990062c8c9a4a91ccbf156ae2607d59045c427b9bc085`