forked from cenkalti/github-flask
-
Notifications
You must be signed in to change notification settings - Fork 1
/
example.py
147 lines (106 loc) · 3.55 KB
/
example.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
"""
GitHub Example
--------------
Shows how to authorize users with Github.
pip install sqlalchemy
"""
from quart import Quart, request, g, session, redirect, url_for
from quart import render_template_string, jsonify
from quart_github import GitHub
from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.orm import scoped_session, sessionmaker
from sqlalchemy.ext.declarative import declarative_base
DATABASE_URI = 'sqlite:////tmp/github-quart.db'
SECRET_KEY = 'development key'
DEBUG = True
# Set these values
GITHUB_CLIENT_ID = 'XXX'
GITHUB_CLIENT_SECRET = 'YYY'
# setup quart
app = Quart(__name__)
app.config.from_object(__name__)
# setup github-quart
github = GitHub(app)
# setup sqlalchemy
engine = create_engine(app.config['DATABASE_URI'])
db_session = scoped_session(sessionmaker(autocommit=False,
autoflush=False,
bind=engine))
Base = declarative_base()
Base.query = db_session.query_property()
def init_db():
Base.metadata.create_all(bind=engine)
class User(Base):
__tablename__ = 'users'
id = Column(Integer, primary_key=True)
github_access_token = Column(String(255))
github_id = Column(Integer)
github_login = Column(String(255))
def __init__(self, github_access_token):
self.github_access_token = github_access_token
@app.before_request
async def before_request():
g.user = None
if 'user_id' in session:
g.user = User.query.get(session['user_id'])
@app.after_request
async def after_request(response):
db_session.remove()
return response
@app.route('/')
async def index():
if g.user:
t = 'Hello! %s <a href="{{ url_for("user") }}">Get user</a> ' \
'<a href="{{ url_for("repo") }}">Get repo</a> ' \
'<a href="{{ url_for("logout") }}">Logout</a>'
t %= g.user.github_login
else:
t = 'Hello! <a href="{{ url_for("login") }}">Login</a>'
return await render_template_string(t)
@github.access_token_getter
async def token_getter():
user = g.user
if user is not None:
return user.github_access_token
@app.route('/github-callback')
@github.authorized_handler
async def authorized(access_token):
next_url = request.args.get('next') or url_for('index')
if access_token is None:
return redirect(next_url)
user = User.query.filter_by(github_access_token=access_token).first()
if user is None:
user = User(access_token)
db_session.add(user)
user.github_access_token = access_token
# Not necessary to get these details here,
# but it helps humans to identify users easily.
g.user = user
github_user = await github.get('/user')
if not isinstance(github_user, dict):
# must be a response with an error
return github_user.status_code, github_user.content
user.github_id = github_user['id']
user.github_login = github_user['login']
db_session.commit()
session['user_id'] = user.id
return redirect(next_url)
@app.route('/login')
async def login():
if session.get('user_id', None) is None:
return github.authorize()
else:
return 'Already logged in'
@app.route('/logout')
async def logout():
session.pop('user_id', None)
return redirect(url_for('index'))
@app.route('/user')
async def user():
return jsonify(await github.get('/user'))
@app.route('/repo')
async def repo():
return jsonify(await github.get('/repos/luckydonald/github-quart'))
if __name__ == '__main__':
init_db()
app.run(debug=True)