Examples/PA-API Python Example – Telldus Technologies AB

Context Navigation

← Previous Version
View Latest Version
Next Version →

Version 1 (modified by edwin, 2 months ago) ( diff )
--

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">&times;</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">&times;</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">&times;</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>

Note: See TracWiki for help on using the wiki.

Download in other formats:

Plain Text