main.py

import base64
import hashlib
import os
import re
import json
import requests
import redis
import ssl
from requests.auth import AuthBase, HTTPBasicAuth
from requests_oauthlib import OAuth2Session, TokenUpdated
from flask import Flask, request, redirect, session, url_for, render_template
from dotenv import load_dotenv

load_dotenv()  # take environment variables from .env.
redis_url = os.getenv('redis_url')

if redis_url:  # run redis on e.g. cloud provider
    print("Running external Redis...")
    r = redis.Redis.from_url(redis_url, ssl_cert_reqs=ssl.CERT_NONE)
else:  # run locally
    print("Running Redis on localhost...")
    r = redis.StrictRedis(host='localhost', port=6379, db=0)


app = Flask(__name__)
app.secret_key = os.urandom(50)


client_id = os.getenv('CLIENT_ID')
client_secret = os.getenv('CLIENT_ID_SECRET')
auth_url = "https://twitter.com/i/oauth2/authorize"
token_url = "https://api.twitter.com/2/oauth2/token"
redirect_uri = os.getenv('redirect_uri')

scopes = ["tweet.read", "users.read", "tweet.write", "offline.access"]

code_verifier = base64.urlsafe_b64encode(os.urandom(30)).decode("utf-8")
code_verifier = re.sub("[^a-zA-Z0-9]+", "", code_verifier)

code_challenge = hashlib.sha256(code_verifier.encode("utf-8")).digest()
code_challenge = base64.urlsafe_b64encode(code_challenge).decode("utf-8")
code_challenge = code_challenge.replace("=", "")


def make_token():
    return OAuth2Session(client_id, redirect_uri=redirect_uri, scope=scopes)


def post_tweet(payload, token):
    print("Tweeting!")
    return requests.request(
        "POST",
        "https://api.twitter.com/2/tweets",
        json=payload,
        headers={
            "Authorization": "Bearer {}".format(token["access_token"]),
            "Content-Type": "application/json",
        },
    )


# Currently not available to free access/essential access for v2 API
def read_tweet(tweet_id, token):
    print("Reading!")
    return requests.request(
        "GET",
        f"https://api.twitter.com/2/tweets/{tweet_id}",
        headers={
            "Authorization": "Bearer {}".format(token["access_token"]),
            "Content-Type": "application/json",
        },
    )


@app.route("/")
def demo():
    global twitter
    twitter = make_token()
    authorization_url, state = twitter.authorization_url(
        auth_url, code_challenge=code_challenge, code_challenge_method="S256"
    )
    session["oauth_state"] = state
    return redirect(authorization_url)


@app.route("/oauth/callback", methods=["GET"])
def callback():
    code = request.args.get("code")
    token = twitter.fetch_token(
        token_url=token_url,
        client_secret=client_secret,
        code_verifier=code_verifier,
        code=code,
    )
    st_token = '"{}"'.format(token)
    j_token = json.loads(st_token)

    r.set("token", j_token)

    # Use auto_tweeter.py for the automated experience
    payload = {f"text": "This tweet was manually authenticated in browser"}
    response = post_tweet(payload, token).json()

    return response


if __name__ == "__main__":
    app.run()