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>
