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.

main.py

"""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())

redditmodel.py

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.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))

redditwrapper.py

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))