Liquid Securities API Tutorial Code

import json
import random
import requests
import string
import time


# Beta/test API:
#API_URL = 'https://securities-beta.blockstream.com/api'
# Live API:
API_URL = 'https://securities.blockstream.com/api'


def get_auth_token_header(username, password):
    # api/user/obtain_token
    url = f'{API_URL}/user/obtain_token'
    headers = {'content-type': 'application/json'}
    payload = {'username': username, 'password': password}
    response = requests.post(url, data=json.dumps(payload), headers=headers)
    assert response.status_code == 200
    json_data = json.loads(response.text)
    token = json_data['token']
    return {'content-type': 'application/json', 'Authorization': f'token {token}'}

def rand_string():
    # Random string used to prevent repeat runs causing key violations
    # This is just for example purposes and not to be used in live
    return ''.join(random.choice(string.ascii_uppercase) for _ in range(6))


def main():
    parser = argparse.ArgumentParser(description='Example use of the Liquid Securities API.')
    parser.add_argument('-u', '--username', help='Liquid Securities API username', required=True)
    parser.add_argument('-p', '--password', help='Liquid Securities API password', required=True)
    args = parser.parse_args()

    username = args.username
    password = args.password

    headers = get_auth_token_header(username, password)

    # Investors - validate GAID
    # api​/gaids​/{gaid}​/validate
    # This can be useful if you want to validate investor GAIDs before trying to create investors.
    # NOTE: investors/add also performs this validation against the GAID.
    gaid = 'INVALIDZZ6SqZZvEpAeVz9QmfHqhh'
    url = f'{API_URL}/gaids/{gaid}/validate'
    response = requests.get(url, headers=headers)
    assert response.status_code == 200
    gaid_validation = json.loads(response.text)
    gaid_is_valid = gaid_validation['is_valid']
    print(f'\nValidating GAID \'{gaid}\'')
    print(f'GAID is Valid: {gaid_is_valid}')

    # Investors add
    # api/investors/add
    # This api also accepts a list of investors.
    # If a list is provided and one or more of the investors fails
    # validation, none of the investors will be created and a list of
    # errors will be returned.
    # You must add a valid GAID before running the example add below.
    url = f'{API_URL}/investors/add'
    investor_name = f'Investor {rand_string()}'
    payload = {'is_company': False,
        'name': investor_name,
        'GAID': ''
    }
    response = requests.post(url, data=json.dumps(payload), headers=headers)
    assert response.status_code == 201
    investor = json.loads(response.text)
    investor_id = int(investor['id'])
    print('\nNew investor (individual):')
    print('--------------------------')
    print(json.dumps(investor, indent=4))

    # Investors details
    # api/investors/{investor_id}
    url = f'{API_URL}/investors/{investor_id}'
    response = requests.get(url, headers=headers)
    assert response.status_code == 200
    investor_details = json.loads(response.text)
    print('\nInvestor details:')
    print('-----------------')
    print(json.dumps(investor_details, indent=4))

    # Investors update name
    # api/investors/{investor_id}/edit
    url = f'{API_URL}/investors/{investor_id}/edit'
    payload = {'name': f'{investor_name} updated'}
    response = requests.put(url, data=json.dumps(payload), headers=headers)
    assert response.status_code == 200
    investor_details = json.loads(response.text)
    print('\nInvestor details:')
    print('-----------------')
    print(json.dumps(investor_details, indent=4))

    # Investors list
    # api/investors
    url = f'{API_URL}/investors'
    response = requests.get(url, headers=headers)
    assert response.status_code == 200
    investors = json.loads(response.text)
    print('\nInvestors list:')
    print('---------------')
    for investor in investors:
        print(json.dumps(investor, indent=4))

    # Category add
    # api/categories/add
    url = f'{API_URL}/categories/add'
    category_name = f'Category {rand_string()}'
    payload = {'name': category_name,
        'description':'Category for example'}
    response = requests.post(url, data=json.dumps(payload), headers=headers)
    assert response.status_code == 201
    category = json.loads(response.text)
    category_id = int(category['id'])
    print('\nNew category:')
    print('-------------')
    print(json.dumps(category, indent=4))

    # Category details
    # api/categories/{category_id}
    url = f'{API_URL}/categories/{category_id}'
    response = requests.get(url, headers=headers)
    assert response.status_code == 200
    category_details = json.loads(response.text)
    print('\nCategory details:')
    print('-----------------')
    print(json.dumps(category_details, indent=4))

    # Category update
    # api/categories/{category_id}/edit
    url = f'{API_URL}/categories/{category_id}/edit'
    payload = {'name': f'{category_name} updated', 'description':'Category example updated'}
    response = requests.put(url=url, data=json.dumps(payload), headers=headers)
    assert response.status_code == 200
    category_details = json.loads(response.text)
    print('\nCategory details:')
    print('-----------------')
    print(json.dumps(category_details, indent=4))

    # Category list
    # api/categories
    url = f'{API_URL}/categories'
    response = requests.get(url, headers=headers)
    assert response.status_code == 200
    categories = json.loads(response.text)
    print('\nCategories list:')
    print('----------------')
    for category in categories:
        print(json.dumps(category, indent=4))

    # Investor category - add investor to category
    # api/categories/{categoryId}/investors/add
    url = f'{API_URL}/categories/{category_id}/investors/add'
    payload = {[investor_id,]}
    response = requests.post(url, data=json.dumps(payload), headers=headers)
    assert response.status_code == 200

    # Investor details - check category was added
    # api/investors/{investor_id}
    url = f'{API_URL}/investors/{investor_id}'
    response = requests.get(url, headers=headers)
    assert response.status_code == 200
    investor_details = json.loads(response.text)
    print('\nInvestor details:')
    print('-----------------')
    print(json.dumps(investor_details, indent=4))

    # You can use the following to remove investors from a category:
    """"
    # Investor category - remove investor from category
    # api/categories/{category_id}/investors/delete
    url = f'{API_URL}/categories/{category_id}/investors/delete'
    payload = {[investor_id,]}
    response = requests.delete(url, data=json.dumps(payload), headers=headers)
    print(response.text)
    assert response.status_code == 200
    """

    # Issue an asset
    # api/assets/issue
    # If the asset contains registration data (ticker, domain, pubkey) later updates will not be possible
    # If is_reissuable is true then reissuance_amount and reissuance_address must be provided,
    #     and reissuance_address must be different from destination_address. Name, ticker,
    # domain and pubkey are committed to the issuance transaction, and cannot be changed later.
    #
    # You will need to amend the following values from the example values given below:
    # name: the name of the Asset as it will appear in Liquid Securities.
    #     Length must be 5 to 255 ascii characters.
    # amount: the amount of the asset to issue.
    #     integer, minimum: 1, maximum: 2100000000000000
    # is_confidential: if true, the issuance amount will not be readable on the Liquid blockchain.
    #     for most issuances it is expected that this will be False.
    # destination_address: an address generated by your Liquid node that will receive the issued asset.
    # domain: the domain that will be used to verify the asset. Must be a valid domain name format,
    #     for example: example.com or sub.example.com. Do not include the http/s or www prefixes.
    # ticker: the ticker you would like to assign to the asset.
    #     length must be 3 to 5 characters (valid characters are a to z, A to Z, '.' and '-').
    # pubkey: pubkey for asset registry, must be a compressed pubkey in hex.
    #     You can obtian the pubkey from an elements address using the elements getaddressinfo rpc command
    # is_reissuable: if true, the asset will be created as reissuable.
    # reissuance_address: the address that will receive the reissuance token if is_reissuable = True
    # reissuance_amount: the amount of reissuance tokens to create if is_reissuable = True

    # An address generated by your Liquid node that will receive the issued asset for the example
    asset_destination_address = ''

    # An address generated by your Liquid node that will receive the reissuance token for the example
    reissuance_token_destination_address = ''

    url = f'{API_URL}/assets/issue'
    payload = {'name': 'An example asset with registration data',
                'amount': 17000000,
                'is_confidential': True,
                'destination_address': asset_destination_address,
                'domain': 'yourdomainforregistrationproof.com',
                'ticker': 'LSEXA',
                'pubkey': '02' * 33,
                'is_reissuable': True,
                'reissuance_address': reissuance_token_destination_address,
                'reissuance_amount': 1,}
    response = requests.post(url=url, data=json.dumps(payload), headers=headers)
    assert response.status_code == 201
    asset = json.loads(response.text)
    asset_uuid = asset['asset_uuid']
    print('\nNewly issued (registration enabled) asset:')
    print('------------------------------------------')
    print(json.dumps(asset, indent=4))

    # Actions such as Reissuance, Authorized asset registration, Assignment need > 1 confirmation of Issuance transaction
    print('\nIssuance transaction needs > 1 confirmation before some other actions can be performed.\nSleeping for 180 seconds.')
    time.sleep(3*60)

    # Register the asset as an Authorized Asset
    # api/assets/{asset_uuid}/register-authorized
    url = f'{API_URL}/assets/{asset_uuid}/register-authorized'
    response = requests.get(url, headers=headers)
    assert response.status_code == 200
    register_authorized = json.loads(response.text)
    print('\nRegister authorized summary:')
    print('----------------------------')
    print(json.dumps(register_authorized, indent=4))

    # List issued assets
    # api/assets
    url = f'{API_URL}/assets'
    response = requests.get(url, headers=headers)
    assert response.status_code == 200
    assets = json.loads(response.text)
    print('\nIssued assets:')
    print('--------------')
    for asset in assets:
        print(json.dumps(asset, indent=4))

    # Asset details
    # api/assets/{asset_uuid}
    url = f'{API_URL}/assets/{asset_uuid}'
    response = requests.get(url, headers=headers)
    assert response.status_code == 200
    asset = json.loads(response.text)
    print(f'\nAsset details:')
    print('--------------')
    print(json.dumps(asset, indent=4))

    # Reissue some of the asset.
    # Request the data needed to carry out a reissuance.
    # api/assets/{assetUuid}/reissue-request
    # Returns json data that should be saved to file. The path to the file should be passed as an argument
    # to a Python client that carries out the reissuance itself. The client will be supplied to users of
    # the api and will handle the reissuance transaction and post back the resulting transaction data
    # to the /assets/{assetUuid}/reissue-confirm endpoint after it has sufficient confirmations.
    # NOTE: Before the reissuance can be made, the transaction in which the asset was issued
    # needs to have had at least 2 confirmations on the Liquid blockchain or the reissuance will fail.
    url = f'{API_URL}/assets/{asset_uuid}/reissue-request'
    payload = {'amount_to_reissue':1}
    response = requests.post(url=url, data=json.dumps(payload), headers=headers)
    assert response.status_code == 200
    reissuance_data = json.loads(response.text)
    # The json data in reissuance_data would then be saved to a file for use in the Python client
    # that carries out the reissuance itself.
    print(json.dumps(reissuance_data, indent=4))

    # Add the Category to the Asset as a requirement
    # api/assets/{assetUuid}/categories/add
    url = f'{API_URL}/assets/{asset_uuid}/categories/add'
    payload = {[category_id,]}
    response = requests.post(url=url, data=json.dumps(payload), headers=headers)
    assert response.status_code == 200
    asset = json.loads(response.text)
    requirements = asset['requirements']
    print('\nAsset Requirements (Investor Categories ids):')
    print('---------------------------------------------')
    print(json.dumps(requirements, indent=4))

    """
    # Remove the Investor Category from the Asset as a requirement
    # api/assets/{asset_uuid}/categories/delete
    url = f'{API_URL}/assets/{asset_uuid}/categories-delete'
    payload = {[category_id,]}
    response = requests.delete(url=url, data=json.dumps(payload), headers=headers)
    assert response.status_code == 200
    asset = json.loads(response.text)
    requirements = asset['requirements']
    print('\nAsset Requirements (Investor Categories ids):')
    print('---------------------------------------------')
    print(json.dumps(requirements, indent=4))
    """

    # Assign some of the asset to the Investor
    # api/assets/{asset_uuid}/assignments/create
    # The assignment will be created as ready for distribution
    # NOTE: Before the assignments can be made, the transaction in which the asset was issued
    # needs to have had at least 2 confirmations on the Liquid blockchain or the assignment
    # will fail with a warning about insufficient confirmed funds.
    url = f'{API_URL}/assets/{asset_uuid}/assignments/create'
    payload = {'assignments': [{
            'investor': investor_id,
            'amount': 1,
            'ready_for_distribution': True}]}
    response = requests.post(url=url, data=json.dumps(payload), headers=headers)
    assert response.status_code == 200
    assignment = json.loads(response.text)
    assignment_id = assignment[0]['id']
    print('\nAsset Assignment:')
    print('-----------------')
    print(json.dumps(assignment, indent=4))

    # Assignment details
    # api/assets/{asset_uuid}/assignments/{assignment_id}
    url = f'{API_URL}/assets/{asset_uuid}/assignments/{assignment_id}'
    response = requests.get(url, headers=headers)
    assert response.status_code == 200
    assignment = json.loads(response.text)
    print('\nAssignment details:')
    print('-------------------')
    print(json.dumps(assignment, indent=4))

    # Investor summary
    # api/investors/{investor_id}
    # Balances and details of assignments, distributions for each asset held by the investor
    url = f'{API_URL}/investors/{investor_id}/summary'
    response = requests.get(url, headers=headers)
    assert response.status_code == 200
    investor_summary = json.loads(response.text)
    print('\nInvestor summary:')
    print('-----------------')
    print(json.dumps(investor_summary, indent=4))

    # Asset assignments list
    # api/assets/{asset_uuid}/assignments
    # The assignments for the asset (expect this to be empty unless you have already created assignments)
    url = f'{API_URL}/assets/{asset_uuid}/assignments'
    response = requests.get(url, headers=headers)
    assert response.status_code == 200
    assignments = json.loads(response.text)
    print('\nAsset assignments:')
    print('------------------')
    for assignment in assignments:
        print(json.dumps(assignment, indent=4))

    # Request the data needed to carry out a distribution.
    # api/assets/{asset_uuid}/distributions/create/
    # Returns json data that should be saved to file. The path to the file should be passed as an argument
    # to a Python client that carries out the distribution itself. The client will be supplied to users of
    # the api and will handle the distribution transactions and post back the resulting transaction data
    # to the distributions/confirm endpoint after it has sufficient confirmations.
    # NOTE: The url intentionally ends with a '/'
    url = f'{API_URL}/assets/{asset_uuid}/distributions/create/'
    response = requests.get(url, headers=headers)
    assert response.status_code == 200
    distribution_data = json.loads(response.text)
    # The json data in distribution_data would then be saved to a file for use in the Python client
    # that carries out the distribution itself.
    print(json.dumps(distribution_data, indent=4))

    # The Assignment will now have a distribution_uuid:
    # Assignment details
    # api/assets/{asset_uuid}/assignments/{assignment_id}
    url = f'{API_URL}/assets/{asset_uuid}/assignments/{assignment_id}'
    response = requests.get(url, headers=headers)
    assert response.status_code == 200
    assignment = json.loads(response.text)
    print('\nAssignment details:')
    print('-------------------')
    print(json.dumps(assignment, indent=4))

    # Distribution details
    # api/assets/{asset_uuid}/distributions/{distribution_uuid}
    # Displays details for a distribution that has been distributed with a valid transaction using the
    # distribution Python client. Until the distribution transaction has confirmed the details api
    # will return a not found error.
    # NOTE: For a list of pending distributions use the assignments list api
    """
    url = f'{API_URL}/assets/{asset_uuid}/distributions/{distribution_uuid}'
    response = requests.get(url, headers=headers)
    assert response.status_code == 200
    distribution = json.loads(response.text)
    print('\nDistribution details:')
    print('---------------------')
    print(json.dumps(distribution, indent=4))
    """

    # Distributions list
    # api/assets/{asset_uuid}/distributions
    # Displays distributions that have been distributed with valid transactions
    # NOTE: For a list of pending distributions use the assignments list api
    url = f'{API_URL}/assets/{asset_uuid}/distributions'
    response = requests.get(url, headers=headers)
    assert response.status_code == 200
    distributions = json.loads(response.text)
    print('\nDistribution:')
    print('-------------')
    for distribution in distributions:
        print(json.dumps(distribution, indent=4))

    # Asset summary
    # api/assets/{assetUuid}/summary
    # Balances and stats of an Asset
    url = f'{API_URL}/assets/{asset_uuid}/summary'
    response = requests.get(url, headers=headers)
    assert response.status_code == 200
    asset_summary = json.loads(response.text)
    print('\nAsset summary:')
    print('--------------')
    print(json.dumps(asset_summary, indent=4))

    # Asset activities
    # api/assets/{assetUuid}/activities
    url = f'{API_URL}/assets/{asset_uuid}/activities'
    response = requests.get(url, headers=headers)
    assert response.status_code == 200
    asset_activities = json.loads(response.text)
    print('\nAsset activities:')
    print('-----------------')
    print(json.dumps(asset_activities, indent=4))

    # Asset ownerships
    # api/assets/{assetUuid}/ownerships
    url = f'{API_URL}/assets/{asset_uuid}/ownerships'
    response = requests.get(url, headers=headers)
    assert response.status_code == 200
    asset_ownerships = json.loads(response.text)
    print('\nAsset ownerships:')
    print('-----------------')
    print(json.dumps(asset_ownerships, indent=4))

    # Asset balance
    # api/assets/{assetUuid}/balance
    url = f'{API_URL}/assets/{asset_uuid}/balance'
    response = requests.get(url, headers=headers)
    assert response.status_code == 200
    asset_balance = json.loads(response.text)
    print('\nAsset balance:')
    print('--------------')
    print(json.dumps(asset_balance, indent=4))

if __name__ == '__main__':
    main()