KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
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.
LambdaTest Learning Hubs:
- JUnit Tutorial
- TestNG Tutorial
- Webdriver Tutorial
- WebDriverIO Tutorial
- Protractor Tutorial
- Selenium 4 Tutorial
- Jenkins Tutorial
- NUnit Tutorial
- Jest Tutorial
- Playwright Tutorial
- Cypress Tutorial
- PyTest Tutorial
YouTube
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Run mailosaur-python automation tests on LambdaTest cloud grid
Perform automation testing on 3000+ real desktop and mobile devices online.
Try LambdaTest Now !!
Get 100 minutes of automation test minutes FREE!!
Was this article helpful?
