Changes between Initial Version and Version 1 of Examples/PA-API Python Example

Timestamp:

Oct 4, 2024, 3:35:10 PM (13 months ago)

Author:

edwin

Comment:

--

Examples/PA-API Python Example

@@ +1 @@
+**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>
+
+}}}
+