281 lines
8.9 KiB
Python
281 lines
8.9 KiB
Python
"""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()
|