I am trying to query Luno API 's API using the official luno python wrapper. GitHub - luno/luno-python: Python SDK for the Luno API
I installed it using pip3 install luno-pytho on my Linux machine hosted on AWS ec2 server, my python script is as follows:
import json
from luno_python.client import Client
c = Client(api_key_id='xxxx', api_key_secret='xxxxxx')
try:
result = c.get_ticker(pair='ETHZAR')
except Exception as e:
print (e)
print(json.dumps(result))
I originally had this code inside a bigger python script(can provide code if needed) which also queried another crypto site’s API which I got to work(manually wrote out the requests), but for luno I found a nice python wrapper and thought I’d try and use it to save some time, but I cannot get it to work.
When running the script(test.py) on my windows machine and my virtually hosted linux instance I get the following result, the first line is the correct output, however, when running it through telegraf it does not give an error, but it simply just does not print out the correct output?
Oh, I wasn’t aware it was returning a string, because for the other site I was querying, I used the exact same method and it just worked…I get a python dict then convert it to json with json.dumps()
Either the API or the Python library are implemented in a somewhat unfortunate way, as they deliver strings instead of floats. I would fix this in your Python script or directly in the Python library itself and make it floats.
The earlier in the chain, the better.
import json
def convert_floats_hook(dictionary):
for key, value in dictionary.items():
if type(value) is str:
try:
float_value = float(value)
except:
pass
else:
dictionary[key] = float_value
return dictionary
result = json.loads(result, object_hook=convert_floats_hook)
print(result)
EDIT:
I don’t have the full code, but this is how it should work. Just keep trying yourself, I gave you the idea. For example, just take the inner part of the function.
import json
from luno_python.client import Client
c = Client(api_key_id='xxxx', api_key_secret='xxxxx')
try:
result = c.get_ticker(pair='ETHZAR')
except Exception as e:
print (e)
def convert_floats_hook(dictionary):
for key, value in dictionary.items():
if type(value) is str:
try:
float_value = float(value)
except:
pass
else:
dictionary[key] = float_value
return dictionary
result = json.loads(json.dumps(result), object_hook=convert_floats_hook)
print(result)
Complete code I have in my test.py, it doesn’t seem to do anything?
Edited my suggestion above.
I don’t have the full code, but this is how it should work. Just keep trying yourself, I gave you the idea. For example, just take the inner part of the function.
for key, value in result.items():
if type(value) is str:
try:
float_value = float(value)
except:
pass
else:
result[key] = float_value
print(json.dumps(result))
This seems to work, it returns a valid json and works with telegraf…now to see if I can merge all 3 python dicts in my larger script and then convert to json, because unless I am mistaken telegraf seems to only be able to take 1 long json input and not multiple streams? I have multiple API calls and each outputs their own json, but telegraf gave me an error so I merged the python dicts together then converted it to json
Yes, but that’s a Json standard , not a limitation of telegraf.
Agreed that long term it’s better to encapsulate all your results into a single Json output from a single script, but in the interim - if you happen to have your different parts of your code in separate scripts while developing , you should be able to run them all on each interval. In the telegraf exec config , see the square brackets on the command section in telegraf config? That usually implies an array, and the telegraf docs do provide an example with multiple commands to execute too, which confirms it.
will have a look at it if I need to run multiple scripts at once…but at the moment my data is not reaching influxdb and I don’t know why it stopped…was working yesterday, the logs don’t help either