Reddit Example¶
Demonstrates authenticating with OAuth 2 to access Reddit.
The Reddit example uses OAuth 2, as supported by Qt Network Authorization, to sign in to Reddit and display the Reddit posts (in text format) associated with the authenticated user.
To use this example, a consumer key from Reddit is needed. To register the application visit https://www.reddit.com/prefs/apps/.
Note
Choose installed app when creating the application.
Note
Set the redirect URI to http://127.0.0.1:1337/ in Reddit settings.
# Copyright (C) 2022 The Qt Company Ltd.
# SPDX-License-Identifier: LicenseRef-Qt-Commercial OR BSD-3-Clause
from __future__ import annotations
"""PySide6 port of the networkauth redditclient example from Qt v6.x"""
from argparse import ArgumentParser, RawTextHelpFormatter
import sys
from PySide6.QtWidgets import QApplication, QListView
from redditmodel import RedditModel
if __name__ == '__main__':
parser = ArgumentParser(description='Qt Reddit client example',
formatter_class=RawTextHelpFormatter)
parser.add_argument('--client', '-i', type=str, help='Client id')
options = parser.parse_args()
if not options.client:
print('Specify a client id', file=sys.stderr)
sys.exit(-1)
app = QApplication(sys.argv)
view = QListView()
model = RedditModel(options.client)
view.setModel(model)
view.show()
sys.exit(app.exec())
# Copyright (C) 2022 The Qt Company Ltd.
# SPDX-License-Identifier: LicenseRef-Qt-Commercial OR BSD-3-Clause
from __future__ import annotations
import functools
from PySide6.QtCore import (QAbstractTableModel, QJsonDocument, QModelIndex,
Qt, Signal, Slot)
from PySide6.QtNetwork import QNetworkReply
from redditwrapper import RedditWrapper
class RedditModel(QAbstractTableModel):
error = Signal(str)
def __init__(self, client_id):
super().__init__()
self._reddit_wrapper = RedditWrapper(client_id)
self._reddit_wrapper.authenticated.connect(self.update)
self._live_thread_reply = None
self._threads = []
self.grant()
def rowCount(self, parent):
return len(self._threads)
def columnCount(self, parent):
return 1 if self._threads else 0
def data(self, index, role):
if not index.isValid():
return None
if role == Qt.ItemDataRole.DisplayRole:
children_object = self._threads[index.row()]
data_object = children_object["data"]
return data_object["title"]
return None
def grant(self):
self._reddit_wrapper.grant()
@Slot(QNetworkReply)
def reply_finished(self, reply):
reply.deleteLater()
if reply.error() != QNetworkReply.NoError:
error = reply.errorString()
print(f"Reddit error: {error}")
self.error.emit(error)
return
json = reply.readAll()
document = QJsonDocument.fromJson(json)
root_object = document.object()
kind = root_object["kind"]
assert kind == "Listing"
data_object = root_object["data"]
children_array = data_object["children"]
if not children_array:
return
self.beginInsertRows(QModelIndex(), len(self._threads),
len(children_array) + len(self._threads) - 1)
for childValue in children_array:
self._threads.append(childValue)
self.endInsertRows()
@Slot()
def update(self):
reply = self._reddit_wrapper.request_hot_threads()
reply.finished.connect(functools.partial(self.reply_finished,
reply=reply))
# Copyright (C) 2022 The Qt Company Ltd.
# SPDX-License-Identifier: LicenseRef-Qt-Commercial OR BSD-3-Clause
from __future__ import annotations
import functools
from PySide6.QtCore import QJsonDocument, QObject, QUrl, Signal, Slot
from PySide6.QtGui import QDesktopServices
from PySide6.QtNetwork import QNetworkReply
from PySide6.QtNetworkAuth import (QAbstractOAuth,
QOAuth2AuthorizationCodeFlow,
QOAuthHttpServerReplyHandler)
AUTHORIZATION_URL = "https://www.reddit.com/api/v1/authorize"
ACCESSTOKEN_URL = "https://www.reddit.com/api/v1/access_token"
NEW_URL = "https://oauth.reddit.com/new"
HOT_URL = "https://oauth.reddit.com/hot"
LIVE_THREADS_URL = "https://oauth.reddit.com/live/XXXX/about.json"
class RedditWrapper(QObject):
authenticated = Signal()
subscribed = Signal(QUrl)
def __init__(self, clientIdentifier, parent=None):
super().__init__(parent)
self._oauth2 = QOAuth2AuthorizationCodeFlow()
self._oauth2.setClientIdentifier(clientIdentifier)
self._reply_handler = QOAuthHttpServerReplyHandler(1337, self)
self._oauth2.setReplyHandler(self._reply_handler)
self._oauth2.setAuthorizationUrl(QUrl(AUTHORIZATION_URL))
self._oauth2.setAccessTokenUrl(QUrl(ACCESSTOKEN_URL))
self._oauth2.setScope("identity read")
self._permanent = True
# connect to slots
self._oauth2.statusChanged.connect(self.status_changed)
self._oauth2.authorizeWithBrowser.connect(QDesktopServices.openUrl)
def modify_parameters_function(stage, parameters):
if stage == QAbstractOAuth.Stage.RequestingAuthorization and self.permanent:
parameters["duration"] = "permanent"
return parameters
self._oauth2.setModifyParametersFunction(modify_parameters_function)
@Slot()
def status_changed(self, status):
if status == QAbstractOAuth.Status.Granted:
self.authenticated.emit()
def request_hot_threads(self):
print("Getting hot threads...")
return self._oauth2.get(QUrl(HOT_URL))
@property
def permanent(self):
return self._permanent
@permanent.setter
def permanent(self, value: bool):
self._permanent = value
def grant(self):
self._oauth2.grant()
@Slot(QNetworkReply)
def reply_finished(self, reply):
print('RedditWrapper.reply_finished()', reply.error())
reply.deleteLater()
if reply.error() != QNetworkReply.NoError:
error = reply.errorString()
print(f"Reddit error: {error}")
return
json = reply.readAll()
document = QJsonDocument.fromJson(json)
assert document.isObject()
root_object = document.object()
data_object = root_object["data"]
websocketUrl = QUrl(data_object["websocket_url"])
self.subscribed.emit(websocketUrl)
def subscribe_to_live_updates(self):
print("Susbscribing...")
reply = self._oauth2.get(QUrl(LIVE_THREADS_URL))
reply.finished.connect(functools.partial(self.reply_finished,
reply=reply))