Unlock 30% off on Manual Testing Annual Plans this Holiday Season.

Copied to Clipboard!

How to use transactions method in mailosaur-python

Best Python code snippet using mailosaur-python_python

test_transactions.py

Source:test_transactions.py

1from indy_common.constants import NYM, NODE, ATTRIB, SCHEMA, CLAIM_DEF, DISCLO, GET_ATTR, GET_NYM, GET_TXNS, \2    GET_SCHEMA, GET_CLAIM_DEF, POOL_UPGRADE, NODE_UPGRADE, \3    POOL_CONFIG, REVOC_REG_DEF, REVOC_REG_ENTRY, \4    GET_REVOC_REG_DEF, GET_REVOC_REG, GET_REVOC_REG_DELTA, POOL_RESTART, VALIDATOR_INFO, CHANGE_KEY, AUTH_RULE, \5    GET_AUTH_RULE, AUTH_RULES, RICH_SCHEMA, RICH_SCHEMA_ENCODING, \6    RICH_SCHEMA_MAPPING, RICH_SCHEMA_CRED_DEF, JSON_LD_CONTEXT, GET_RICH_SCHEMA_OBJECT_BY_METADATA, \7    GET_RICH_SCHEMA_OBJECT_BY_ID, RICH_SCHEMA_PRES_DEF8from indy_common.transactions import IndyTransactions9def test_transactions_are_encoded():10    assert NODE == "0"11    assert NYM == "1"12    assert GET_TXNS == "3"13    assert ATTRIB == "100"14    assert SCHEMA == "101"15    assert CLAIM_DEF == "102"16    assert DISCLO == "103"17    assert GET_ATTR == "104"18    assert GET_NYM == "105"19    assert GET_SCHEMA == "107"20    assert GET_CLAIM_DEF == "108"21    assert POOL_UPGRADE == "109"22    assert NODE_UPGRADE == "110"23    assert POOL_CONFIG == "111"24    assert CHANGE_KEY == "112"25    assert REVOC_REG_DEF == "113"26    assert REVOC_REG_ENTRY == "114"27    assert GET_REVOC_REG_DEF == "115"28    assert GET_REVOC_REG == "116"29    assert GET_REVOC_REG_DELTA == "117"30    assert POOL_RESTART == "118"31    assert VALIDATOR_INFO == "119"32    assert AUTH_RULE == "120"33    assert GET_AUTH_RULE == "121"34    assert AUTH_RULES == "122"35    assert JSON_LD_CONTEXT == "200"36    assert RICH_SCHEMA == "201"37    assert RICH_SCHEMA_ENCODING == "202"38    assert RICH_SCHEMA_MAPPING == "203"39    assert RICH_SCHEMA_CRED_DEF == "204"40    assert RICH_SCHEMA_PRES_DEF == "205"41    assert GET_RICH_SCHEMA_OBJECT_BY_ID == "300"42    assert GET_RICH_SCHEMA_OBJECT_BY_METADATA == "301"43def test_transaction_enum_decoded():44    assert IndyTransactions.NODE.name == "NODE"45    assert IndyTransactions.NYM.name == "NYM"46    assert IndyTransactions.ATTRIB.name == "ATTRIB"47    assert IndyTransactions.SCHEMA.name == "SCHEMA"48    assert IndyTransactions.CLAIM_DEF.name == "CLAIM_DEF"49    assert IndyTransactions.DISCLO.name == "DISCLO"50    assert IndyTransactions.GET_ATTR.name == "GET_ATTR"51    assert IndyTransactions.GET_NYM.name == "GET_NYM"52    assert IndyTransactions.GET_TXNS.name == "GET_TXNS"53    assert IndyTransactions.GET_SCHEMA.name == "GET_SCHEMA"54    assert IndyTransactions.GET_CLAIM_DEF.name == "GET_CLAIM_DEF"55    assert IndyTransactions.POOL_UPGRADE.name == "POOL_UPGRADE"56    assert IndyTransactions.NODE_UPGRADE.name == "NODE_UPGRADE"57    assert IndyTransactions.POOL_CONFIG.name == "POOL_CONFIG"58    assert IndyTransactions.POOL_RESTART.name == "POOL_RESTART"59    assert IndyTransactions.CHANGE_KEY.name == "CHANGE_KEY"60    assert IndyTransactions.REVOC_REG_DEF.name == "REVOC_REG_DEF"61    assert IndyTransactions.REVOC_REG_ENTRY.name == "REVOC_REG_ENTRY"62    assert IndyTransactions.GET_REVOC_REG_DEF.name == "GET_REVOC_REG_DEF"63    assert IndyTransactions.GET_REVOC_REG.name == "GET_REVOC_REG"64    assert IndyTransactions.GET_REVOC_REG_DELTA.name == "GET_REVOC_REG_DELTA"65    assert IndyTransactions.VALIDATOR_INFO.name == "VALIDATOR_INFO"66    assert IndyTransactions.JSON_LD_CONTEXT.name == "JSON_LD_CONTEXT"67    assert IndyTransactions.RICH_SCHEMA.name == "RICH_SCHEMA"68    assert IndyTransactions.RICH_SCHEMA_ENCODING.name == "RICH_SCHEMA_ENCODING"69    assert IndyTransactions.RICH_SCHEMA_MAPPING.name == "RICH_SCHEMA_MAPPING"70    assert IndyTransactions.RICH_SCHEMA_CRED_DEF.name == "RICH_SCHEMA_CRED_DEF"71    assert IndyTransactions.RICH_SCHEMA_PRES_DEF.name == "RICH_SCHEMA_PRES_DEF"72    assert IndyTransactions.GET_RICH_SCHEMA_OBJECT_BY_ID.name == "GET_RICH_SCHEMA_OBJECT_BY_ID"73    assert IndyTransactions.GET_RICH_SCHEMA_OBJECT_BY_METADATA.name == "GET_RICH_SCHEMA_OBJECT_BY_METADATA"74def test_transaction_enum_encoded():75    assert IndyTransactions.NODE.value == "0"76    assert IndyTransactions.NYM.value == "1"77    assert IndyTransactions.GET_TXNS.value == "3"78    assert IndyTransactions.ATTRIB.value == "100"79    assert IndyTransactions.SCHEMA.value == "101"80    assert IndyTransactions.CLAIM_DEF.value == "102"81    assert IndyTransactions.DISCLO.value == "103"82    assert IndyTransactions.GET_ATTR.value == "104"83    assert IndyTransactions.GET_NYM.value == "105"84    assert IndyTransactions.GET_SCHEMA.value == "107"85    assert IndyTransactions.GET_CLAIM_DEF.value == "108"86    assert IndyTransactions.POOL_UPGRADE.value == "109"87    assert IndyTransactions.NODE_UPGRADE.value == "110"88    assert IndyTransactions.POOL_CONFIG.value == "111"89    assert IndyTransactions.CHANGE_KEY.value == "112"90    assert IndyTransactions.REVOC_REG_DEF.value == "113"91    assert IndyTransactions.REVOC_REG_ENTRY.value == "114"92    assert IndyTransactions.GET_REVOC_REG_DEF.value == "115"93    assert IndyTransactions.GET_REVOC_REG.value == "116"94    assert IndyTransactions.GET_REVOC_REG_DELTA.value == "117"95    assert IndyTransactions.POOL_RESTART.value == "118"96    assert IndyTransactions.VALIDATOR_INFO.value == "119"97    assert IndyTransactions.JSON_LD_CONTEXT.value == "200"98    assert IndyTransactions.RICH_SCHEMA.value == "201"99    assert IndyTransactions.RICH_SCHEMA_ENCODING.value == "202"100    assert IndyTransactions.RICH_SCHEMA_MAPPING.value == "203"101    assert IndyTransactions.RICH_SCHEMA_CRED_DEF.value == "204"102    assert IndyTransactions.RICH_SCHEMA_PRES_DEF.value == "205"103    assert IndyTransactions.GET_RICH_SCHEMA_OBJECT_BY_ID.value == "300"104    assert IndyTransactions.GET_RICH_SCHEMA_OBJECT_BY_METADATA.value == "301"105def test_get_name_from_code():106    assert IndyTransactions.get_name_from_code(IndyTransactions.NODE.value) == "NODE"107    assert IndyTransactions.get_name_from_code(IndyTransactions.NYM.value) == "NYM"108    assert IndyTransactions.get_name_from_code(IndyTransactions.ATTRIB.value) == "ATTRIB"109    assert IndyTransactions.get_name_from_code(IndyTransactions.SCHEMA.value) == "SCHEMA"110    assert IndyTransactions.get_name_from_code(IndyTransactions.CLAIM_DEF.value) == "CLAIM_DEF"111    assert IndyTransactions.get_name_from_code(IndyTransactions.DISCLO.value) == "DISCLO"112    assert IndyTransactions.get_name_from_code(IndyTransactions.GET_ATTR.value) == "GET_ATTR"113    assert IndyTransactions.get_name_from_code(IndyTransactions.GET_NYM.value) == "GET_NYM"114    assert IndyTransactions.get_name_from_code(IndyTransactions.GET_TXNS.value) == "GET_TXNS"115    assert IndyTransactions.get_name_from_code(IndyTransactions.GET_SCHEMA.value) == "GET_SCHEMA"116    assert IndyTransactions.get_name_from_code(IndyTransactions.GET_CLAIM_DEF.value) == "GET_CLAIM_DEF"117    assert IndyTransactions.get_name_from_code(IndyTransactions.POOL_UPGRADE.value) == "POOL_UPGRADE"118    assert IndyTransactions.get_name_from_code(IndyTransactions.NODE_UPGRADE.value) == "NODE_UPGRADE"119    assert IndyTransactions.get_name_from_code(IndyTransactions.POOL_CONFIG.value) == "POOL_CONFIG"120    assert IndyTransactions.get_name_from_code(IndyTransactions.POOL_RESTART.value) == "POOL_RESTART"121    assert IndyTransactions.get_name_from_code(IndyTransactions.CHANGE_KEY.value) == "CHANGE_KEY"122    assert IndyTransactions.get_name_from_code(IndyTransactions.REVOC_REG_DEF.value) == "REVOC_REG_DEF"123    assert IndyTransactions.get_name_from_code(IndyTransactions.REVOC_REG_ENTRY.value) == "REVOC_REG_ENTRY"124    assert IndyTransactions.get_name_from_code(IndyTransactions.GET_REVOC_REG_DEF.value) == "GET_REVOC_REG_DEF"125    assert IndyTransactions.get_name_from_code(IndyTransactions.GET_REVOC_REG.value) == "GET_REVOC_REG"126    assert IndyTransactions.get_name_from_code(IndyTransactions.GET_REVOC_REG_DELTA.value) == "GET_REVOC_REG_DELTA"127    assert IndyTransactions.get_name_from_code(IndyTransactions.VALIDATOR_INFO.value) == "VALIDATOR_INFO"128    assert IndyTransactions.get_name_from_code(IndyTransactions.JSON_LD_CONTEXT.value) == "JSON_LD_CONTEXT"129    assert IndyTransactions.get_name_from_code(IndyTransactions.RICH_SCHEMA.value) == "RICH_SCHEMA"130    assert IndyTransactions.get_name_from_code(131        IndyTransactions.RICH_SCHEMA_ENCODING.value) == "RICH_SCHEMA_ENCODING"132    assert IndyTransactions.get_name_from_code(133        IndyTransactions.RICH_SCHEMA_MAPPING.value) == "RICH_SCHEMA_MAPPING"134    assert IndyTransactions.get_name_from_code(135        IndyTransactions.RICH_SCHEMA_CRED_DEF.value) == "RICH_SCHEMA_CRED_DEF"136    assert IndyTransactions.get_name_from_code(137        IndyTransactions.RICH_SCHEMA_PRES_DEF.value) == "RICH_SCHEMA_PRES_DEF"138    assert IndyTransactions.get_name_from_code(139        IndyTransactions.GET_RICH_SCHEMA_OBJECT_BY_ID.value) == "GET_RICH_SCHEMA_OBJECT_BY_ID"140    assert IndyTransactions.get_name_from_code(141        IndyTransactions.GET_RICH_SCHEMA_OBJECT_BY_METADATA.value) == "GET_RICH_SCHEMA_OBJECT_BY_METADATA"...

tasks.py

Source:tasks.py

1from __future__ import absolute_import, unicode_literals2from datetime import datetime3from datetime import timedelta4from celery import shared_task5from PlaidAPIImpl.settings import PLAID_CLIENT_ID, PLAID_SECRET6from .models import Item, RequestIdentifier, Transaction, Account, ItemMetadata7from .serializers import AccountSerializer, TransactionSerializer, ItemMetadataSerializer8import plaid9from plaid.api import plaid_api10from datetime import datetime11from datetime import timedelta12from .utils import clean_accounts_data, clean_itemmetadata, clean_transaction_data13from plaid.model.accounts_get_request import AccountsGetRequest14from plaid.model.item_get_request import ItemGetRequest15from plaid.model.transactions_get_request import TransactionsGetRequest16from plaid.model.transactions_get_request_options import TransactionsGetRequestOptions17from django.contrib.auth.models import User18# plaid client configuration19configuration = plaid.Configuration(20    host=plaid.Environment.Sandbox,21    api_key={22        'clientId': PLAID_CLIENT_ID,23        'secret': PLAID_SECRET,24    }25)26api_client = plaid.ApiClient(configuration)27client = plaid_api.PlaidApi(api_client)28@shared_task29def fetch_item_metadata(item_id):30    item = Item.objects.filter(item_id=item_id)31    access_token = item[0].access_token32    response = None33    # pull item meta data for associated with an item from plaid34    metadata_fetch_request = ItemGetRequest(access_token=access_token)35    try:36        response = client.item_get(metadata_fetch_request)37    except:38        print("item meta data fetch request failed")39    # log plaid API request response40    requestIdentifier = RequestIdentifier.objects.create(41        request_id=response['request_id'], request_body=str(response.data))42    requestIdentifier.save()43    data = clean_itemmetadata(item[0].pk, response['item'])44    # validate and save Item metadata45    serialier = ItemMetadataSerializer(data=data)46    serialier.is_valid(raise_exception=True)47    serialier.save()48    return "User Item metadata saved to DB"49@shared_task50def delete_transactions_from_db(removed_transactions):51    for trans_id in removed_transactions:52        Transaction.objects.filter(transaction_id=trans_id).delete()53    return "Transactions removed"54@shared_task55def save_transactions_to_db(item_id):56    item = Item.objects.filter(item_id=item_id)57    access_token = item[0].access_token58    # fetching all the transactions of last 30 days: can be changed on developer discretion59    start_dt = (datetime.now() - timedelta(days=30))60    end_dt = datetime.now()61    # offset filters out newer transactions as received from webhook62    request = TransactionsGetRequest(63        access_token=access_token,64        start_date=start_dt.date(),65        end_date=end_dt.date(),66        options=TransactionsGetRequestOptions()67    )68    try:69        response = client.transactions_get(request)70        if response['item'].error is not None:71            raise Exception("Plaid API Error: " +72                            response['item'].error.error_message)73    except Exception as error:74        print(repr(error))75    # log plaid API request response76    requestIdentifier = RequestIdentifier.objects.create(77        request_id=response['request_id'], request_body=str(response.data))78    requestIdentifier.save()79    # print(response['transactions'])80    accounts = clean_accounts_data(item[0].pk, response['accounts'])81    accnts = Account.objects.filter(item=item[0])82    # Update records in user accounts table83    for acc in accnts:84        account_value = next(85            (elem for elem in accounts if elem["account_id"] == acc.account_id), None)86        if account_value:87            accnts.filter(account_id=acc.account_id).update(88                available_balance=account_value['available_balance'],89                current_balance=account_value['current_balance'],90                name=account_value['name'],91                account_type=account_value['account_type'],92                account_subtype=account_value['account_subtype']93            )94    # Save new account records in accounts table95    save_accounts = []96    for acc in accounts:97        # if account_id fetched from request is not already present in DB, save the accounts98        if not accnts.filter(account_id=acc['account_id']).exists():99            save_accounts.append(acc)100    # validate and save new user accounts101    acc_serializer = AccountSerializer(data=save_accounts, many=True)102    acc_serializer.is_valid(raise_exception=True)103    acc_serializer.save()104    transactions = clean_transaction_data(response['transactions'])105    trans_id = [trans['transaction_id'] for trans in transactions]106    # filter transactions that already exist in DB (in case one or many fields are updated)107    trans = Transaction.objects.filter(transaction_id__in=trans_id)108    # Update existing records in transactions table109    for tran in trans:110        trans_value = next(111            (elem for elem in transactions if elem["transaction_id"] == tran.transaction_id), None)112        if trans_value:113            trans.filter(transaction_id=tran.transaction_id).update(114                amount=trans_value['amount'],115                date=trans_value['date'],116                name=trans_value['name'],117                payment_channel=trans_value['payment_channel']118            )119    # Save new records in transactions table120    save_transactions = []121    for tran in transactions:122        # transactions that are not already present in DB (new transactions)123        if not trans.filter(transaction_id=tran['transaction_id']).exists():124            save_transactions.append(tran)125    # validate and save new transactions to DB126    tr_serializer = TransactionSerializer(data=save_transactions, many=True)127    tr_serializer.is_valid(raise_exception=True)128    tr_serializer.save()129    result = f"DB entries updated, {len(save_accounts)} created. {len(save_transactions)} created"130    # print(result)131    return result132"""133@shared_task134def fetch_account_data(item_id):135    item = Item.objects.filter(item_id=item_id)136    access_token = item[0].access_token137    response = None138    # pull realtime account information for each account associated with the item from plaid139    accounts_fetch_request = AccountsGetRequest(access_token=access_token)140    try:141        response = client.accounts_get(accounts_fetch_request)142    except:143        print("")144    accounts = clean_accounts_data(item[0].pk, response['accounts'])145    # validate and save user accounts146    serializer = AccountSerializer(data=accounts, many=True)147    serializer.is_valid(raise_exception=True)148    serializer.save()149    return "User Accounts saved to DB"150@shared_task151def fetch_transaction_data(item_id):152    # fetching transactions for past 30 days: developer descretion153    start_dt = (datetime.now() - timedelta(days=30))154    end_dt = datetime.now()155    item = Item.objects.filter(item_id=item_id)156    access_token = item[0].access_token157    # fetch transactions associated with user from plaid158    request = TransactionsGetRequest(159        access_token=access_token,160        start_date=start_dt.date(),161        end_date=end_dt.date(),162        options=TransactionsGetRequestOptions()163    )164    try:165        response = client.transactions_get(request)166    except:167        print("")168    transactions = response['transactions']169    # print(transactions)170    transactions = clean_transaction_data(transactions)171    # validate and save transactions to db172    serializer = TransactionSerializer(data=transactions, many=True)173    serializer.is_valid(raise_exception=True)174    serializer.save()175    return "User Transactions saved to DB"...

main.py

Source:main.py

1from labels import Label2from santander_transactions_parser import get_transactions_from_txt_export3from transactions import TransactionsPeriodSummary4from transactions_period_comparator import TransactionsSummaryComparator5import sys6if __name__ == '__main__':7    export_file = sys.argv[1]8    print(f'Processing export file {export_file}...')9    all_time_transactions = get_transactions_from_txt_export(export_file)10    # MONTHLY SUMMARIES11    jan_transactions = TransactionsPeriodSummary('/01/2019', all_time_transactions, TransactionsPeriodSummary.Type.ALL)12    # jan_transactions.print()13    feb_transactions = TransactionsPeriodSummary('/02/2019', all_time_transactions, TransactionsPeriodSummary.Type.ALL)14    # feb_transactions.print()15    mar_transactions = TransactionsPeriodSummary('/03/2019', all_time_transactions, TransactionsPeriodSummary.Type.ALL)16    # mar_transactions.print()17    apr_transactions = TransactionsPeriodSummary('/04/2019', all_time_transactions, TransactionsPeriodSummary.Type.ALL)18    # apr_transactions.print()19    may_transactions = TransactionsPeriodSummary('/05/2019', all_time_transactions, TransactionsPeriodSummary.Type.ALL)20    # may_transactions.print()21    june_transactions = TransactionsPeriodSummary('/06/2019', all_time_transactions, TransactionsPeriodSummary.Type.ALL)22    # june_transactions.print()23    july_transactions = TransactionsPeriodSummary('/07/2019', all_time_transactions, TransactionsPeriodSummary.Type.ALL)24    # july_transactions.print()25    august_transactions = TransactionsPeriodSummary('/08/2019', all_time_transactions, TransactionsPeriodSummary.Type.ALL)26    # august_transactions.print()27    sept_transactions = TransactionsPeriodSummary('/09/2019', all_time_transactions, TransactionsPeriodSummary.Type.ALL)28    # sept_transactions.print()29    oct_transactions = TransactionsPeriodSummary('/10/2019', all_time_transactions, TransactionsPeriodSummary.Type.ALL)30    oct_transactions.print()31    # DETAILS OF A SPECIFIC CATEGORY/LABEL32    for label in Label:33        oct_transactions.print_details_of(label)34    # COMPARISONS35    # TransactionsSummaryComparator(jan_transactions, feb_transactions).print()36    # TransactionsSummaryComparator(feb_transactions, mar_transactions).print()37    # TransactionsSummaryComparator(mar_transactions, apr_transactions).print()38    # TransactionsSummaryComparator(apr_transactions, may_transactions).print()39    # TransactionsSummaryComparator(may_transactions, june_transactions).print()40    # TransactionsSummaryComparator(june_transactions, july_transactions).print()41    # TransactionsSummaryComparator(july_transactions, august_transactions).print()42    # TransactionsSummaryComparator(august_transactions, sept_transactions).print()...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.