avro-helper-devlibx

Python DSL for setting up Flask app CDC

Project description

How to use

Install pip install avro-helper-devlibx

Quick example - V1

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)

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

0.2.3

Apr 19, 2023

0.2.2

Mar 31, 2023

0.2.0

Mar 31, 2023

0.1.6

Mar 31, 2023

0.1.5

Mar 30, 2023

0.1.4

Mar 30, 2023

0.1.3

Sep 21, 2022

0.1.2

Sep 21, 2022

This version

0.1.1

Sep 21, 2022

0.1.0

Aug 29, 2022

0.0.12

Aug 25, 2022

0.0.11

Aug 22, 2022

0.0.10

Aug 15, 2022

0.0.9

Aug 15, 2022

0.0.8

Aug 15, 2022

0.0.7

Aug 7, 2022

0.0.6

Aug 7, 2022

0.0.5

Aug 7, 2022

0.0.4

Aug 6, 2022

0.0.3

Aug 6, 2022

0.0.2

Aug 6, 2022

0.0.1

Aug 6, 2022

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-0.1.1.tar.gz (7.8 kB view details)

Uploaded Sep 21, 2022 Source

Built Distribution

If you're not sure about the file name format, learn more about wheel file names.

The dropdown lists show the available interpreters, ABIs, and platforms. Enable javascript to be able to filter the list of wheel files.

avro_helper_devlibx-0.1.1-py3-none-any.whl (8.9 kB view details)

Uploaded Sep 21, 2022 Python 3

File details

Details for the file avro-helper-devlibx-0.1.1.tar.gz.

File metadata

Download URL: avro-helper-devlibx-0.1.1.tar.gz
Upload date: Sep 21, 2022
Size: 7.8 kB
Tags: Source
Uploaded using Trusted Publishing? No
Uploaded via: twine/4.0.1 CPython/3.9.14

File hashes

Hashes for avro-helper-devlibx-0.1.1.tar.gz
Algorithm	Hash digest
SHA256	`b60c201a184aeceb1d3f3831f5852275db42479f7e0001b70877e942e6dd9346`
MD5	`b761b7e0ed5b4ae673b027c0d084a3e7`
BLAKE2b-256	`f2c5918d4afc9a393b574d5dfd345499b3a4ae911e66a72ebb5b544571febf48`

See more details on using hashes here.

File details

Details for the file avro_helper_devlibx-0.1.1-py3-none-any.whl.

File metadata

Download URL: avro_helper_devlibx-0.1.1-py3-none-any.whl
Upload date: Sep 21, 2022
Size: 8.9 kB
Tags: Python 3
Uploaded using Trusted Publishing? No
Uploaded via: twine/4.0.1 CPython/3.9.14

File hashes

Hashes for avro_helper_devlibx-0.1.1-py3-none-any.whl
Algorithm	Hash digest
SHA256	`78ed2d55ba1c6754adc3c60bc8fc86c5337a1c9b65cbf5198907dd8c656308b5`
MD5	`00e5c7db397c8d5555b47a2f65f0f6c9`
BLAKE2b-256	`db651071fe73781e4dbc2d9c13e5abbbb7cf01f3421fbebfacb7ace8bb8ec2e1`

See more details on using hashes here.

avro-helper-devlibx 0.1.1

Navigation

Verified details

Maintainers

Unverified details

Project links

Meta

Project description

How to use

Quick example - V1

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

Quick example

Get data for this month

Get data for this week (last 7 days)

Get data for day

Get string data for this month

Get string data for this week

Get string data for the day

Project details

Verified details

Maintainers

Unverified details

Project links

Meta

Release history Release notifications | RSS feed

Download files

Source Distribution

Built Distribution

File details

File metadata

File hashes

File details

File metadata

File hashes