Skip to content
Important! Yanzi IoT product line will be sunsetted end of 2025. Click here to learn more.

Python example applications

These examples require Python 3.7 and the httpx library which can be installed using pip.

Getting a sessionId

Copy this code and put it in a file called login.py. Run it with python login.py <API-HOST> you@example.com, and it will ask for your password. If everything goes well it should then print out a sessionId that can be used to talk to the api.

Python
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
import httpx, getpass, sys

def login(username, host):
    r = httpx.post(
        f'https://{host}/login',
        json={'username': username, 'password': getpass.getpass()}
    )
    return r.json()['sessionId']

if __name__ == '__main__':
    try:
        _, host, username = sys.argv
    except:
        print('Usage: python login.py cirrushost mattias@yanzi.se')

    print(login(username, host))

Getting a list of groups

Copy the code below into a file called groups.py and run it by passing a cirrus host and sessionId. It should then make a request and print out all the groups which you are a member of. For legacy reasons groups are still called accounts in the API.

Python
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
import httpx, sys

accounts_query = """
query {
  accounts {
    accountName
    accountNumber
  }
}
"""

def graphql(query, variables={}):
    r = httpx.post(
        f'https://{host}/graphql',
        headers={'Authorization': f'bearer {sessionId}'},
        json={'query': query, 'variables': variables}
    )
    return r.json()

if __name__ == '__main__':
    try:
        _, host, sessionId = sys.argv
    except:
        print("usage: python groups.py devshrek.yanzi.nu YW5kcmVhcy5tYXJ0ZW5...")
        sys.exit(1)

    print(graphql(accounts_query))

Making a simple CLI

Copy the below code into another file and run it. As you might notice, this program handles input of host, username, and password in three different ways. As the first three arguments on the command line, as environment variables, and by asking after launch. This is still just a very quick example, and as some comments say, it is probably a good idea look at some more libraries to help.

Python
  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
import httpx, getpass, pprint, sys, os, urllib.parse

accounts_query = """
query {
  accounts {
    accountName
    accountNumber
  }
}
"""

locations_query = """
query($after: String, $search: String) {
    locations(first: 10, after: $after, search: $search) {
        cursor
        list {
            name
            locationId
        }
    }
}
"""

devices_query = """
query($locationId: String!, $after: String) {
    location(locationId: $locationId) {
        units(first: 10, after: $after) {
            cursor
            list {
                name
                unitAddress {
                    did
                }
            }
        }
    }
}
"""

datasources_query = """
query($locationId: String!, $did: String!) {
    location(locationId: $locationId) {
        unit(did: $did) {
            dataSources {
                variableName
                siUnit
            }
        }
    }
}
"""

def login(host, username, password):
    r = httpx.post(
        f'https://{host}/login',
        json={'username': username, 'password': password}
    )
    return r.json()['sessionId']


def graphql(host, sessionId, query, variables={}):
    r = httpx.post(
        f'https://{host}/graphql',
        headers={'Authorization': f'bearer {sessionId}'},
        json={'query': query, 'variables': variables}
    )
    return r.json()

# This could be replaced with the argparse lib https://docs.python.org/3/library/argparse.html
def argv(index): return sys.argv[index] if len(sys.argv) > index else None

def get_host():
    return argv(1) or os.getenv('CIRRUS_HOST') or input('Host: ')

def get_session(host):
    username = argv(2) or os.getenv('CIRRUS_USER') or input('Username: ')
    password = argv(3) or os.getenv('CIRRUS_PASS') or getpass.getpass()

    return login(host, username, password)

def print_help():
    print("""
        exit
        help
        accounts
        locations [after=<cursor>&search=<name>]
        devices <locationId> [cursor]
        datasources <locationId> <did>

        """)

def main():
    host = os.getenv('CIRRUS_HOST') or get_host()
    sessionId = os.getenv('CIRRUS_SESSIONID') or get_session(host)

    # this could be replaced with the cmd lib https://docs.python.org/3/library/cmd.html
    while True:
        try:
            command, *args = input('>>> ').split(' ')
        except KeyboardInterrupt:
            break

        if command == 'exit':
            break
        elif command == 'help':
            print_help()
        elif command == 'accounts':
            pprint.pprint(graphql(host, sessionId, accounts_query))
        elif command == 'locations':
            variables = dict(urllib.parse.parse_qsl(args[0])) if len(args) else {}
            pprint.pprint(graphql(host, sessionId, locations_query, variables))
        elif command == 'devices':
            pprint.pprint(graphql(host, sessionId, devices_query, {'locationId': args[0], 'after': args[1] if len(args) > 1 else None}))
        elif command == 'datasources':
            pprint.pprint(graphql(host, sessionId, datasources_query, {'locationId': args[0], 'did': args[1]}))
        else:
            print('Unknown command', command)
            print_help()


if __name__ == '__main__':
    main()