BASE_URL = "https://industrial.api.ubidots.com"
REQUESTS_FUNCTIONS = {"get": requests.get, "post": requests.post}
Main function - runs every time the function is executed.
"args" is a dictionary containing both the URL params and the HTTP body (for POST requests).
token = args.get('token', None)
device = args.get('device', None)
if token is None or device is None:
print("[ERROR] Please send your Ubidots Token and device label to update in your args")
return {"status": "error"}
# Log the payload to the console, for debugging purposes. You may access the function's logs using
# the option in the above header.
print("[INFO] Payload to send: {}".format(args))
# Use the remaining parameters as payload
req = update_device(device, args, token)
# Prints the request result
print("[INFO] Request result:")
return {"status": "Ok", "result": req.json()}
def update_device(device, payload, token):
updates a variable with a single dot
url = "{}/api/v1.6/devices/{}".format(BASE_URL, device)
headers = {"X-Auth-Token": token, "Content-Type": "application/json"}
req = create_request(url, headers, payload, attempts=5, request_type="post")
def create_request(url, headers, data, attempts, request_type):
Function to create a request to the server
request_func = REQUESTS_FUNCTIONS.get(request_type)
kwargs = {"url": url, "headers": headers}
if request_type == "post":
req = request_func(**kwargs)
print("[INFO] Request result: {}".format(req.text))
status_code = req.status_code
while status_code >= 400 and attempts < 5:
req = request_func(**kwargs)
print("[INFO] Request result: {}".format(req.text))
status_code = req.status_code
print("[ERROR] There was an error with the request, details:")