Created
April 1, 2018 07:12
-
-
Save chrisroat/d2aa08b0b60ec83c64a47332357fd292 to your computer and use it in GitHub Desktop.
multi-user flask-dance with sqlalchemy via twitter
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
<!DOCTYPE html> | |
<head> | |
<title>Flask-Dance Multi-User SQLAlchemy</title> | |
</head> | |
<body> | |
{% with messages = get_flashed_messages(with_categories=true) %} | |
{% if messages %} | |
<ul class="flash"> | |
{% for category, message in messages %} | |
<li class="{{ category }}">{{ message }}</li> | |
{% endfor %} | |
</ul> | |
{% endif %} | |
{% endwith %} | |
{% if current_user.is_authenticated %} | |
You are logged in as {{ current_user.name }}! | |
<a href="{{ url_for("logout") }}">Log out</a> | |
{% else %} | |
You are not logged in. | |
<a href="{{ url_for("twitter.login") }}">Log in</a> | |
{% endif %} | |
</body> |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
import os | |
import sys | |
from flask import Flask, redirect, url_for, flash, render_template | |
from flask_sqlalchemy import SQLAlchemy | |
from sqlalchemy.orm.exc import NoResultFound | |
from flask_dance.contrib.twitter import make_twitter_blueprint, twitter | |
from flask_dance.consumer.backend.sqla import OAuthConsumerMixin, SQLAlchemyBackend | |
from flask_dance.consumer import oauth_authorized, oauth_error | |
from flask_login import ( | |
LoginManager, UserMixin, current_user, | |
login_required, login_user, logout_user | |
) | |
# setup Flask application | |
app = Flask(__name__) | |
app.secret_key = "supersekrit" | |
blueprint = make_twitter_blueprint( | |
api_key=os.environ['TWITTER_ID'], | |
api_secret=os.environ['TWITTER_SECRET'], | |
) | |
app.register_blueprint(blueprint, url_prefix="/login") | |
# setup database models | |
app.config["SQLALCHEMY_DATABASE_URI"] = "sqlite:///multi.db" | |
db = SQLAlchemy() | |
class User(db.Model, UserMixin): | |
id = db.Column(db.Integer, primary_key=True) | |
name = db.Column(db.String(256), unique=True) | |
class OAuth(OAuthConsumerMixin, db.Model): | |
provider_user_id = db.Column(db.String(256), unique=True) | |
user_id = db.Column(db.Integer, db.ForeignKey(User.id)) | |
user = db.relationship(User) | |
# setup login manager | |
login_manager = LoginManager() | |
login_manager.login_view = 'twitter.login' | |
@login_manager.user_loader | |
def load_user(user_id): | |
return User.query.get(int(user_id)) | |
# setup SQLAlchemy backend | |
blueprint.backend = SQLAlchemyBackend(OAuth, db.session, user=current_user) | |
# create/login local user on successful OAuth login | |
@oauth_authorized.connect_via(blueprint) | |
def twitter_logged_in(blueprint, token): | |
if not token: | |
flash("Failed to log in.", category="error") | |
return False | |
resp = blueprint.session.get("account/settings.json") | |
if not resp.ok: | |
msg = "Failed to fetch user info." | |
flash(msg, category="error") | |
return False | |
info = resp.json() | |
user_id = str(info["screen_name"]) | |
# Find this OAuth token in the database, or create it | |
query = OAuth.query.filter_by( | |
provider=blueprint.name, | |
provider_user_id=user_id, | |
) | |
try: | |
oauth = query.one() | |
except NoResultFound: | |
oauth = OAuth( | |
provider=blueprint.name, | |
provider_user_id=user_id, | |
token=token, | |
) | |
if oauth.user: | |
login_user(oauth.user) | |
flash("Successfully signed in.") | |
else: | |
# Create a new local user account for this user | |
user = User( | |
name=info["screen_name"], | |
) | |
# Associate the new local user account with the OAuth token | |
oauth.user = user | |
# Save and commit our database models | |
db.session.add_all([user, oauth]) | |
db.session.commit() | |
# Log in the new local user account | |
login_user(user) | |
flash("Successfully signed in.") | |
# Disable Flask-Dance's default behavior for saving the OAuth token | |
return False | |
# notify on OAuth provider error | |
@oauth_error.connect_via(blueprint) | |
def twitter_error(blueprint, error, error_description=None, error_uri=None): | |
msg = ( | |
"OAuth error from {name}! " | |
"error={error} description={description} uri={uri}" | |
).format( | |
name=blueprint.name, | |
error=error, | |
description=error_description, | |
uri=error_uri, | |
) | |
flash(msg, category="error") | |
@app.route("/logout") | |
@login_required | |
def logout(): | |
logout_user() | |
flash("You have logged out") | |
return redirect(url_for("index")) | |
@app.route("/") | |
def index(): | |
return render_template("home.html") | |
# hook up extensions to app | |
db.init_app(app) | |
login_manager.init_app(app) | |
if __name__ == "__main__": | |
if "--setup" in sys.argv: | |
with app.app_context(): | |
db.create_all() | |
db.session.commit() | |
print("Database tables created") | |
else: | |
app.run(debug=True) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment