Filling the Flask – Management Commands – Database Access



Filling the Flask – Management Commands – Database Access

0 0


filling-the-flask

PyOhio 2015 Presentation

On Github jasonamyers / filling-the-flask

Filling the Flask

Created by Jason A Myers / @jasonamyers

I know what your thinking... "I'm awesome and I got full hearts.
I mean I beat ocarina of time without the ocarina"

Our Empty Flask

I have two files setup: flaskfilled/__init__.py and config.py

__init__.py

from flask import Flask

from config import config


def create_app(config_name):
    app = Flask(__name__)
    app.config.from_object(config[config_name])
    return app

config.py

import os


basedir = os.path.abspath(os.path.dirname(__file__))


class Config:
    SECRET_KEY = 'development key'
    ADMINS = frozenset(['jason@jasonamyers.com', ])


class DevelopmentConfig(Config):
    DEBUG = True


config = {
    'development': DevelopmentConfig,
    'default': DevelopmentConfig
}

Management Commands

flask-script

Commands for running a development server, a customised Python shell, etc

pip install flask-script

manage.py

#! /usr/bin/env python
import os

from flask.ext.script import Manager

from flaskfilled import create_app

app = create_app(os.getenv('FLASK_CONFIG') or 'default')
manager = Manager(app)

if __name__ == '__main__':
    manager.run()

Shell with Context

from flask.ext.script import Shell

def make_shell_context():
    return dict(app=app)

manager.add_command('shell', Shell(make_context=make_shell_context))
$ python manage.py
usage: manage.py [-?] {runserver,shell} ...

positional arguments:
  {runserver,shell}
    runserver        Runs the Flask development server i.e. app.run()
    shell            Runs a Python shell inside Flask application context.

optional arguments:
  -?, --help         show this help message and exit
$ python manage.py shell

In [1]: app.config['DEBUG']
Out[1]: True

Database Access

flask-SQLAlchemy

  • Single wrapper for most of SQLAlchemy
  • Preconfigured scope session
  • Sessions are tied to the page lifecycle
pip install flask-sqlalchemy

__init__.py

from flask.ext.sqlalchemy import SQLAlchemy

db = SQLAlchemy()

def create_app(config_name):
    app = Flask(__name__)
    app.config.from_object(config[config_name])

    db.init_app(app)
    return app

config.py

class DevelopmentConfig(Config):
    DEBUG = True
    SQLALCHEMY_DATABASE_URI = "sqlite:////tmp/dev.db"

flaskfilled/models.py

from flaskfilled import db


class Cookie(db.Model):
    __tablename__ = 'cookies'

    cookie_id = db.Column(db.Integer(), primary_key=True)
    cookie_name = db.Column(db.String(50), index=True)
    cookie_recipe_url = db.Column(db.String(255))
    quantity = db.Column(db.Integer())

manage.py

from flask.ext.script import Command

from flaskfilled import db
from flaskfilled.models import Cookies

def make_shell_context():
    return dict(app=app, db=db)


class DevDbInit(Command):
    '''Creates database tables from sqlalchemy models'''

    def __init__(self, db):
        self.db = db

    def run(self):
        self.db.create_all()
$ python manage.py db_init
$ python manage.py shell

In [1]: db.metadata.tables
Out[1]: immutabledict({'cookies': Table('cookies', 'stuff')})

In [2]: from flaskfilled.models import Cookie
c = Cookie(cookie_name="Chocolate Chip",
           cookie_recipe_url="http://zenofthecookie.com/chocolatechip.html",
           quantity=2)
db.session.add(c)
db.session.commit()

Migrations

flask-migrate

Ties Alembic into flask-script!

pip install flask-migrate

manage.py

from flask.ext.migrate import Migrate, MigrateCommand

migrate = Migrate(app, db)

manager.add_command('db', MigrateCommand)

Initializing Alembic

$ python manage.py db init

Generating a migration

$ python manage.py db migrate -m "initial migration"
INFO  [alembic.migration] Context impl SQLiteImpl.
INFO  [alembic.migration] Will assume non-transactional DDL.
  Generating flask-filled/migrations/versions/586131216f6_initial_migration.py ... done

Running migrations

$ python manage.py db upgrade
INFO  [alembic.migration] Context impl SQLiteImpl.
INFO  [alembic.migration] Will assume non-transactional DDL.
INFO  [alembic.migration] Running upgrade  -> 586131216f6, initial migration

User Authentication

Flask-Login

  • Simplifies logging users in and out
  • Secures view functions with decorators
  • Protects session cookies

pip install flask-login

flaskfilled/__init__.py

from flask.ext.login import LoginManager

login_manager = LoginManager()

def create_app(config_name):
    app = Flask(__name__)
    app.config.from_object(config[config_name])

    db.init_app(app)
    login_manager.setup_app(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')

    return app

models.py

from werkzeug.security import generate_password_hash, check_password_hash

from flaskfilled import login_manager

class User(db.Model, UserMixin):
    __tablename__ = 'users'

    id = db.Column(db.Integer(), primary_key=True)
    username = db.Column(db.String, primary_key=True)
    password = db.Column(db.String)
    authenticated = db.Column(db.Boolean, default=False)

User Model required methods

Provided by UserMixin

def is_active(self):
    return True

def get_id(self):
    return self.id

def is_authenticated(self):
    return self.authenticated

def is_anonymous(self):
    return False

User model password handling

@property
  def password(self):
      raise AttributeError('password is not a readable attribute')

  @password.setter
  def password(self, password):
      self.password_hash = generate_password_hash(password)

  def verify_password(self, password):
      return check_password_hash(self.password_hash, password)

Setting up the Auth Blueprint

auth/__init__.py

from flask import Blueprint

auth = Blueprint('auth', __name__)

from . import views

auth/views.py

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

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

from . import auth
from flaskfilled.models import User

login

@auth.route('/login', methods=['GET', 'POST'])
def login():
    if request.method == 'POST':
        username = request.form.get('username', '')
        password = request.form.get('password', '')
        user = User.query.filter_by(username=username).first()
        if user is not None and user.verify_password(password):
            login_user(user)
            next = request.args.get('next')
            return redirect(next or url_for('main.index'))
        else:
            flash('Wrong username or password.')
    return render_template('auth/login.html')

logout

@auth.route('/logout')
@login_required
def logout():
    logout_user()
    flash('You have been logged out.')
    return redirect(url_for('main.index'))

login template

{% extends "base.html" %}

{% block title %}Login{% endblock %}

{% block page_content %}
<div class="page-header">
    <h1>Login</h1>
</div>
<div class="col-md-4">
    <form action="">
    Username: <input type="text" name="username"><br>
    Password: <input type="password" name="password"><br>
    <input type="submit">
    </form>
    <br>
    <p>Forgot your password? <a href="{{ url_for('auth.password_reset_request') }}">Click here to reset it</a>.</p>
    <p>New user? <a href="{{ url_for('auth.register') }}">Click here to register</a>.</p>
</div>
{% endblock %}

main/views.py

from flask import render_template

from . import main


@main.route('/', methods=['GET'])
def index():
    return render_template('main/index.html')

index template

{% extends "base.html" %}

{% block title %}The Index{% endblock %}

{% block page_content %}
{% if not current_user.is_authenticated() %}
  <p><a href="{{ url_for('auth.login') }}">Click here to login</a>.</p>
{% else %}
  <p><a href="{{ url_for('auth.logout') }}">Click here to logout</a>.</p>
{% endif %}
{% endblock %}

Create Users Migration and Apply it

$ python manage.py db migrate -m "User"
  Generating /Users/jasonamyers/dev/flask-filled/migrations/versions/8d9327f04f_user.py ... done

$ python manage.py db upgrade
INFO  [alembic.migration] Running upgrade 586131216f6 -> 8d9327f04f, User

Run Server

$ python manage.py runserver

Forms...

Flask-WTF

  • Validation
  • CSRF protection
  • File Uploads
pip install flask-wtf

auth/forms.py

from flask.ext.wtf import Form
from wtforms import StringField, PasswordField, SubmitField
from wtforms.validators import Required, Length


class LoginForm(Form):
      username = StringField('username', validators=[Required(),
                                                     Length(1, 64)])
      password = PasswordField('Password', validators=[Required()])
      submit = SubmitField('Log In')

auth/views.py

@auth.route('/login', methods=['GET', 'POST'])
def login():
    form = LoginForm()
    if form.validate_on_submit():
        user = User.query.filter_by(username=form.username.data).first()
        if user is not None and user.verify_password(form.password.data):
            login_user(user)
            next = request.args.get('next')
            return redirect(next or url_for('main.index'))
        else:
            flash('Wrong username or password.')
    return render_template('auth/login.html', form=form)

templates/auth/login.html

{% block page_content %}
<div class="col-md-4">
    <form action="" method="POST">
      {{ form.csrf_token }}
      {% if form.csrf_token.errors %}
        <div class="warning">You have submitted an invalid CSRF token</div>
      {% endif %}
      {{form.username.label }}: {{ form.username }}
      {% if form.username.errors %}
        {% for error in form.username.errors %}
          {{ error }}
        {% endfor %}
        {% endif %}<br>
      {{form.password.label }}: {{ form.password }}
      {% if form.password.errors %}
        {% for error in form.password.errors %}
          {{ error }}
        {% endfor %}
        {% endif %}<br>
      {{ form.submit }}
    </form>
    <br>
</div>

Authorization

flask-principal

pip install flask-principal

__init__.py

from flask.ext.principal import Principal

principal = Principal()


def create_app(config_name):
    principal.init_app(app)

models.py

roles_users = db.Table('roles_users',
                       db.Column('user_id', db.Integer(),
                                 db.ForeignKey('users.user_id')),
                       db.Column('role_id', db.Integer(),
                                 db.ForeignKey('roles.id')))


class Role(db.Model):
    __tablename__ = 'roles'

    id = db.Column(db.Integer(), primary_key=True)
    name = db.Column(db.String(80), unique=True)
    description = db.Column(db.String(255))

models.py - User Class

class User(db.Model, UserMixin):
    roles = db.relationship('Role', secondary=roles_users,
                            primaryjoin=user_id == roles_users.c.user_id,
                            backref='users')

models.py - Identity loader

@identity_loaded.connect
def on_identity_loaded(sender, identity):
    # Set the identity user object
    identity.user = current_user

    # Add the UserNeed to the identity
    if hasattr(current_user, 'id'):
        identity.provides.add(UserNeed(current_user.id))

    # Assuming the User model has a list of roles, update the
    # identity with the roles that the user provides
    if hasattr(current_user, 'roles'):
        for role in current_user.roles:
            identity.provides.add(RoleNeed(role.name))

auth/views.py

from flask import current_app

from flask.ext.principal import identity_changed, Identity


@auth.route('/login', methods=['GET', 'POST'])
def login():
    form = LoginForm()
    if form.validate_on_submit():
        user = User.query.filter_by(username=form.username.data).first()
        if user is not None and user.verify_password(form.password.data):
            login_user(user)
            identity_changed.send(current_app._get_current_object(),
                                  identity=Identity(user.user_id))
            next = request.args.get('next')
            return redirect(next or url_for('main.index'))
        else:
            flash('Wrong username or password.')
    return render_template('auth/login.html', form=form)

auth/__init__.py

from flask.ext.principal import Permission, RoleNeed

admin_permission = Permission(RoleNeed('admin'))

main/views.py

from flaskfilled.auth import admin_permission

@main.route('/settings', methods=['GET'])
@admin_permission.require()
def settings():
    return render_template('main/settings.html')

Sending Mail

flask-mail

  • Works with Flask config
  • Simplies Message Construction

__init__.py

from flask.ext.mail import Mail
mail = Mail()


def create_app(config_name):
    mail.init_app(app)

__init__.py

from flask.ext.mail import Mail
mail = Mail()


def create_app(config_name):
    mail.init_app(app)

main/views.py

from flask_mail import Message


@main.route('/mailme', methods=['GET'])
def mail():
    msg = Message('COOKIES!',
                  sender='from@example.com',
                  recipients=['to@example.com'])
    msg.body = 'There all mine!'
    msg.html = '<b>There all mine!</b>'
    mail.send(msg)

What other things are out there?

  • flask-security
  • flask-moment

https://github.com/humiaozuzu/awesome-flask

Questions

Jason Myers / @jasonamyers / Essential SQLAlchemy

Filling the Flask Created by Jason A Myers / @jasonamyers