"""Bank account transaction processor. Reads account state and a list of transactions from JSON files, validates and applies each transaction, then writes updated account state and a transaction log (accepted / rejected) to output files. """ import json from typing import TypedDict, Optional # --------------------------------------------------------------------------- # Explicit data model -- defines the exact shape of every data structure # --------------------------------------------------------------------------- class Account(TypedDict): """A bank account with its current state.""" account_id: str holder: str balance: float currency: str status: str # "active" or "frozen" class Transaction(TypedDict, total=False): """A financial transaction to be processed. Fields marked total=False are optional (e.g. to_account_id only exists for transfers; status/reason are added during processing). """ id: str type: str # "deposit", "withdrawal", or "transfer" account_id: str amount: float description: str to_account_id: str # only for transfers status: str # added after processing: "accepted" / "rejected" reason: str # added on rejection # --------------------------------------------------------------------------- # Constants # --------------------------------------------------------------------------- ACCOUNTS_INPUT = "accounts.json" TRANSACTIONS_INPUT = "transactions.json" ACCOUNTS_OUTPUT = "accounts_updated_good.json" TRANSACTION_LOG_OUTPUT = "transaction_log_good.json" ACTIVE_STATUS = "active" # --------------------------------------------------------------------------- # File I/O # --------------------------------------------------------------------------- def load_json(file_path: str) -> dict: """Read and parse a JSON file, returning the parsed data.""" with open(file_path, "r", encoding="utf-8") as file_handle: return json.load(file_handle) def save_json(file_path: str, data: dict) -> None: """Write data to a JSON file with readable indentation.""" with open(file_path, "w", encoding="utf-8") as file_handle: json.dump(data, file_handle, indent=2, ensure_ascii=False) def load_accounts(file_path: str) -> list[Account]: """Load and return the list of accounts from a JSON file.""" data = load_json(file_path) return data["accounts"] def load_transactions(file_path: str) -> list[Transaction]: """Load and return the list of transactions from a JSON file.""" data = load_json(file_path) return data["transactions"] # --------------------------------------------------------------------------- # Account lookup # --------------------------------------------------------------------------- def find_account(accounts: list[Account], account_id: str) -> Optional[Account]: """Find an account by its ID. Returns the account dict or None.""" for account in accounts: if account["account_id"] == account_id: return account return None # --------------------------------------------------------------------------- # Validation # --------------------------------------------------------------------------- def validate_common( account: Optional[Account], amount: float, ) -> Optional[str]: """Run validations shared by all transaction types. Returns an error message string, or None if valid. """ if account is None: return "account not found" if account["status"] != ACTIVE_STATUS: return f"account is {account['status']}" if amount is None or amount <= 0: return "amount must be positive" return None # --------------------------------------------------------------------------- # Transaction handlers -- one function per transaction type # --------------------------------------------------------------------------- def process_deposit( accounts: list[Account], transaction: Transaction, ) -> tuple[bool, str]: """Apply a deposit transaction. Returns (success, reason).""" account = find_account(accounts, transaction["account_id"]) error = validate_common(account, transaction["amount"]) if error: return False, error account["balance"] += transaction["amount"] return True, "accepted" def process_withdrawal( accounts: list[Account], transaction: Transaction, ) -> tuple[bool, str]: """Apply a withdrawal transaction. Returns (success, reason).""" account = find_account(accounts, transaction["account_id"]) error = validate_common(account, transaction["amount"]) if error: return False, error if account["balance"] < transaction["amount"]: return False, "insufficient funds" account["balance"] -= transaction["amount"] return True, "accepted" def process_transfer( accounts: list[Account], transaction: Transaction, ) -> tuple[bool, str]: """Apply a transfer between two accounts. Returns (success, reason).""" source = find_account(accounts, transaction["account_id"]) error = validate_common(source, transaction["amount"]) if error: return False, f"source: {error}" target_id = transaction.get("to_account_id", "") target = find_account(accounts, target_id) if target is None: return False, "target account not found" if target["status"] != ACTIVE_STATUS: return False, f"target account is {target['status']}" if source["balance"] < transaction["amount"]: return False, "insufficient funds" source["balance"] -= transaction["amount"] target["balance"] += transaction["amount"] return True, "accepted" TRANSACTION_HANDLERS = { "deposit": process_deposit, "withdrawal": process_withdrawal, "transfer": process_transfer, } # --------------------------------------------------------------------------- # Processing # --------------------------------------------------------------------------- def process_all_transactions( accounts: list[Account], transactions: list[Transaction], ) -> tuple[list[Transaction], list[Transaction]]: """Process a list of transactions against the account state. Returns two lists: (accepted_transactions, rejected_transactions). Each transaction is augmented with 'status' and optionally 'reason'. """ accepted: list[Transaction] = [] rejected: list[Transaction] = [] for transaction in transactions: transaction_type = transaction.get("type", "") handler = TRANSACTION_HANDLERS.get(transaction_type) if handler is None: transaction["status"] = "rejected" transaction["reason"] = f"unknown transaction type '{transaction_type}'" rejected.append(transaction) continue success, reason = handler(accounts, transaction) if success: transaction["status"] = "accepted" accepted.append(transaction) else: transaction["status"] = "rejected" transaction["reason"] = reason rejected.append(transaction) return accepted, rejected # --------------------------------------------------------------------------- # Output # --------------------------------------------------------------------------- def print_results( accounts: list[Account], accepted: list[Transaction], rejected: list[Transaction], ) -> None: """Print a human-readable summary to the console.""" print("=== UPDATED ACCOUNTS ===") for account in accounts: print( f" {account['account_id']} {account['holder']}: " f"{account['balance']:.2f} {account['currency']} " f"({account['status']})" ) print(f"\n=== ACCEPTED TRANSACTIONS ({len(accepted)}) ===") for txn in accepted: print( f" {txn['id']} {txn['type']:12s} {txn['amount']:>10.2f} " f"{txn.get('description', '')}" ) print(f"\n=== REJECTED TRANSACTIONS ({len(rejected)}) ===") for txn in rejected: print( f" {txn['id']} {txn['type']:12s} {txn['amount']:>10.2f} " f"Reason: {txn.get('reason', 'unknown')}" ) # --------------------------------------------------------------------------- # Main # --------------------------------------------------------------------------- def main() -> None: """Load data, process transactions, print and save results.""" accounts: list[Account] = load_accounts(ACCOUNTS_INPUT) transactions: list[Transaction] = load_transactions(TRANSACTIONS_INPUT) accepted, rejected = process_all_transactions(accounts, transactions) print_results(accounts, accepted, rejected) save_json(ACCOUNTS_OUTPUT, {"accounts": accounts}) save_json(TRANSACTION_LOG_OUTPUT, { "accepted": accepted, "rejected": rejected, }) print(f"\nOutput written to {ACCOUNTS_OUTPUT} and {TRANSACTION_LOG_OUTPUT}") if __name__ == "__main__": main()