Flasky

This repository contains the source code examples for my O'Reilly book [Flask Web Development](http://www.flaskbook.com).

The commits and tags in this repository were carefully created to match the sequence in which concepts are presented in the book. Please read the section titled "How to Work with the Example Code" in the book's preface for instructions.

from flask import Flask

from flask.ext.bootstrap import Bootstrap

from flask.ext.mail import Mail

from flask.ext.moment import Moment

from flask.ext.sqlalchemy import SQLAlchemy

from flask.ext.login import LoginManager

from flask.ext.pagedown import PageDown

from config import config

bootstrap = Bootstrap()

mail = Mail()

moment = Moment()

db = SQLAlchemy()

pagedown = PageDown()

login_manager = LoginManager()

login_manager.session_protection = 'strong'

login_manager.login_view = 'auth.login'

def create_app(config_name):

app = Flask(__name__)

app.config.from_object(config[config_name])

config[config_name].init_app(app)

bootstrap.init_app(app)

mail.init_app(app)

moment.init_app(app)

db.init_app(app)

login_manager.init_app(app)

pagedown.init_app(app)

if not app.debug and not app.testing and not app.config['SSL_DISABLE']:

from flask.ext.sslify import SSLify

sslify = SSLify(app)

from .main import main as main_blueprint

app.register_blueprint(main_blueprint)

from .auth import auth as auth_blueprint

app.register_blueprint(auth_blueprint, url_prefix='/auth')

from .api_1_0 import api as api_1_0_blueprint

app.register_blueprint(api_1_0_blueprint, url_prefix='/api/v1.0')

return app

from flask import Blueprint

api = Blueprint('api', __name__)

from . import authentication, users, errors

from flask import g, jsonify

from flask.ext.httpauth import HTTPBasicAuth

from ..models import User, AnonymousUser

from . import api

from .errors import unauthorized, forbidden

auth = HTTPBasicAuth()

def verify_password(email_or_token, password):

if email_or_token == '':

g.current_user = AnonymousUser()

return True

if password == '':

g.current_user = User.verify_auth_token(email_or_token)

g.token_used = True

return g.current_user is not None

user = User.query.filter_by(email=email_or_token).first()

if not user:

return False

g.current_user = user

g.token_used = False

return user.verify_password(password)

def auth_error():

return unauthorized('Invalid credentials')

def before_request():

if not g.current_user.is_anonymous and \

not g.current_user.confirmed:

return forbidden('未授权的用户.')

def get_token():

if g.current_user.is_anonymous or g.token_used:

return unauthorized('Invalid credentials')

return jsonify({'token': g.current_user.generate_auth_token(

expiration=3600), 'expiration': 3600})

from functools import wraps

from flask import g

from .errors import forbidden

def permission_required(permission):

def decorator(f):

@wraps(f)

def decorated_function(*args, **kwargs):

if not g.current_user.can(permission):

return forbidden('Insufficient permissions')

return f(*args, **kwargs)

return decorated_function

return decorator

from flask import jsonify

from app.exceptions import ValidationError

from . import api

def bad_request(message):

response = jsonify({'error': 'bad request', 'message': message})

response.status_code = 400

return response

def unauthorized(message):

response = jsonify({'error': 'unauthorized', 'message': message})

response.status_code = 401

return response

def forbidden(message):

response = jsonify({'error': 'forbidden', 'message': message})

response.status_code = 403

return response

def validation_error(e):

return bad_request(e.args[0])

from flask import jsonify, request, current_app, url_for

from . import api

from ..models import User

def get_user(id):

user = User.query.get_or_404(id)

return jsonify(user.to_json())

from flask import Blueprint

auth = Blueprint('auth', __name__)

from . import views

from flask.ext.wtf import Form

from wtforms import StringField, PasswordField, BooleanField, SubmitField

from wtforms.validators import Required, Length, Email, Regexp, EqualTo

from wtforms import ValidationError

from ..models import User

import sys

reload(sys)

sys.setdefaultencoding('utf-8')

class LoginForm(Form):

email = StringField('邮箱', validators=[Required(), Length(1, 64),

Email()])

password = PasswordField('密码', validators=[Required()])

remember_me = BooleanField('下次自动登录')

submit = SubmitField('登录')

class RegistrationForm(Form):

email = StringField('邮箱', validators=[Required(), Length(1, 64),

Email()])

username = StringField('用户名', validators=[

Required(), Length(1, 64), Regexp('^[A-Za-z][A-Za-z0-9_.]*$', 0,

'用户名只能有字母, '

'数字, 点 or 下划线')])

password = PasswordField('密码', validators=[

Required(), EqualTo('password2', message='密码不一致.')])

password2 = PasswordField('密码确认', validators=[Required()])

submit = SubmitField('注册')

def validate_email(self, field):

if User.query.filter_by(email=field.data).first():

raise ValidationError('邮箱已注册.')

def validate_username(self, field):

if User.query.filter_by(username=field.data).first():

raise ValidationError('用户名已注册.')

class ChangePasswordForm(Form):

old_password = PasswordField('旧密码', validators=[Required()])

password = PasswordField('新密码', validators=[

Required(), EqualTo('password2', message='Passwords must match')])

password2 = PasswordField('确认密码', validators=[Required()])

submit = SubmitField('提交')

from flask import render_template, redirect, request, url_for, flash, current_app

from flask.ext.login import login_user, logout_user, login_required, \

current_user

from . import auth

from .. import db

from ..models import User

from ..email import send_email

from .forms import LoginForm, RegistrationForm, ChangePasswordForm

from ..decorators import admin_required

def before_request():

if current_user.is_authenticated:

current_user.ping()

if not current_user.confirmed \

and request.endpoint[:5] != 'auth.' \

and request.endpoint != 'static':

return redirect(url_for('auth.unconfirmed'))

def unconfirmed():

if current_user.is_anonymous or current_user.confirmed:

return redirect(url_for('main.index'))

return render_template('auth/unconfirmed.html')

def login():

form = LoginForm()

if form.validate_on_submit():

user = User.query.filter_by(email=form.email.data).first()

if user is not None and user.verify_password(form.password.data):

login_user(user, form.remember_me.data)

return redirect(request.args.get('next') or url_for('main.index'))

flash('用户名或密码不正确.')

return render_template('auth/login.html', form=form)

def logout():

logout_user()

flash('您已退出登录.')

return redirect(url_for('main.index'))

def register():

form = RegistrationForm()

if form.validate_on_submit():

user = User(email=form.email.data,

username=form.username.data,

password=form.password.data)

db.session.add(user)

db.session.commit()

# token = user.generate_confirmation_token()

# send_email(user.email, 'Confirm Your Account',

# 'auth/email/confirm', user=user, token=token)

flash('注册申请已提交, 等待管理员确认.')

return redirect(url_for('auth.login'))

return render_template('auth/register.html', form=form)

def list_users():

page = request.args.get('page', 1, type=int)

query = User.query

pagination = query.paginate(

page, per_page=current_app.config['FLASKY_USERS_PER_PAGE'],

error_out=False)

users = pagination.items

return render_template('auth/confirm.html', pagination=pagination, users=users)

def confirm(user_id):

user = User.query.filter_by(id=user_id).first()

user.confirm()

flash('授权成功.')

return redirect(url_for('auth.list_users'))

def unconfirm(user_id):

user = User.query.filter_by(id=user_id).first()

user.unconfirm()

flash('取消授权成功.')

return redirect(url_for('auth.list_users'))

def change_password():

form = ChangePasswordForm()

if form.validate_on_submit():

if current_user.verify_password(form.old_password.data):

current_user.password = form.password.data

db.session.add(current_user)

flash('Your password has been updated.')

return redirect(url_for('main.index'))

else:

flash('Invalid password.')

return render_template("auth/change_password.html", form=form)

from functools import wraps

from flask import abort

from flask.ext.login import current_user

from .models import Permission

def permission_required(permission):

def decorator(f):

@wraps(f)

def decorated_function(*args, **kwargs):

if not current_user.can(permission):

abort(403)

return f(*args, **kwargs)

return decorated_function

return decorator

def admin_required(f):

return permission_required(Permission.ADMINISTER)(f)

from threading import Thread

from flask import current_app, render_template

from flask.ext.mail import Message

from . import mail

def send_async_email(app, msg):

with app.app_context():

mail.send(msg)

def send_email(to, subject, template, **kwargs):

app = current_app._get_current_object()

msg = Message(app.config['FLASKY_MAIL_SUBJECT_PREFIX'] + ' ' + subject,

sender=app.config['FLASKY_MAIL_SENDER'], recipients=[to])

msg.body = render_template(template + '.txt', **kwargs)

msg.html = render_template(template + '.html', **kwargs)

thr = Thread(target=send_async_email, args=[app, msg])

thr.start()

return thr

class ValidationError(ValueError):

pass

from flask import Blueprint

main = Blueprint('main', __name__)

from . import views, errors

from ..models import Permission

def inject_permissions():

return dict(Permission=Permission)