Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def sendWaves(self, recipient, amount, attachment='', txFee=pywaves.DEFAULT_TX_FEE, timestamp=0):
if not self.privateKey:
msg = 'Private key required'
logging.error(msg)
pywaves.throw_error(msg)
elif amount <= 0:
msg = 'Amount must be > 0'
logging.error(msg)
pywaves.throw_error(msg)
elif not pywaves.OFFLINE and self.balance() < amount + txFee:
msg = 'Insufficient Waves balance'
logging.error(msg)
pywaves.throw_error(msg)
else:
if timestamp == 0:
timestamp = int(time.time() * 1000)
sData = b'\4' + \
b'\2' + \
base58.b58decode(self.publicKey) + \
b'\0\0' + \
struct.pack(">Q", timestamp) + \
struct.pack(">Q", amount) + \
struct.pack(">Q", txFee) + \
base58.b58decode(recipient.address) + \
def leaseCancel(self, leaseId, txFee=pywaves.DEFAULT_LEASE_FEE, timestamp=0):
if not self.privateKey:
msg = 'Private key required'
logging.error(msg)
pywaves.throw_error(msg)
elif not pywaves.OFFLINE and self.balance() < txFee:
msg = 'Insufficient Waves balance'
logging.error(msg)
pywaves.throw_error(msg)
else:
if timestamp == 0:
timestamp = int(time.time() * 1000)
sData = b'\x09' + \
base58.b58decode(self.publicKey) + \
struct.pack(">Q", txFee) + \
struct.pack(">Q", timestamp) + \
base58.b58decode(leaseId)
signature = crypto.sign(self.privateKey, sData)
data = json.dumps({
"senderPublicKey": self.publicKey,
"txId": leaseId,
"fee": txFee,
"timestamp": timestamp,
"signature": signature
})
def sendWaves(self, recipient, amount, attachment='', txFee=pywaves.DEFAULT_TX_FEE, timestamp=0):
if not self.privateKey:
msg = 'Private key required'
logging.error(msg)
pywaves.throw_error(msg)
elif amount <= 0:
msg = 'Amount must be > 0'
logging.error(msg)
pywaves.throw_error(msg)
elif not pywaves.OFFLINE and self.balance() < amount + txFee:
msg = 'Insufficient Waves balance'
logging.error(msg)
pywaves.throw_error(msg)
else:
if timestamp == 0:
timestamp = int(time.time() * 1000)
sData = b'\4' + \
b'\2' + \
base58.b58decode(self.publicKey) + \
b'\0\0' + \
struct.pack(">Q", timestamp) + \
struct.pack(">Q", amount) + \
struct.pack(">Q", txFee) + \
base58.b58decode(recipient.address) + \
struct.pack(">H", len(attachment)) + \
crypto.str2bytes(attachment)
signature = crypto.sign(self.privateKey, sData)
data = json.dumps({
def issueAsset(self, name, description, quantity, decimals=0, reissuable=False, txFee=pywaves.DEFAULT_ASSET_FEE):
if not self.privateKey:
msg = 'Private key required'
logging.error(msg)
pywaves.throw_error(msg)
elif len(name) < 4 or len(name) > 16:
msg = 'Asset name must be between 4 and 16 characters long'
logging.error(msg)
pywaves.throw_error(msg)
else:
timestamp = int(time.time() * 1000)
sData = b'\3' + \
base58.b58decode(self.publicKey) + \
struct.pack(">H", len(name)) + \
crypto.str2bytes(name) + \
struct.pack(">H", len(description)) + \
crypto.str2bytes(description) + \
struct.pack(">Q", quantity) + \
struct.pack(">B", decimals) + \
(b'\1' if reissuable else b'\0') + \
struct.pack(">Q", txFee) + \
struct.pack(">Q", timestamp)
signature=crypto.sign(self.privateKey, sData)
data = json.dumps({
"senderPublicKey": self.publicKey,
def lease(self, recipient, amount, txFee=pywaves.DEFAULT_LEASE_FEE, timestamp=0):
if not self.privateKey:
msg = 'Private key required'
logging.error(msg)
pywaves.throw_error(msg)
elif amount <= 0:
msg = 'Amount must be > 0'
logging.error(msg)
pywaves.throw_error(msg)
elif not pywaves.OFFLINE and self.balance() < amount + txFee:
msg = 'Insufficient Waves balance'
logging.error(msg)
pywaves.throw_error(msg)
else:
if timestamp == 0:
timestamp = int(time.time() * 1000)
sData = b'\x08' + \
base58.b58decode(self.publicKey) + \
base58.b58decode(recipient.address) + \
struct.pack(">Q", amount) + \
struct.pack(">Q", txFee) + \
struct.pack(">Q", timestamp)
signature = crypto.sign(self.privateKey, sData)
data = json.dumps({
"senderPublicKey": self.publicKey,
"recipient": recipient.address,
"amount": amount,
"fee": txFee,
"timestamp": timestamp,
def sendWaves(self, recipient, amount, attachment='', txFee=pywaves.DEFAULT_TX_FEE, timestamp=0):
if not self.privateKey:
msg = 'Private key required'
logging.error(msg)
pywaves.throw_error(msg)
elif amount <= 0:
msg = 'Amount must be > 0'
logging.error(msg)
pywaves.throw_error(msg)
elif not pywaves.OFFLINE and self.balance() < amount + txFee:
msg = 'Insufficient Waves balance'
logging.error(msg)
pywaves.throw_error(msg)
else:
if timestamp == 0:
timestamp = int(time.time() * 1000)
sData = b'\4' + \
b'\2' + \
base58.b58decode(self.publicKey) + \
def cancelOrder(self, assetPair, order):
if not pywaves.OFFLINE:
if order.status() == 'Filled':
msg = "Order already filled"
logging.error(msg)
pywaves.throw_error(msg)
elif not order.status():
msg = "Order not found"
logging.error(msg)
pywaves.throw_error(msg)
sData = base58.b58decode(self.publicKey) + \
base58.b58decode(order.orderId)
signature = crypto.sign(self.privateKey, sData)
data = json.dumps({
"sender": self.publicKey,
"orderId": order.orderId,
"signature": signature
})
req = pywaves.wrapper('/matcher/orderbook/%s/%s/cancel' % ('WAVES' if assetPair.asset1.assetId=='' else assetPair.asset1.assetId, 'WAVES' if assetPair.asset2.assetId=='' else assetPair.asset2.assetId), data, host=pywaves.MATCHER)
if pywaves.OFFLINE:
return req
def lease(self, recipient, amount, txFee=pywaves.DEFAULT_LEASE_FEE, timestamp=0):
if not self.privateKey:
msg = 'Private key required'
logging.error(msg)
pywaves.throw_error(msg)
elif amount <= 0:
msg = 'Amount must be > 0'
logging.error(msg)
pywaves.throw_error(msg)
elif not pywaves.OFFLINE and self.balance() < amount + txFee:
msg = 'Insufficient Waves balance'
logging.error(msg)
pywaves.throw_error(msg)
else:
if timestamp == 0:
timestamp = int(time.time() * 1000)
sData = b'\x08' + \
base58.b58decode(self.publicKey) + \
base58.b58decode(recipient.address) + \
struct.pack(">Q", amount) + \
struct.pack(">Q", txFee) + \
def candles(self, *args):
if len(args)==2:
timeframe = args[0]
limit = args[1]
if timeframe not in pywaves.VALID_TIMEFRAMES:
msg = 'Invalid timeframe'
pywaves.throw_error(msg)
return logging.error(msg)
elif limit > 0 and limit <= pywaves.MAX_WDF_REQUEST:
return self._getMarketData('/api/candles', '%d/%d' % (timeframe, limit))
else:
msg = 'Invalid request. Limit must be >0 and <= 100'
pywaves.throw_error(msg)
return logging.error(msg)
elif len(args)==3:
timeframe = args[0]
fromTimestamp = args[1]
toTimestamp = args[2]
if timeframe not in pywaves.VALID_TIMEFRAMES:
msg = 'Invalid timeframe'
pywaves.throw_error(msg)
return logging.error(msg)
else:
return self._getMarketData('/api/candles', '%d/%d/%d' % (timeframe, fromTimestamp, toTimestamp))