| | 1 | == PA-API OAuth2 Python Example |
| | 2 | |
| | 3 | The following is an example of accessing the Telldus PA-API using Python and OAuth 2.0. |
| | 4 | |
| | 5 | This example requires the following 3rd party Python packages: **Flask**, **requests** |
| | 6 | |
| | 7 | You can install them using pip: |
| | 8 | |
| | 9 | ```bash |
| | 10 | pip install -U flask requests |
| | 11 | ``` |
| | 12 | |
| | 13 | **The Example Project** |
| | 14 | |
| | 15 | ### Running the Project |
| | 16 | |
| | 17 | You can run the project's main application using Python from the root of the project folder: |
| | 18 | |
| | 19 | {{{ |
| | 20 | python app.py |
| | 21 | }}} |
| | 22 | |
| | 23 | Then open your browser at `http://127.0.0.1:5000`. |
| | 24 | |
| | 25 | ### Project Directory |
| | 26 | |
| | 27 | You can organise your project directory as follows: |
| | 28 | |
| | 29 | {{{ |
| | 30 | telldus_oauth2_flask_app/ |
| | 31 | │ |
| | 32 | ├── app.py |
| | 33 | ├── config.py |
| | 34 | ├── requirements.txt |
| | 35 | └── templates/ |
| | 36 | └── index.html |
| | 37 | }}} |
| | 38 | |
| | 39 | == App & Configurations |
| | 40 | |
| | 41 | **Configuration** |
| | 42 | |
| | 43 | We define some commonly used constants here. |
| | 44 | |
| | 45 | `config.py` |
| | 46 | |
| | 47 | {{{ |
| | 48 | # config.py |
| | 49 | import os |
| | 50 | |
| | 51 | # Obtain the client ID and client secret from https://pa-api.telldus.com/keys/showToken |
| | 52 | CLIENT_ID = 'Your client ID (Public key)' |
| | 53 | CLIENT_SECRET = 'Your client secret (Private key)' |
| | 54 | |
| | 55 | # Default Telldus Live! account credentials |
| | 56 | USERNAME = 'your@email.com' |
| | 57 | PASSWORD = 'your_password' |
| | 58 | |
| | 59 | # OAuth2 and API endpoints |
| | 60 | TOKEN_URL = 'https://pa-api.telldus.com/oauth2/accessToken' |
| | 61 | API_BASE_URL = 'https://pa-api.telldus.com/oauth2' |
| | 62 | |
| | 63 | # Scope requested during token exchange |
| | 64 | SCOPE = 'live-app' |
| | 65 | |
| | 66 | # Flask secret key for session management |
| | 67 | SECRET_KEY = os.urandom(24) |
| | 68 | |
| | 69 | # User-Agent sent with all API requests |
| | 70 | USER_AGENT = 'MyTelldusApp/1.0' |
| | 71 | }}} |
| | 72 | |
| | 73 | `requirements.txt` |
| | 74 | |
| | 75 | {{{ |
| | 76 | Flask |
| | 77 | requests |
| | 78 | }}} |
| | 79 | |
| | 80 | **The Flask Application** |
| | 81 | |
| | 82 | The main application. |
| | 83 | |
| | 84 | `app.py` |
| | 85 | |
| | 86 | {{{ |
| | 87 | # app.py |
| | 88 | |
| | 89 | from flask import Flask, session, redirect, url_for, request, render_template, flash |
| | 90 | import requests |
| | 91 | import config |
| | 92 | import json |
| | 93 | |
| | 94 | app = Flask(__name__) |
| | 95 | app.config.from_object('config') |
| | 96 | |
| | 97 | # --------------------------------------------------------------------------- |
| | 98 | # OAuth2 helpers |
| | 99 | # --------------------------------------------------------------------------- |
| | 100 | |
| | 101 | def obtain_tokens(username, password): |
| | 102 | """Obtain OAuth2 access and refresh tokens using the password grant.""" |
| | 103 | payload = { |
| | 104 | 'grant_type': 'password', |
| | 105 | 'username': username, |
| | 106 | 'password': password, |
| | 107 | 'scope': config.SCOPE, |
| | 108 | 'client_id': config.CLIENT_ID, |
| | 109 | 'client_secret': config.CLIENT_SECRET |
| | 110 | } |
| | 111 | headers = { |
| | 112 | 'Content-Type': 'application/x-www-form-urlencoded', |
| | 113 | 'User-Agent': config.USER_AGENT |
| | 114 | } |
| | 115 | response = requests.post(config.TOKEN_URL, data=payload, headers=headers) |
| | 116 | response.raise_for_status() |
| | 117 | data = response.json() |
| | 118 | access_token = data.get('access_token') |
| | 119 | refresh_token = data.get('refresh_token') |
| | 120 | if not access_token: |
| | 121 | raise ValueError(data.get('error_description', 'Failed to obtain access token.')) |
| | 122 | return access_token, refresh_token |
| | 123 | |
| | 124 | |
| | 125 | def refresh_tokens(refresh_token): |
| | 126 | """Obtain a new access token using a refresh token.""" |
| | 127 | payload = { |
| | 128 | 'grant_type': 'refresh_token', |
| | 129 | 'refresh_token': refresh_token, |
| | 130 | 'client_id': config.CLIENT_ID, |
| | 131 | 'client_secret': config.CLIENT_SECRET |
| | 132 | } |
| | 133 | headers = { |
| | 134 | 'Content-Type': 'application/x-www-form-urlencoded', |
| | 135 | 'User-Agent': config.USER_AGENT |
| | 136 | } |
| | 137 | response = requests.post(config.TOKEN_URL, data=payload, headers=headers) |
| | 138 | response.raise_for_status() |
| | 139 | data = response.json() |
| | 140 | return data.get('access_token'), data.get('refresh_token') |
| | 141 | |
| | 142 | |
| | 143 | def api_get(path, access_token, params=None): |
| | 144 | """Perform an authenticated GET request to the Telldus OAuth2 API.""" |
| | 145 | headers = { |
| | 146 | 'Authorization': f'Bearer {access_token}', |
| | 147 | 'User-Agent': config.USER_AGENT |
| | 148 | } |
| | 149 | url = f'{config.API_BASE_URL}/{path}' |
| | 150 | response = requests.get(url, headers=headers, params=params or {}) |
| | 151 | response.raise_for_status() |
| | 152 | return response.json() |
| | 153 | |
| | 154 | |
| | 155 | def print_debug_message(message): |
| | 156 | print(f'[DEBUG] {message}') |
| | 157 | |
| | 158 | |
| | 159 | # --------------------------------------------------------------------------- |
| | 160 | # Routes |
| | 161 | # --------------------------------------------------------------------------- |
| | 162 | |
| | 163 | @app.route('/') |
| | 164 | def index(): |
| | 165 | access_token = session.get('access_token') |
| | 166 | print_debug_message('Accessing home page.') |
| | 167 | |
| | 168 | if not access_token: |
| | 169 | print_debug_message('No access token found. User not authenticated.') |
| | 170 | return render_template('index.html', authenticated=False) |
| | 171 | |
| | 172 | # Example API call: list devices |
| | 173 | params = { |
| | 174 | 'supportedMethods': 65535, |
| | 175 | 'format': 'json' |
| | 176 | } |
| | 177 | print_debug_message(f'Making API call to {config.API_BASE_URL}/devices/list') |
| | 178 | |
| | 179 | try: |
| | 180 | result = api_get('devices/list', access_token, params) |
| | 181 | devices = result.get('device', []) |
| | 182 | print_debug_message(f'Devices fetched: {json.dumps(devices, indent=2)}') |
| | 183 | except Exception as e: |
| | 184 | flash('An error occurred while fetching devices.', 'danger') |
| | 185 | print_debug_message(f'Exception during API call: {str(e)}') |
| | 186 | devices = [] |
| | 187 | |
| | 188 | return render_template('index.html', |
| | 189 | authenticated=True, |
| | 190 | devices=devices, |
| | 191 | access_token=access_token) |
| | 192 | |
| | 193 | |
| | 194 | @app.route('/login', methods=['GET', 'POST']) |
| | 195 | def login(): |
| | 196 | if request.method == 'POST': |
| | 197 | username = request.form.get('username', '').strip() |
| | 198 | password = request.form.get('password', '').strip() |
| | 199 | |
| | 200 | if not username or not password: |
| | 201 | flash('Username and password are required.', 'warning') |
| | 202 | return redirect(url_for('login')) |
| | 203 | |
| | 204 | print_debug_message(f'Attempting OAuth2 login for user: {username}') |
| | 205 | try: |
| | 206 | access_token, refresh_token = obtain_tokens(username, password) |
| | 207 | print_debug_message('OAuth2 tokens obtained successfully.') |
| | 208 | except Exception as e: |
| | 209 | flash(f'Login failed: {e}', 'danger') |
| | 210 | print_debug_message(f'Login failed: {e}') |
| | 211 | return redirect(url_for('login')) |
| | 212 | |
| | 213 | session['access_token'] = access_token |
| | 214 | session['refresh_token'] = refresh_token |
| | 215 | flash('You have successfully logged in.', 'success') |
| | 216 | return redirect(url_for('index')) |
| | 217 | |
| | 218 | return render_template('index.html', authenticated=False, show_login_form=True) |
| | 219 | |
| | 220 | |
| | 221 | @app.route('/logout') |
| | 222 | def logout(): |
| | 223 | print_debug_message('Logging out user.') |
| | 224 | session.clear() |
| | 225 | flash('You have been logged out.', 'success') |
| | 226 | return redirect(url_for('index')) |
| | 227 | |
| | 228 | |
| | 229 | @app.route('/devices') |
| | 230 | def devices(): |
| | 231 | access_token = session.get('access_token') |
| | 232 | print_debug_message('Accessing devices route.') |
| | 233 | |
| | 234 | if not access_token: |
| | 235 | flash('You need to log in first.', 'warning') |
| | 236 | return redirect(url_for('index')) |
| | 237 | |
| | 238 | params = {'supportedMethods': 65535, 'format': 'json'} |
| | 239 | try: |
| | 240 | result = api_get('devices/list', access_token, params) |
| | 241 | devices = result.get('device', []) |
| | 242 | except Exception as e: |
| | 243 | flash('Failed to fetch devices.', 'danger') |
| | 244 | print_debug_message(f'Exception: {e}') |
| | 245 | devices = [] |
| | 246 | |
| | 247 | return render_template('index.html', |
| | 248 | authenticated=True, |
| | 249 | devices=devices, |
| | 250 | show_devices=True) |
| | 251 | |
| | 252 | |
| | 253 | @app.route('/clients') |
| | 254 | def clients(): |
| | 255 | access_token = session.get('access_token') |
| | 256 | print_debug_message('Accessing clients route.') |
| | 257 | |
| | 258 | if not access_token: |
| | 259 | flash('You need to log in first.', 'warning') |
| | 260 | return redirect(url_for('index')) |
| | 261 | |
| | 262 | params = {'format': 'json'} |
| | 263 | try: |
| | 264 | result = api_get('clients/list', access_token, params) |
| | 265 | clients = result.get('client', []) |
| | 266 | except Exception as e: |
| | 267 | flash('Failed to fetch clients.', 'danger') |
| | 268 | print_debug_message(f'Exception: {e}') |
| | 269 | clients = [] |
| | 270 | |
| | 271 | return render_template('index.html', |
| | 272 | authenticated=True, |
| | 273 | clients=clients, |
| | 274 | show_clients=True) |
| | 275 | |
| | 276 | |
| | 277 | if __name__ == '__main__': |
| | 278 | app.secret_key = config.SECRET_KEY |
| | 279 | app.run(debug=True) |
| | 280 | }}} |
| | 281 | |
| | 282 | **The HTML Template** |
| | 283 | |
| | 284 | The home page also serves as the login form and results view. |
| | 285 | |
| | 286 | `templates/index.html` |
| | 287 | |
| | 288 | {{{ |
| | 289 | <!DOCTYPE html> |
| | 290 | <html lang="en"> |
| | 291 | <head> |
| | 292 | <meta charset="UTF-8"> |
| | 293 | <title>Telldus OAuth2 Flask App</title> |
| | 294 | <link rel="stylesheet" href="https://cdn.jsdelivr.net/npm/bootstrap@5.3.0/dist/css/bootstrap.min.css"> |
| | 295 | </head> |
| | 296 | <body> |
| | 297 | <div class="container mt-5"> |
| | 298 | |
| | 299 | {% with messages = get_flashed_messages(with_categories=true) %} |
| | 300 | {% if messages %} |
| | 301 | {% for category, message in messages %} |
| | 302 | <div class="alert alert-{{ category }} alert-dismissible fade show" role="alert"> |
| | 303 | {{ message }} |
| | 304 | <button type="button" class="btn-close" data-bs-dismiss="alert" aria-label="Close"></button> |
| | 305 | </div> |
| | 306 | {% endfor %} |
| | 307 | {% endif %} |
| | 308 | {% endwith %} |
| | 309 | |
| | 310 | <h1 class="mb-4">Telldus Live! OAuth2 Integration</h1> |
| | 311 | |
| | 312 | {% if not authenticated %} |
| | 313 | |
| | 314 | {% if show_login_form %} |
| | 315 | <p>Enter your Telldus Live! credentials to connect.</p> |
| | 316 | <form method="POST" action="{{ url_for('login') }}"> |
| | 317 | <div class="mb-3"> |
| | 318 | <label for="username" class="form-label">Username (email)</label> |
| | 319 | <input type="email" class="form-control" id="username" name="username" required> |
| | 320 | </div> |
| | 321 | <div class="mb-3"> |
| | 322 | <label for="password" class="form-label">Password</label> |
| | 323 | <input type="password" class="form-control" id="password" name="password" required> |
| | 324 | </div> |
| | 325 | <button type="submit" class="btn btn-primary">Connect with Telldus</button> |
| | 326 | </form> |
| | 327 | {% else %} |
| | 328 | <p>You are not connected to Telldus Live!.</p> |
| | 329 | <a href="{{ url_for('login') }}" class="btn btn-primary">Connect with Telldus</a> |
| | 330 | {% endif %} |
| | 331 | |
| | 332 | {% else %} |
| | 333 | <p>You are connected to Telldus Live!</p> |
| | 334 | <a href="{{ url_for('logout') }}" class="btn btn-danger">Logout</a> |
| | 335 | |
| | 336 | <hr> |
| | 337 | |
| | 338 | {% if show_devices or devices is defined %} |
| | 339 | <h2>Devices</h2> |
| | 340 | {% if devices %} |
| | 341 | <ul class="list-group"> |
| | 342 | {% for device in devices %} |
| | 343 | <li class="list-group-item"> |
| | 344 | <strong>{{ device.name }}</strong> (ID: {{ device.id }}) |
| | 345 | </li> |
| | 346 | {% endfor %} |
| | 347 | </ul> |
| | 348 | {% else %} |
| | 349 | <p>No devices found.</p> |
| | 350 | {% endif %} |
| | 351 | {% endif %} |
| | 352 | |
| | 353 | {% if show_clients and clients is defined %} |
| | 354 | <h2>Clients</h2> |
| | 355 | {% if clients %} |
| | 356 | <ul class="list-group"> |
| | 357 | {% for client in clients %} |
| | 358 | <li class="list-group-item"> |
| | 359 | <strong>{{ client.name }}</strong> (ID: {{ client.id }}) |
| | 360 | </li> |
| | 361 | {% endfor %} |
| | 362 | </ul> |
| | 363 | {% else %} |
| | 364 | <p>No clients found.</p> |
| | 365 | {% endif %} |
| | 366 | {% endif %} |
| | 367 | |
| | 368 | {% if not show_devices and not show_clients %} |
| | 369 | <h2>Quick Links</h2> |
| | 370 | <a href="{{ url_for('devices') }}" class="btn btn-info me-2">View Devices</a> |
| | 371 | <a href="{{ url_for('clients') }}" class="btn btn-info">View Clients</a> |
| | 372 | {% endif %} |
| | 373 | |
| | 374 | {% endif %} |
| | 375 | </div> |
| | 376 | |
| | 377 | <script src="https://cdn.jsdelivr.net/npm/bootstrap@5.3.0/dist/js/bootstrap.bundle.min.js"></script> |
| | 378 | </body> |
| | 379 | </html> |
| | 380 | }}} |
| | 381 | |
| | 382 | == Command-Line Alternative |
| | 383 | |
| | 384 | If you only need to obtain tokens without a web interface, you can use a standalone script. The repository includes `oauth2.py` for this purpose. |
| | 385 | |
| | 386 | Create a `config.ini` file: |
| | 387 | |
| | 388 | {{{ |
| | 389 | [oauth2] |
| | 390 | client_id = Your client ID |
| | 391 | client_secret = Your client secret |
| | 392 | username = your@email.com |
| | 393 | password = your_password |
| | 394 | |
| | 395 | [server] |
| | 396 | user_agent = MyTelldusApp/1.0 |
| | 397 | }}} |
| | 398 | |
| | 399 | Run the script: |
| | 400 | |
| | 401 | {{{ |
| | 402 | python oauth2.py |
| | 403 | }}} |
| | 404 | |
| | 405 | On success, the access token and refresh token are printed to the console. |
| | 406 | |
| | 407 | == Testing an API Call from the Command Line |
| | 408 | |
| | 409 | Once you have an access token, you can test an API call directly with curl: |
| | 410 | |
| | 411 | {{{ |
| | 412 | curl -X GET "https://pa-api.telldus.com/oauth2/devices/list?supportedMethods=65535&format=json" \ |
| | 413 | -H "Authorization: Bearer YOUR_ACCESS_TOKEN" \ |
| | 414 | -H "User-Agent: MyTelldusApp/1.0" |
| | 415 | }}} |
| | 416 | |
| | 417 | Replace `YOUR_ACCESS_TOKEN` with the token obtained from the login flow or the command-line script. |