def begin(self, request=None):
"""Return new request specific transaction
:param request: current request
"""
if request is None:
request = get_current_request()
user = get_authenticated_user_id(request)
txn = getattr(request, '_txn', None)
if txn is not None:
txn.abort()
txn = request._txn = transaction.Transaction(self._synchs, self)
if user is not None:
txn.user = user
_new_transaction(txn, self._synchs)
request._txn_time = time.time()
return txn
# with
python类Transaction()的实例源码
def gettransactions():
global privapi
global transactions
window.lsttransactions.delete(0, tk.END)
debugout("Requesting Transaction History...")
# The API Doesn't allow for Null type or "All" so we have to pull both dicts and combine them.
txlist = privapi.gettransactions("Deposit")
# After the first API call we'll check for an error string.
if isinstance(txlist, str):
debugout(txlist)
else:
txlist = txlist + privapi.gettransactions("Withdraw")
itx = 0
for tx in txlist:
tid = tx["Id"]
currency = tx["Currency"]
txid = tx["TxId"]
ttype = tx["Type"]
amount = tx["Amount"]
fee = tx["Fee"]
status = tx["Status"]
confirmations = tx["Confirmations"]
timestamp = tx["Timestamp"]
address = tx["Address"]
tmptx = transaction.Transaction(tid, currency, txid, ttype, amount, fee, status, confirmations, timestamp,
address)
key = tmptx.tostring()
transactions[key] = tmptx
window.lsttransactions.insert(tk.END, key)
itx = itx + 1
debugout("Transaction History data received, " + str(itx) + " transactions found.")
# Cancel a single trade via it's OrderId
def __init__(self, filepath):
with open(filepath, "r") as f:
lines = []
index_line = []
count = 0
for line in f:
items = line.strip().split(",")
if len(items) > 5:
if count > 0:
lines.append(items)
else:
index_line = items
count += 1
for line in lines:
transaction = Transaction()
transaction.process_values(index_line, line)
self.transactions.append(transaction)
print("{0} records found!\n".format(count))
f.close()
def __init__(self, filepath):
with open(filepath, "r") as f:
lines = []
index_line = []
count = 0
for line in f:
items = line.strip().split(",")
if len(items) > 5:
if count > 0:
lines.append(items)
else:
index_line = items[2:]
count += 1
for line in lines:
transaction = Transaction()
transaction.process_values(index_line, line[2:])
cat = int(line[0])
transaction.set_category(cat)
if cat not in self.transactions:
self.transactions[cat] = [transaction]
else:
self.transactions[cat].append(transaction)
print("{0} records found!\n".format(count))
f.close()
def select_user_and_add_transaction(self):
self.prompt_user_selection().add_transaction(transaction.Transaction(input("Amount of transaction:")))
def select_user_and_add_transaction(self):
"""
Select a user and add a transaction with custom amount.
"""
def add_transaction(to_user):
print("Amount of transaction:")
amount = input()
new_transaction = transaction.Transaction(amount)
to_user.add_transaction(new_transaction)
try:
selected_user = self.prompt_user_selection()
add_transaction(selected_user)
except ValueError:
print("No changes made.")
def select_user_and_add_transaction(self):
"""
Select a user and add a transaction with custom amount.
"""
self.prompt_user_selection().add_transaction(transaction.Transaction(input("Amount of transaction:")))
def get(self, request=None):
"""Return the current request specific transaction
:param request: current request
"""
if request is None:
request = get_current_request()
if getattr(request, '_txn', None) is None:
request._txn = transaction.Transaction(self._synchs, self)
request._txn_time = time.time()
return request._txn
# ITransactionManager
