-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathmain.py
286 lines (228 loc) · 8.53 KB
/
main.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
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
# -*- coding: utf-8 -*-
"""Entry point for the Pins4Days app that runs on Google App Engine (standard).
This app utilizes:
- Flask framework
- flask-login for user session management
- NDB client library for connecting to Google Cloud Datastore (NoSQL!), and
storing pins and user info.
- Google Cloud Storage (GCS) for storing configs
- Memcache for session storage
Attributes:
app (obj): Flask app.
config (AppConfig): An object that handles reading in configs from GCS.
These configs are inserted into app.config and made available to the Flask
app.
login_manager (LoginManager): User session manager.
Todo:
* Handle duplicate user creation in signup().
* Investigate possible exceptions for User creation and add exception
handling to signup().
* Handle pagination, aka return more than 10 pins from the /pins view and
/api/pins endpoint!
* Read in existing pins and store them.
* Add auth to GET /api/pins.
"""
import logging
import json
from flask import Flask
from flask import request
from flask import jsonify
from flask import redirect
from flask import make_response
from flask import url_for
from flask import render_template
from flask_login import login_user
from flask_login import LoginManager
from flask_login import login_required
from flask_login import current_user
from werkzeug.contrib.cache import MemcachedCache
from werkzeug.urls import Href
from google.appengine.api import taskqueue
from pins4days.constants import KEY_FLASK_APP_CONFIG
from pins4days.constants import KEY_FLASK_SECRET_KEY
from pins4days.utils import load_config
from pins4days.utils import get_channel_pins
from pins4days.event import PinnedMessage
from pins4days.models.pin import Pin
from pins4days.models.user import User
from pins4days.models.exceptions import EntityDoesNotExistException
from pins4days.models.exceptions import IncorrectPasswordException
from pins4days.appuser import AppUser
app = Flask(__name__)
config = load_config()
app.config.update(config[KEY_FLASK_APP_CONFIG])
app.config['SESSION_TYPE'] = 'memcached'
app.config['SESSION_MEMCACHED'] = MemcachedCache()
app.secret_key = config[KEY_FLASK_SECRET_KEY]
login_manager = LoginManager()
login_manager.init_app(app)
@login_manager.user_loader
def load_user(username):
"""Loads a User model instance based on their unique username.
Args:
username (str): The user's username which they created upon sign up.
Returns:
AppUser or None: Returns an AppUser if the User exists in the DB.
Returns None if the user does not exist.
"""
user = User.get_by_id(username)
return AppUser(user) if user else None
@app.route('/signup', methods=['POST', 'GET'])
def signup():
"""Handles user sign up. This requires that the user create a unique
username, and a password.
If the user is already logged in, redirect them to the main /pins page.
If the user is not logged in, show them the signup form.
If they've submitted the signup form, create a new user, and then redirect
them to the /login page.
Returns:
Response: See flow described above.
"""
if current_user.is_authenticated:
return redirect(url_for('pins'))
if request.method == 'GET':
return render_template('signup.html')
username = request.form['username']
password = request.form['password']
User.create_with_encryption(id=username, password=password)
return redirect(url_for('login'))
@app.route('/login', methods=['GET', 'POST'])
def login():
"""Logs in user, creating a session in memcache.
If the user is already logged in, redirect them the /pins page.
If the user is not logged in, show them the login form. See
handle_login_post() for more details on this step..
Returns:
Response: See flow described above.
"""
if current_user.is_authenticated:
return redirect(url_for('pins'))
if request.method == 'GET':
return render_template('login.html')
return handle_login_post(request)
def handle_login_post(request):
"""Validates user form data and logs in user.
If the form is submitted with invalid inputs, show the error template.
If the user attempts to log in as a user that does not exist, show the error
template.
If the user attempts to log in with an existing username and incorrect
password, show the error template.
Otherwise, if the form was valid and the username exists with the correct
password submitted, that user is logged in, and redirected to the /pins
page.
Args:
request (Request): HTTP request.
Returns:
Response:
"""
if not validate_form(request.form):
return make_response(
render_template('login.html', error='E_BAD_FORM'), 400)
try:
user = User.login(request.form['username'], request.form['password'])
app_user = AppUser(user)
if app_user:
login_user(app_user)
return redirect(url_for('pins'), 302)
except EntityDoesNotExistException as e:
return make_response(
render_template('login.html', error='E_ENTITY_DOES_NOT_EXIST'), 404)
except IncorrectPasswordException as e:
return make_response(
render_template('login.html', error='E_INCORRECT_PASSWORD'), 400)
def validate_form(form):
"""Executes simple form validation. Checks for None or empty string values.
Args:
form (MultiDict): The form data that was POST'd.
Returns:
bool: If username or password contain None or empty strings, False is
returned. Otherwise, True is returned.
"""
return (form['username'].replace(' ', '') and form['username'] is not None and
form['password'].replace(' ', '') and form['password'] is not None)
@app.route('/pins', methods=['GET'])
@login_required
def pins():
"""Renders the /pins page template.
Returns:
TYPE: Description
"""
username = current_user.username
limit = request.args.get('limit', 10)
page = int(request.args.get('page', 1))
offset = (page - 1) * limit
href = Href(url_for('pins'))
next_url = href({'page': page + 1})
return render_template(
'pins.html',
username=username,
next_url=next_url,
pins=Pin.query_all().fetch(limit, offset=offset))
@app.route('/api/pins', methods=['POST', 'GET'])
def api_pins():
"""Fetches or creates Pins.
See handle_api_pins_post() and handle_api_pins_get() for more details.
Returns:
Response:
"""
if request.method == 'POST':
return handle_api_pins_post(request)
elif request.method == 'GET':
return handle_api_pins_get(request)
@app.route('/channels/<channel_id>/pins/enqueue', methods=['GET'])
@login_required
def channels_pins_enqueue(channel_id):
if request.method == 'GET':
resp = get_channel_pins(channel_id, app.config['slack_user_token'])
pins = resp['items']
for pin in pins:
task = taskqueue.add(
url='/worker/create_pin',
target='worker',
payload=json.dumps(pin),
method='POST')
return make_response('', 200)
def handle_api_pins_post(request):
"""Handles POST requests to /api/pins.
If the JSON POST request body contains the 'challenge' or 'token' keys,
perform Slack's URL verification handshake. For more details, see:
https://api.slack.com/events-api#url_verification.
Otherwise, create a Pin (and its attachments).
Args:
request (Request):
Returns:
Response:
"""
json = request.get_json()
if ('token' not in json or
json['token'] != app.config['slack_verification_token']):
return make_response(
jsonify(message='Missing or unrecognized token.'),
401)
if 'challenge' in json:
# Handle the very first and only auth call.
challenge = json['challenge']
return jsonify(challenge=challenge)
pin = PinnedMessage.factory(json)
pin.put()
return make_response('', 201)
def handle_api_pins_get(request):
"""Handles GET requests to /api/pins.
If 'user_id' query param is set, pins for that user are returned.
Otherwise, the most recently pinned messages are returned.
Args:
request (Request):
Returns:
Response:
"""
user_id = request.args.get('user_id')
if user_id:
pins = Pin.query_user(user_id).fetch(10)
else:
pins = Pin.query_all().fetch(10)
response = {
'data': {
'pins': [pin.to_dict() for pin in pins]
}
}
return jsonify(response)