From ac4902e902262beffdf003f305afb6dbef57267f Mon Sep 17 00:00:00 2001
From: Vincent Texier <vit@free.fr>
Date: Sun, 30 Oct 2016 20:41:45 +0100
Subject: [PATCH] Add examples scripts
---
examples/create_and_publish_identity.py | 102 ++++++++++++++
examples/create_public_key.py | 30 +++++
examples/request_data.py | 65 +++++++++
examples/save_revoke_document.py | 131 ++++++++++++++++++
examples/send_certification.py | 141 ++++++++++++++++++++
examples/send_membership.py | 151 +++++++++++++++++++++
examples/send_transaction.py | 169 ++++++++++++++++++++++++
7 files changed, 789 insertions(+)
create mode 100644 examples/create_and_publish_identity.py
create mode 100644 examples/create_public_key.py
create mode 100644 examples/request_data.py
create mode 100644 examples/save_revoke_document.py
create mode 100644 examples/send_certification.py
create mode 100644 examples/send_membership.py
create mode 100644 examples/send_transaction.py
diff --git a/examples/create_and_publish_identity.py b/examples/create_and_publish_identity.py
new file mode 100644
index 00000000..0a7bcf23
--- /dev/null
+++ b/examples/create_and_publish_identity.py
@@ -0,0 +1,102 @@
+import asyncio
+import aiohttp
+
+import duniterpy.api.bma as bma
+from duniterpy.documents import BMAEndpoint, BlockUID, SelfCertification
+from duniterpy.key import SigningKey
+
+
+# CONFIG #######################################
+
+# You can either use a complete defined endpoint : [NAME_OF_THE_API] [DOMAIN] [IPv4] [IPv6] [PORT]
+# or the simple definition : [NAME_OF_THE_API] [DOMAIN] [PORT]
+# Here we use the BASIC_MERKLED_API
+BMA_ENDPOINT = "BASIC_MERKLED_API cgeek.fr 9330"
+
+# Credentials should be prompted or kept in a separate secure file
+# create a file with the salt on the first line and the password on the second line
+# the script will load them from the file
+FROM_CREDENTIALS_FILE = "/home/username/.credentials.txt"
+
+# Your unique identifier in the Web of Trust
+UID = "MyIdentity"
+
+################################################
+
+# Latest duniter-python-api is asynchronous and you have to create an aiohttp session to send request
+# ( http://pythonhosted.org/aiohttp )
+AIOHTTP_SESSION = aiohttp.ClientSession()
+
+async def get_current_block():
+ """
+ Get the current block data
+
+ :rtype: dict
+ """
+ # Here we request for the path blockchain/block/N
+ return await bma.blockchain.Current(BMAEndpoint.from_inline(BMA_ENDPOINT).conn_handler()) \
+ .get(AIOHTTP_SESSION)
+
+
+def get_identity_document(current_block, uid, salt, password):
+ """
+ Get an Identity document
+
+ :param dict current_block: Current block data
+ :param str uid: Unique Identifier
+ :param str salt: Passphrase of the account
+ :param str password: Password of the account
+
+ :rtype: SelfCertification
+ """
+
+ # get current block BlockStamp
+ timestamp = BlockUID(current_block['number'], current_block['hash'])
+
+ # create keys from credentials
+ key = SigningKey(salt, password)
+
+ # create identity document
+ identity = SelfCertification(
+ version=2,
+ currency=current_block['currency'],
+ pubkey=key.pubkey,
+ uid=uid,
+ ts=timestamp,
+ signature=None
+ )
+
+ # sign document
+ identity.sign([key])
+
+ return identity
+
+async def main():
+ """
+ Main code
+ """
+ # capture current block to get version and currency and blockstamp
+ current_block = await get_current_block()
+
+ # load credentials from a text file
+ salt, password = open(FROM_CREDENTIALS_FILE).readlines()
+
+ # cleanup newlines
+ salt, password = salt.strip(), password.strip()
+
+ # create our signed identity document
+ identity = get_identity_document(current_block, UID, salt, password)
+
+ # send the identity document to the node
+ data = {identity: identity.signed_raw()}
+ response = await bma.wot.Add(BMAEndpoint.from_inline(BMA_ENDPOINT).conn_handler()).post(AIOHTTP_SESSION, **data)
+
+ print(response)
+
+ response.close()
+
+with AIOHTTP_SESSION:
+
+ # Latest duniter-python-api is asynchronous and you have to use asyncio, an asyncio loop and a "as" on the data.
+ # ( https://docs.python.org/3/library/asyncio.html )
+ asyncio.get_event_loop().run_until_complete(main())
diff --git a/examples/create_public_key.py b/examples/create_public_key.py
new file mode 100644
index 00000000..ea7ab135
--- /dev/null
+++ b/examples/create_public_key.py
@@ -0,0 +1,30 @@
+from duniterpy.key import SigningKey
+
+# CONFIG #######################################
+
+# CREDENTIALS in Duniter are a couple of strings:
+# - A secret pass-phrase
+# - A password
+# They create a seed which create keys (some are private and one is public)
+
+# Credentials should be prompted or kept in a separate secure file
+# create a file with the salt on the first line and the password on the second line
+# the script will load them from the file
+CREDENTIALS_FILE_PATH = "/home/username/.credentials.txt"
+
+################################################
+
+if __name__ == '__main__':
+
+ # Load your credentials from a text file
+ salt, password = open(CREDENTIALS_FILE_PATH).readlines()
+ # Cleanup newlines
+ salt, password = salt.strip(), password.strip()
+
+ # Create key object
+ key = SigningKey(salt, password)
+
+ # Display your public key
+ print("Public key for your credentials: %s" % key.pubkey)
+
+
diff --git a/examples/request_data.py b/examples/request_data.py
new file mode 100644
index 00000000..cbbc5a71
--- /dev/null
+++ b/examples/request_data.py
@@ -0,0 +1,65 @@
+import asyncio
+import aiohttp
+from aiohttp.client_reqrep import ClientResponse
+import duniterpy.api.bma as bma
+from duniterpy.documents import BMAEndpoint
+
+# CONFIG #######################################
+
+# You can either use a complete defined endpoint : [NAME_OF_THE_API] [DOMAIN] [IPv4] [IPv6] [PORT]
+# or the simple definition : [NAME_OF_THE_API] [DOMAIN] [PORT]
+# Here we use the BASIC_MERKLED_API
+BMA_ENDPOINT = "BASIC_MERKLED_API cgeek.fr 9330"
+
+################################################
+
+# Latest duniter-python-api is asynchronous and you have to create an aiohttp session to send request
+# ( http://pythonhosted.org/aiohttp )
+AIOHTTP_SESSION = aiohttp.ClientSession()
+
+async def get_summary_info():
+ """
+ Get the node info
+
+ :rtype: ClientResponse
+ """
+ # Here we request for the path /node/summary
+ return await bma.node.Summary(BMAEndpoint.from_inline(BMA_ENDPOINT).conn_handler()).get(AIOHTTP_SESSION)
+
+async def get_current_block():
+ """
+ Get the current block data
+
+ :rtype: ClientResponse
+ """
+ # Here we request for the path blockchain/current
+ return await bma.blockchain.Current(BMAEndpoint.from_inline(BMA_ENDPOINT).conn_handler()) \
+ .get(AIOHTTP_SESSION)
+
+async def get_block(block_number):
+ """
+ Get the a block data
+ :param: int block_number Number of the block
+
+ :rtype: ClientResponse
+ """
+ # Here we request for the path blockchain/block/N
+ return await bma.blockchain.Block(BMAEndpoint.from_inline(BMA_ENDPOINT).conn_handler(), block_number)\
+ .get(AIOHTTP_SESSION)
+
+
+async def main():
+ """
+ Main code
+ """
+ print(await get_summary_info())
+
+ print(await get_current_block())
+
+ print(await get_block(0))
+
+with AIOHTTP_SESSION:
+
+ # Latest duniter-python-api is asynchronous and you have to use asyncio, an asyncio loop and a "as" on the data.
+ # ( https://docs.python.org/3/library/asyncio.html )
+ asyncio.get_event_loop().run_until_complete(main())
diff --git a/examples/save_revoke_document.py b/examples/save_revoke_document.py
new file mode 100644
index 00000000..0731217b
--- /dev/null
+++ b/examples/save_revoke_document.py
@@ -0,0 +1,131 @@
+import asyncio
+import aiohttp
+import duniterpy.api.bma as bma
+from duniterpy.documents import BMAEndpoint, BlockUID, SelfCertification
+from duniterpy.documents import Revocation
+from duniterpy.key import SigningKey
+
+# CONFIG #######################################
+
+# You can either use a complete defined endpoint : [NAME_OF_THE_API] [DOMAIN] [IPv4] [IPv6] [PORT]
+# or the simple definition : [NAME_OF_THE_API] [DOMAIN] [PORT]
+# Here we use the BASIC_MERKLED_API
+BMA_ENDPOINT = "BASIC_MERKLED_API cgeek.fr 9330"
+
+# Credentials should be prompted or kept in a separate secure file
+# create a file with the salt on the first line and the password on the second line
+# the script will load them from the file
+CREDENTIALS_FILE_PATH = "/home/username/.credentials.txt"
+
+# Public key of the revoked identity account
+PUBKEY = "XXXXXXXX"
+
+# WARNING : Hide this file in a safe and secure place
+# If one day you forget your credentials,
+# you'll have to use your private key instead
+REVOKE_DOCUMENT_FILE_PATH = "/home/username/duniter_account_revoke_document.txt"
+
+################################################
+
+# Latest duniter-python-api is asynchronous and you have to create an aiohttp session to send request
+# ( http://pythonhosted.org/aiohttp )
+AIOHTTP_SESSION = aiohttp.ClientSession()
+
+# Current protocole version
+PROTOCOL_VERSION = 2
+
+async def get_current_block():
+ """
+ Get the current block data
+
+ :rtype: dict
+ """
+ # Here we request for the path blockchain/current
+ return await bma.blockchain.Current(BMAEndpoint.from_inline(BMA_ENDPOINT).conn_handler()) \
+ .get(AIOHTTP_SESSION)
+
+async def get_self_certification_document(currency, pubkey):
+ """
+ Get the SelfCertification document of the pubkey
+
+ :param str currency: Currency name
+ :param str pubkey: Public key
+
+ :rtype: SelfCertification
+ """
+ # Here we request for the path wot/lookup/pubkey
+ lookup_data = await bma.wot.Lookup(BMAEndpoint.from_inline(BMA_ENDPOINT).conn_handler(), pubkey)\
+ .get(AIOHTTP_SESSION)
+
+ # init vars
+ uid = None
+ timestamp = BlockUID.empty()
+ signature = None
+
+ # parse results
+ for result in lookup_data['results']:
+ if result["pubkey"] == pubkey:
+ uids = result['uids']
+ for uid_data in uids:
+ # capture data
+ timestamp = BlockUID.from_str(uid_data["meta"]["timestamp"])
+ uid = uid_data["uid"]
+ signature = uid_data["self"]
+
+ # return self-certification document
+ return SelfCertification(
+ version=PROTOCOL_VERSION,
+ currency=currency,
+ pubkey=pubkey,
+ uid=uid,
+ ts=timestamp,
+ signature=signature
+ )
+
+
+async def get_revoke_document(self_certification, salt, password):
+ """
+ Generate account revokation document for given identity
+
+ :param SelfCertification self_certification: Self Certification of the identity
+ :param str salt: Salt
+ :param str password: Password
+
+ :return: the revokation document
+ :rtype: duniterpy.documents.certification.Revocation
+ """
+ document = Revocation(PROTOCOL_VERSION, self_certification.currency, self_certification.pubkey, "")
+
+ key = SigningKey(salt, password)
+ document.sign(self_certification, [key])
+ return document.signed_raw(self_certification)
+
+async def main():
+ """
+ Main code
+ """
+ # capture current block to get currency name
+ current_block = await get_current_block()
+
+ # create our SelfCertification document to sign the revoke document
+ self_cert_document = await get_self_certification_document(current_block['currency'], PUBKEY)
+
+ # load credentials from a text file
+ salt, password = open(CREDENTIALS_FILE_PATH).readlines()
+
+ # get the revoke document
+ revoke_document = await get_revoke_document(self_cert_document, salt, password)
+
+ # save revoke document in a file
+ fp = open(REVOKE_DOCUMENT_FILE_PATH, 'w')
+ fp.write(revoke_document)
+ fp.close()
+
+ # document saved
+ print("Revoke document saved in %s" % REVOKE_DOCUMENT_FILE_PATH)
+
+with AIOHTTP_SESSION:
+
+ # Latest duniter-python-api is asynchronous and you have to use asyncio, an asyncio loop and a "as" on the data.
+ # ( https://docs.python.org/3/library/asyncio.html )
+ asyncio.get_event_loop().run_until_complete(main())
diff --git a/examples/send_certification.py b/examples/send_certification.py
new file mode 100644
index 00000000..4f9812a9
--- /dev/null
+++ b/examples/send_certification.py
@@ -0,0 +1,141 @@
+import asyncio
+import aiohttp
+import duniterpy.api.bma as bma
+from duniterpy.documents import BMAEndpoint, BlockUID, SelfCertification, Certification
+from duniterpy.key import SigningKey
+
+
+# CONFIG #######################################
+
+# You can either use a complete defined endpoint : [NAME_OF_THE_API] [DOMAIN] [IPv4] [IPv6] [PORT]
+# or the simple definition : [NAME_OF_THE_API] [DOMAIN] [PORT]
+# Here we use the BASIC_MERKLED_API
+BMA_ENDPOINT = "BASIC_MERKLED_API cgeek.fr 9330"
+
+# Public key of the certifier
+FROM_PUBKEY = "XXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXX"
+
+# Credentials should be prompted or kept in a separate secure file
+# create a file with the salt on the first line and the password on the second line
+# the script will load them from the file
+FROM_CREDENTIALS_FILE = "/home/username/.credentials.txt"
+
+# Public key to certified
+TO_PUBKEY = "YYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYY"
+
+################################################
+
+
+# Latest duniter-python-api is asynchronous and you have to create an aiohttp session to send request
+# ( http://pythonhosted.org/aiohttp )
+AIOHTTP_SESSION = aiohttp.ClientSession()
+
+async def get_current_block():
+ """
+ Get the current block data
+
+ :rtype: dict
+ """
+ # Here we request for the path blockchain/block/N
+ return await bma.blockchain.Current(BMAEndpoint.from_inline(BMA_ENDPOINT).conn_handler()) \
+ .get(AIOHTTP_SESSION)
+
+async def get_identity_document(current_block, pubkey):
+ """
+ Get the identity document of the pubkey
+
+ :param dict current_block: Current block data
+ :param str pubkey: Public key
+
+ :rtype: SelfCertification
+ """
+ # Here we request for the path wot/lookup/pubkey
+ lookup_data = await bma.wot.Lookup(BMAEndpoint.from_inline(BMA_ENDPOINT).conn_handler(), pubkey)\
+ .get(AIOHTTP_SESSION)
+
+ # init vars
+ uid = None
+ timestamp = BlockUID.empty()
+ signature = None
+
+ # parse results
+ for result in lookup_data['results']:
+ if result["pubkey"] == pubkey:
+ uids = result['uids']
+ for uid_data in uids:
+ # capture data
+ timestamp = BlockUID.from_str(uid_data["meta"]["timestamp"])
+ uid = uid_data["uid"]
+ signature = uid_data["self"]
+
+ # return self-certification document
+ return SelfCertification(
+ version=2,
+ currency=current_block['currency'],
+ pubkey=pubkey,
+ uid=uid,
+ ts=timestamp,
+ signature=signature
+ )
+
+
+def get_certification_document(current_block, self_cert_document, from_pubkey, salt, password):
+ """
+ Create and return a Certification document
+
+ :param dict current_block: Current block data
+ :param SelfCertification self_cert_document: SelfCertification document
+ :param str from_pubkey: Pubkey of the certifier
+ :param str salt: Secret salt (DO NOT SHOW IT ANYWHERE, IT IS SECRET !!!)
+ :param str password: Secret password (DO NOT SHOW IT ANYWHERE, IT IS SECRET !!!)
+
+ :rtype: Certification
+ """
+ # construct Certification Document
+ certification = Certification(
+ version=2,
+ currency=current_block['currency'],
+ pubkey_from=from_pubkey,
+ pubkey_to=self_cert_document.pubkey,
+ timestamp=BlockUID(current_block['number'], current_block['hash']),
+ signature=""
+ )
+ # sign document
+ key = SigningKey(salt, password)
+ certification.sign(self_cert_document, [key])
+
+ return certification
+
+async def main():
+ """
+ Main code
+ """
+ # capture current block to get version and currency and blockstamp
+ current_block = await get_current_block()
+
+ # create our SelfCertification document to sign the Certification document
+ identity = await get_identity_document(current_block, TO_PUBKEY)
+
+ # load credentials from a text file
+ salt, password = open(FROM_CREDENTIALS_FILE).readlines()
+
+ # cleanup newlines
+ salt, password = salt.strip(), password.strip()
+
+ # send the Certification document to the node
+ certification = get_certification_document(current_block, identity, FROM_PUBKEY, salt, password)
+
+ # Here we request for the path wot/certify
+ data = {'cert': certification.signed_raw(identity)}
+ response = await bma.wot.Certify(BMAEndpoint.from_inline(BMA_ENDPOINT).conn_handler()) \
+ .post(AIOHTTP_SESSION, **data)
+
+ print(response)
+
+ response.close()
+
+with AIOHTTP_SESSION:
+
+ # Latest duniter-python-api is asynchronous and you have to use asyncio, an asyncio loop and a "as" on the data.
+ # ( https://docs.python.org/3/library/asyncio.html )
+ asyncio.get_event_loop().run_until_complete(main())
diff --git a/examples/send_membership.py b/examples/send_membership.py
new file mode 100644
index 00000000..b0d679e0
--- /dev/null
+++ b/examples/send_membership.py
@@ -0,0 +1,151 @@
+import asyncio
+import aiohttp
+
+import duniterpy.api.bma as bma
+from duniterpy.documents import BMAEndpoint, BlockUID, SelfCertification, Membership
+from duniterpy.key import SigningKey
+
+
+# CONFIG #######################################
+
+# You can either use a complete defined endpoint : [NAME_OF_THE_API] [DOMAIN] [IPv4] [IPv6] [PORT]
+# or the simple definition : [NAME_OF_THE_API] [DOMAIN] [PORT]
+# Here we use the BASIC_MERKLED_API
+BMA_ENDPOINT = "BASIC_MERKLED_API cgeek.fr 9330"
+
+# Credentials should be prompted or kept in a separate secure file
+# create a file with the salt on the first line and the password on the second line
+# the script will load them from the file
+FROM_CREDENTIALS_FILE = "/home/username/.credentials.txt"
+
+# Your unique identifier in the Web of Trust
+UID = "MyIdentity"
+
+################################################
+
+# Latest duniter-python-api is asynchronous and you have to create an aiohttp session to send request
+# ( http://pythonhosted.org/aiohttp )
+AIOHTTP_SESSION = aiohttp.ClientSession()
+
+async def get_current_block():
+ """
+ Get the current block data
+
+ :rtype: dict
+ """
+ # Here we request for the path blockchain/block/N
+ return await bma.blockchain.Current(BMAEndpoint.from_inline(BMA_ENDPOINT).conn_handler()) \
+ .get(AIOHTTP_SESSION)
+
+
+def get_identity_document(current_block, uid, salt, password):
+ """
+ Get an Identity document
+
+ :param dict current_block: Current block data
+ :param str uid: Unique Identifier
+ :param str salt: Passphrase of the account
+ :param str password: Password of the account
+
+ :rtype: SelfCertification
+ """
+
+ # get current block BlockStamp
+ timestamp = BlockUID(current_block['number'], current_block['hash'])
+
+ # create keys from credentials
+ key = SigningKey(salt, password)
+
+ # create identity document
+ identity = SelfCertification(
+ version=2,
+ currency=current_block['currency'],
+ pubkey=key.pubkey,
+ uid=uid,
+ ts=timestamp,
+ signature=None
+ )
+
+ # sign document
+ identity.sign([key])
+
+ return identity
+
+
+def get_membership_document(type, current_block, identity, salt, password):
+ """
+ Get a Membership document
+
+ :param str type: "IN" to ask for membership or "OUT" to cancel membership
+ :param dict current_block: Current block data
+ :param SelfCertification identity: Identity document
+ :param str salt: Passphrase of the account
+ :param str password: Password of the account
+
+ :rtype: Membership
+ """
+
+ # get current block BlockStamp
+ timestamp = BlockUID(current_block['number'], current_block['hash'])
+
+ # create keys from credentials
+ key = SigningKey(salt, password)
+
+ # create identity document
+ membership = Membership(
+ version=2,
+ currency=current_block['currency'],
+ issuer=key.pubkey,
+ membership_ts=timestamp,
+ membership_type=type,
+ uid=identity.uid,
+ identity_ts=identity.timestamp,
+ signature=None
+ )
+
+ # sign document
+ membership.sign([key])
+
+ return membership
+
+async def main():
+ """
+ Main code
+ """
+ # capture current block to get version and currency and blockstamp
+ current_block = await get_current_block()
+
+ # load credentials from a text file
+ salt, password = open(FROM_CREDENTIALS_FILE).readlines()
+
+ # cleanup newlines
+ salt, password = salt.strip(), password.strip()
+
+ # create our signed identity document
+ identity = get_identity_document(current_block, UID, salt, password)
+
+ # create a membership demand document
+ membership = get_membership_document("IN", current_block, identity, salt, password)
+
+ # send the identity document to the node
+ data = {identity: identity.signed_raw()}
+ response = await bma.wot.Add(BMAEndpoint.from_inline(BMA_ENDPOINT).conn_handler())\
+ .post(AIOHTTP_SESSION, **data)
+
+ print(response)
+ response.close()
+
+ # send the membership document to the node
+ data = {membership: membership.signed_raw()}
+ response = await bma.blockchain.Membership(BMAEndpoint.from_inline(BMA_ENDPOINT).conn_handler())\
+ .post(AIOHTTP_SESSION, **data)
+
+ print(response)
+ response.close()
+
+
+with AIOHTTP_SESSION:
+
+ # Latest duniter-python-api is asynchronous and you have to use asyncio, an asyncio loop and a "as" on the data.
+ # ( https://docs.python.org/3/library/asyncio.html )
+ asyncio.get_event_loop().run_until_complete(main())
diff --git a/examples/send_transaction.py b/examples/send_transaction.py
new file mode 100644
index 00000000..c39d06e9
--- /dev/null
+++ b/examples/send_transaction.py
@@ -0,0 +1,169 @@
+import asyncio
+
+import aiohttp
+
+import duniterpy.api.bma as bma
+from duniterpy.documents import BMAEndpoint, BlockUID, Transaction
+from duniterpy.documents.transaction import InputSource, OutputSource, Unlock, SIGParameter
+from duniterpy.grammars.output import Condition, SIG
+from duniterpy.key import SigningKey
+
+# CONFIG #######################################
+
+# You can either use a complete defined endpoint : [NAME_OF_THE_API] [DOMAIN] [IPv4] [IPv6] [PORT]
+# or the simple definition : [NAME_OF_THE_API] [DOMAIN] [PORT]
+# Here we use the BASIC_MERKLED_API
+BMA_ENDPOINT = "BASIC_MERKLED_API cgeek.fr 9330"
+
+# Credentials should be prompted or kept in a separate secure file
+# create a file with the salt on the first line and the password on the second line
+# the script will load them from the file
+FROM_CREDENTIALS_FILE = "/home/username/.credentials.txt"
+
+# Public key of the source account
+FROM_PUBKEY = "XXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXX"
+
+# Public key of the target account
+TO_PUBKEY = "YYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYY"
+
+
+################################################
+
+
+# Latest duniter-python-api is asynchronous and you have to create an aiohttp session to send request
+# ( http://pythonhosted.org/aiohttp )
+AIOHTTP_SESSION = aiohttp.ClientSession()
+
+# Version of the transaction document
+TRANSACTION_VERSION = 3
+
+async def get_current_block():
+ """
+ Get the current block data
+
+ :rtype: dict
+ """
+ # Here we request for the path blockchain/block/N
+ return await bma.blockchain.Current(BMAEndpoint.from_inline(BMA_ENDPOINT).conn_handler()) \
+ .get(AIOHTTP_SESSION)
+
+async def get_sources(pubkey):
+ """
+ Get the current block data
+
+ :param str pubkey: Public key of the sources account
+
+ :rtype: dict
+ """
+ # Here we request for the path blockchain/block/N
+ return await bma.tx.Sources(BMAEndpoint.from_inline(BMA_ENDPOINT).conn_handler(), pubkey) \
+ .get(AIOHTTP_SESSION)
+
+
+def get_transaction_document(current_block, source, from_pubkey, to_pubkey):
+ """
+ Return a Transaction document
+
+ :param dict current_block: Current block infos
+ :param dict source: Source to send
+ :param str from_pubkey: Public key of the issuer
+ :param str to_pubkey: Public key of the receiver
+
+ :return: Transaction
+ """
+ # list of inputs (sources)
+ inputs = [
+ InputSource(
+ amount=source['amount'],
+ base=source['base'],
+ source=source['type'],
+ origin_id=source['identifier'],
+ index=source['noffset']
+ )
+ ]
+
+ # list of issuers of the inputs
+ issuers = [
+ from_pubkey
+ ]
+
+ # list of unlocks of the inputs
+ unlocks = [
+ Unlock(
+ # inputs[index]
+ index=0,
+ # unlock inputs[index] if signatures[0] is from public key of issuers[0]
+ parameters=[SIGParameter(0)]
+ )
+ ]
+
+ # lists of outputs
+ outputs = [
+ OutputSource(
+ amount=source['amount'],
+ base=source['base'],
+ # only the receiver of the output can use it as input in another transaction
+ conditions=Condition.token(SIG.token(to_pubkey))
+ )
+ ]
+
+ transaction = Transaction(
+ version=TRANSACTION_VERSION,
+ currency=current_block['currency'],
+ blockstamp=BlockUID(current_block['number'], current_block['hash']),
+ locktime=0,
+ issuers=issuers,
+ inputs=inputs,
+ unlocks=unlocks,
+ outputs=outputs,
+ comment='',
+ signatures=None
+ )
+
+ return transaction
+
+async def main():
+ """
+ Main code
+ """
+ # capture current block to get version and currency and blockstamp
+ current_block = await get_current_block()
+
+ response = await get_sources(FROM_PUBKEY)
+
+ if len(response['sources']) == 0:
+ print("no sources found for account %s" % FROM_PUBKEY)
+ exit(1)
+
+ # get the first source
+ source = response['sources'][0]
+
+ # create the transaction document
+ transaction = get_transaction_document(current_block, source, FROM_PUBKEY, TO_PUBKEY)
+
+ # load credentials from a text file
+ salt, password = open(FROM_CREDENTIALS_FILE).readlines()
+
+ # cleanup newlines
+ salt, password = salt.strip(), password.strip()
+
+ # create keys from credentials
+ key = SigningKey(salt, password)
+
+ # sign document
+ transaction.sign([key])
+
+ # send the Transaction document to the node
+ data = {'transaction': transaction.signed_raw()}
+ response = await bma.tx.Process(BMAEndpoint.from_inline(BMA_ENDPOINT).conn_handler()) \
+ .post(AIOHTTP_SESSION, **data)
+
+ print(response)
+
+ response.close()
+
+with AIOHTTP_SESSION:
+
+ # Latest duniter-python-api is asynchronous and you have to use asyncio, an asyncio loop and a "as" on the data.
+ # ( https://docs.python.org/3/library/asyncio.html )
+ asyncio.get_event_loop().run_until_complete(main())
--
GitLab