The following is an example of accessing the Telldus PA-API using Python
This example requires the following 3rd party Python packages:
Flask
requests_oauthlib
You can install them using pip:
pip install -U flask requests_oauthlib
The example project
Running the project
You can run the project's main application using python from the root of the project folder:
python app.py
Project directory
You can organise your project directory as follows:
telldus_flask_app/ │ ├── app.py ├── config.py ├── requirements.txt └── templates/ ├── index.html ├── clients.html ├── devices.html └── authorization_failed.html
App & configurations
Configuration
We define some commonly used constants here.
config.py
# config.py import os # Obtain the public and private key from https://pa-api.telldus.com/keys/showToken CONSUMER_KEY = 'Public key' CONSUMER_SECRET = 'Private key' # Defines the API end points REQUEST_TOKEN_URL = 'https://pa-api.telldus.com/oauth/requestToken' AUTHORIZE_URL = 'https://pa-api.telldus.com/oauth/authorize' ACCESS_TOKEN_URL = 'https://pa-api.telldus.com/oauth/accessToken' API_BASE_URL = 'https://pa-api.telldus.com/json' CALLBACK_URL = 'http://127.0.0.1:5000/callback' # Update with your domain # Flask secret key for session management SECRET_KEY = os.urandom(24) # Add to config.py TURNON = 1 # Example value, replace with actual constant TURNOFF = 2 # Example value, replace with actual constant
The Flask application
The main application.
app.py
# app.py from flask import Flask, session, redirect, url_for, request, render_template, flash from requests_oauthlib import OAuth1Session import config import json app = Flask(__name__) app.config.from_object('config') # OAuth1Session setup def get_oauth_session(token=None, token_secret=None): return OAuth1Session( client_key=config.CONSUMER_KEY, client_secret=config.CONSUMER_SECRET, resource_owner_key=token, resource_owner_secret=token_secret, callback_uri=config.CALLBACK_URL ) def print_debug_message(message): print(f'[DEBUG] {message}') @app.route('/') def index(): access_token = session.get('access_token') access_token_secret = session.get('access_token_secret') print_debug_message("Accessing home page.") print_debug_message(f"Access Token: {access_token}") print_debug_message(f"Access Token Secret: {access_token_secret}") if not access_token: print_debug_message("No access token found. User not authenticated.") return render_template('index.html', authenticated=False) # Example API call: List devices oauth = get_oauth_session(access_token, access_token_secret) params = { 'supportedMethods': config.TURNOFF|config.TURNON, # Example parameter 'format': 'json' } print_debug_message(f"Making API call to {config.API_BASE_URL}/devices/list with params {params}") try: response = oauth.get(f"{config.API_BASE_URL}/devices/list", params=params) print_debug_message(f"API Response Status Code: {response.status_code}") if response.status_code != 200: flash('Failed to fetch devices.', 'danger') print_debug_message("Failed to fetch devices from API.") devices = [] else: devices = response.json().get('device', []) print_debug_message(f"Devices fetched: {json.dumps(devices, indent=2)}") except Exception as e: flash('An error occurred while fetching devices.', 'danger') print_debug_message(f"Exception during API call: {str(e)}") devices = [] return render_template('index.html', authenticated=True, devices=devices) @app.route('/login') def login(): print_debug_message("Initiating OAuth login process.") oauth = get_oauth_session() try: fetch_response = oauth.fetch_request_token(config.REQUEST_TOKEN_URL) print_debug_message(f"Request Token fetched: {fetch_response}") except ValueError as e: flash('Failed to obtain request token.', 'danger') print_debug_message(f"Failed to fetch request token: {str(e)}") return redirect(url_for('index')) session['resource_owner_key'] = fetch_response.get('oauth_token') session['resource_owner_secret'] = fetch_response.get('oauth_token_secret') print_debug_message(f"Stored request token in session: {session['resource_owner_key']}") authorization_url = oauth.authorization_url(config.AUTHORIZE_URL) print_debug_message(f"Redirecting to authorization URL: {authorization_url}") return redirect(authorization_url) @app.route('/callback') def callback(): print_debug_message("Handling OAuth callback.") resource_owner_key = session.get('resource_owner_key') resource_owner_secret = session.get('resource_owner_secret') print_debug_message(f"Resource Owner Key: {resource_owner_key}") print_debug_message(f"Resource Owner Secret: {resource_owner_secret}") if not resource_owner_key or not resource_owner_secret: flash('Missing request token.', 'danger') print_debug_message("Missing request token in session.") return redirect(url_for('index')) oauth = get_oauth_session(resource_owner_key, resource_owner_secret) oauth_response = oauth.parse_authorization_response(request.url) verifier = oauth_response.get('oauth_verifier') print_debug_message(f"OAuth Verifier obtained: {verifier}") if not verifier: flash('Authorization failed or was denied.', 'danger') print_debug_message("Authorization failed or was denied by the user.") return redirect(url_for('index')) try: oauth_tokens = oauth.fetch_access_token(config.ACCESS_TOKEN_URL, verifier=verifier) print_debug_message(f"Access Token fetched: {oauth_tokens}") except ValueError as e: flash('Failed to obtain access token.', 'danger') print_debug_message(f"Failed to fetch access token: {str(e)}") return redirect(url_for('index')) session['access_token'] = oauth_tokens.get('oauth_token') session['access_token_secret'] = oauth_tokens.get('oauth_token_secret') print_debug_message(f"Stored access token in session: {session['access_token']}") # Clear request tokens session.pop('resource_owner_key', None) session.pop('resource_owner_secret', None) print_debug_message("Cleared request tokens from session.") flash('You have successfully logged in.', 'success') print_debug_message("User successfully authenticated and logged in.") return redirect(url_for('index')) @app.route('/logout') def logout(): print_debug_message("Logging out user.") session.pop('access_token', None) session.pop('access_token_secret', None) flash('You have been logged out.', 'success') print_debug_message("Access tokens removed from session.") return redirect(url_for('index')) @app.route('/devices') def devices(): print_debug_message("Accessing devices route.") access_token = session.get('access_token') access_token_secret = session.get('access_token_secret') print_debug_message(f"Access Token: {access_token}") print_debug_message(f"Access Token Secret: {access_token_secret}") if not access_token: flash('You need to log in first.', 'warning') print_debug_message("User not authenticated. Redirecting to home.") return redirect(url_for('index')) oauth = get_oauth_session(access_token, access_token_secret) params = { 'supportedMethods': config.TURNON|config.TURNOFF, # Example parameter 'format': 'json' } print_debug_message(f"Making API call to {config.API_BASE_URL}/devices/list with params {params}") try: response = oauth.get(f"{config.API_BASE_URL}/devices/list", params=params) print_debug_message(f"API Response Status Code: {response.status_code}") if response.status_code != 200: flash('Failed to fetch devices.', 'danger') print_debug_message("Failed to fetch devices from API.") devices = [] else: devices = response.json().get('devices', []) print_debug_message(f"Devices fetched: {json.dumps(devices, indent=2)}") except Exception as e: flash('An error occurred while fetching devices.', 'danger') print_debug_message(f"Exception during API call: {str(e)}") devices = [] return render_template('devices.html', devices=devices) @app.route('/clients') def clients(): print_debug_message("Accessing clients route.") access_token = session.get('access_token') access_token_secret = session.get('access_token_secret') print_debug_message(f"Access Token: {access_token}") print_debug_message(f"Access Token Secret: {access_token_secret}") if not access_token: flash('You need to log in first.', 'warning') print_debug_message("User not authenticated. Redirecting to home.") return redirect(url_for('index')) oauth = get_oauth_session(access_token, access_token_secret) params = { 'format': 'json' } print_debug_message(f"Making API call to {config.API_BASE_URL}/json/clients/list with params {params}") try: response = oauth.get(f"{config.API_BASE_URL}/clients/list", params=params) print_debug_message(f"API Response Status Code: {response.status_code}") if response.status_code != 200: flash('Failed to fetch clients.', 'danger') print_debug_message("Failed to fetch clients from API.") clients = [] else: clients = response.json().get('client', []) print_debug_message(f"Clients fetched: {json.dumps(clients, indent=2)}") except Exception as e: flash('An error occurred while fetching clients.', 'danger') print_debug_message(f"Exception during API call: {str(e)}") clients = [] return render_template('clients.html', clients=clients) if __name__ == '__main__': app.secret_key = config.SECRET_KEY app.run(debug=True)
The html templates
The home page
templates/index.html
<!DOCTYPE html> <html lang="en"> <head> <meta charset="UTF-8"> <title>Telldus Flask App</title> <link rel="stylesheet" href="https://stackpath.bootstrapcdn.com/bootstrap/4.5.0/css/bootstrap.min.css"> </head> <body> <div class="container mt-5"> {% with messages = get_flashed_messages(with_categories=true) %} {% if messages %} {% for category, message in messages %} <div class="alert alert-{{ category }} alert-dismissible fade show" role="alert"> {{ message }} <button type="button" class="close" data-dismiss="alert" aria-label="Close"> <span aria-hidden="true">×</span> </button> </div> {% endfor %} {% endif %} {% endwith %} <h1 class="mb-4">Telldus Live! Integration</h1> {% if not authenticated %} <p>You are not connected to Telldus Live!.</p> <a href="{{ url_for('login') }}" class="btn btn-primary">Connect with Telldus</a> {% else %} <p>You are connected to Telldus Live!</p> <a href="{{ url_for('logout') }}" class="btn btn-danger">Logout</a> <hr> <h2>Devices</h2> {% if devices %} <ul class="list-group"> {% for device in devices %} <li class="list-group-item"> <strong>{{ device.name }}</strong> (ID: {{ device.id }}) <!-- Add more device details as needed --> </li> {% endfor %} </ul> {% else %} <p>No devices found.</p> {% endif %} <hr> <h2>Clients</h2> <a href="{{ url_for('clients') }}" class="btn btn-info">View Clients</a> {% endif %} </div> <script src="https://code.jquery.com/jquery-3.5.1.slim.min.js"></script> <script src="https://cdn.jsdelivr.net/npm/bootstrap@4.5.0/dist/js/bootstrap.bundle.min.js"></script> </body> </html>
The devices page
templates/devices.html
<!DOCTYPE html> <html lang="en"> <head> <meta charset="UTF-8"> <title>Your Devices</title> <link rel="stylesheet" href="https://stackpath.bootstrapcdn.com/bootstrap/4.5.0/css/bootstrap.min.css"> </head> <body> <div class="container mt-5"> {% with messages = get_flashed_messages(with_categories=true) %} {% if messages %} {% for category, message in messages %} <div class="alert alert-{{ category }} alert-dismissible fade show" role="alert"> {{ message }} <button type="button" class="close" data-dismiss="alert" aria-label="Close"> <span aria-hidden="true">×</span> </button> </div> {% endfor %} {% endif %} {% endwith %} <h1>Your Devices</h1> <a href="{{ url_for('index') }}" class="btn btn-secondary mb-3">Back to Home</a> {% if devices %} <ul class="list-group"> {% for device in devices %} <li class="list-group-item"> <strong>{{ device.name }}</strong> (ID: {{ device.id }}) <!-- Add more device details as needed --> </li> {% endfor %} </ul> {% else %} <p>No devices found.</p> {% endif %} </div> <script src="https://code.jquery.com/jquery-3.5.1.slim.min.js"></script> <script src="https://cdn.jsdelivr.net/npm/bootstrap@4.5.0/dist/js/bootstrap.bundle.min.js"></script> </body> </html>
The clients page
templates/clients.html
<!DOCTYPE html> <html lang="en"> <head> <meta charset="UTF-8"> <title>Your Clients</title> <link rel="stylesheet" href="https://stackpath.bootstrapcdn.com/bootstrap/4.5.0/css/bootstrap.min.css"> </head> <body> <div class="container mt-5"> {% with messages = get_flashed_messages(with_categories=true) %} {% if messages %} {% for category, message in messages %} <div class="alert alert-{{ category }} alert-dismissible fade show" role="alert"> {{ message }} <button type="button" class="close" data-dismiss="alert" aria-label="Close"> <span aria-hidden="true">×</span> </button> </div> {% endfor %} {% endif %} {% endwith %} <h1>Your Clients</h1> <a href="{{ url_for('index') }}" class="btn btn-secondary mb-3">Back to Home</a> {% if clients %} <ul class="list-group"> {% for client in clients %} <li class="list-group-item"> <strong>{{ client.name }}</strong> (ID: {{ client.id }}) <!-- Add more client details as needed --> </li> {% endfor %} </ul> {% else %} <p>No clients found.</p> {% endif %} </div> <script src="https://code.jquery.com/jquery-3.5.1.slim.min.js"></script> <script src="https://cdn.jsdelivr.net/npm/bootstrap@4.5.0/dist/js/bootstrap.bundle.min.js"></script> </body> </html>
The authorisation failed page
templates/authorization_failed.html
<!-- templates/authorisation_failed.html --> <!DOCTYPE html> <html lang="en"> <head> <meta charset="UTF-8"> <title>Authorisation Failed</title> </head> <body> <h1>Authorization Failed!</h1> <p><a href="{{ url_for('index') }}">Go back</a></p> </body> </html>