How to use the prawcore.Requestor function in prawcore

To help you get started, we’ve selected a few prawcore examples, based on popular ways it is used in public projects.

Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.

github praw-dev / praw / tests / unit / test_reddit.py View on Github external
def test_requestor_class(self):
        class CustomRequestor(Requestor):
            pass

        reddit = Reddit(
            client_id="dummy",
            client_secret="dummy",
            password="dummy",
            user_agent="dummy",
            username="dummy",
            requestor_class=CustomRequestor,
        )
        assert isinstance(reddit._core._requestor, CustomRequestor)
        assert not isinstance(self.reddit._core._requestor, CustomRequestor)

        reddit = Reddit(
            client_id="dummy",
            client_secret="dummy",
github praw-dev / prawcore / tests / test_requestor.py View on Github external
def test_initialize(self):
        requestor = prawcore.Requestor("prawcore:test (by /u/bboe)")
        self.assertEqual(
            "prawcore:test (by /u/bboe) prawcore/{}".format(
                prawcore.__version__
            ),
            requestor._http.headers["User-Agent"],
        )
github praw-dev / prawcore / tests / test_sessions.py View on Github external
def test_request__read_timeout_retry(self, mock_session):
        session_instance = mock_session.return_value

        # Handle Auth
        response_dict = {"access_token": "", "expires_in": 99, "scope": ""}
        session_instance.request.return_value = Mock(
            headers={}, json=lambda: response_dict, status_code=200
        )
        requestor = prawcore.Requestor("prawcore:test (by /u/bboe)")
        authorizer = readonly_authorizer(requestor=requestor)
        session_instance.request.reset_mock()

        # Fail on subsequent request
        exception = ReadTimeout()
        session_instance.request.side_effect = exception

        expected = (
            "prawcore",
            "WARNING",
            "Retrying due to ReadTimeout() status: GET "
            "https://oauth.reddit.com/",
        )

        with LogCapture(level=logging.WARNING) as log_capture:
            with self.assertRaises(RequestException) as context_manager:
github praw-dev / prawcore / tests / test_sessions.py View on Github external
def test_request__chunked_encoding_retry(self, mock_session):
        session_instance = mock_session.return_value

        # Handle Auth
        response_dict = {"access_token": "", "expires_in": 99, "scope": ""}
        session_instance.request.return_value = Mock(
            headers={}, json=lambda: response_dict, status_code=200
        )
        requestor = prawcore.Requestor("prawcore:test (by /u/bboe)")
        authorizer = readonly_authorizer(requestor=requestor)
        session_instance.request.reset_mock()

        # Fail on subsequent request
        exception = ChunkedEncodingError()
        session_instance.request.side_effect = exception

        expected = (
            "prawcore",
            "WARNING",
            "Retrying due to ChunkedEncodingError() status: GET "
            "https://oauth.reddit.com/",
        )

        with LogCapture(level=logging.WARNING) as log_capture:
            with self.assertRaises(RequestException) as context_manager:
github praw-dev / prawcore / tests / test_requestor.py View on Github external
def test_request__use_custom_session(self):
        override = "REQUEST OVERRIDDEN"
        custom_header = "CUSTOM SESSION HEADER"
        headers = {"session_header": custom_header}
        attrs = {"request.return_value": override, "headers": headers}
        session = Mock(**attrs)

        requestor = prawcore.Requestor(
            "prawcore:test (by /u/bboe)", session=session
        )

        self.assertEqual(
            "prawcore:test (by /u/bboe) prawcore/{}".format(
                prawcore.__version__
            ),
            requestor._http.headers["User-Agent"],
        )
        self.assertEqual(
            requestor._http.headers["session_header"], custom_header
        )

        self.assertEqual(requestor.request("https://reddit.com"), override)
github praw-dev / prawcore / examples / obtain_refresh_token.py View on Github external
def main():
    """Provide the program's entry point when directly executed."""
    if len(sys.argv) < 2:
        print("Usage: {} SCOPE...".format(sys.argv[0]))
        return 1

    authenticator = prawcore.TrustedAuthenticator(
        prawcore.Requestor("prawcore_refresh_token_example"),
        os.environ["PRAWCORE_CLIENT_ID"],
        os.environ["PRAWCORE_CLIENT_SECRET"],
        os.environ["PRAWCORE_REDIRECT_URI"],
    )

    state = str(random.randint(0, 65000))
    url = authenticator.authorize_url("permanent", sys.argv[1:], state)
    print(url)

    client = receive_connection()
    data = client.recv(1024).decode("utf-8")
    param_tokens = data.split(" ", 2)[1].split("?", 1)[1].split("&")
    params = {
        key: value
        for (key, value) in [token.split("=") for token in param_tokens]
    }
github praw-dev / prawcore / examples / script_auth_friend_list.py View on Github external
def main():
    """Provide the program's entry point when directly executed."""
    authenticator = prawcore.TrustedAuthenticator(
        prawcore.Requestor("prawcore_script_auth_example"),
        os.environ["PRAWCORE_CLIENT_ID"],
        os.environ["PRAWCORE_CLIENT_SECRET"],
    )
    authorizer = prawcore.ScriptAuthorizer(
        authenticator,
        os.environ["PRAWCORE_USERNAME"],
        os.environ["PRAWCORE_PASSWORD"],
    )
    authorizer.refresh()

    with prawcore.session(authorizer) as session:
        data = session.request("GET", "/api/v1/me/friends")

    for friend in data["data"]["children"]:
        print(friend["name"])
github praw-dev / prawcore / examples / caching_requestor.py View on Github external
def main():
    """Provide the program's entry point when directly executed."""
    if len(sys.argv) != 2:
        print("Usage: {} USERNAME".format(sys.argv[0]))
        return 1

    caching_requestor = prawcore.Requestor(
        "prawcore_device_id_auth_example", session=CachingSession()
    )
    authenticator = prawcore.TrustedAuthenticator(
        caching_requestor,
        os.environ["PRAWCORE_CLIENT_ID"],
        os.environ["PRAWCORE_CLIENT_SECRET"],
    )
    authorizer = prawcore.ReadOnlyAuthorizer(authenticator)
    authorizer.refresh()

    user = sys.argv[1]
    with prawcore.session(authorizer) as session:
        data1 = session.request("GET", "/api/v1/user/{}/trophies".format(user))

    with prawcore.session(authorizer) as session:
        data2 = session.request("GET", "/api/v1/user/{}/trophies".format(user))