1. 程式人生 > >記錄一次完整的flask小型應用開發(1)

記錄一次完整的flask小型應用開發(1)

首先建立虛擬環境:

virtualenv venv
source venv/bin/activate

建立專案結構:

|-flasky
	|-app/ #四個頂級資料夾之一,flask程式儲存在這個裡面
		|-templates/
		|-static/
		|-main/
			|-__init__.py
			|-errors.py
			|-forms.py
			|-views.py #這是程式的路由
		|__init__.py
		|-email.py
		|-models.py #資料庫模型
	|-migrations/ #四個頂級資料夾之一,包含資料庫遷移指令碼
	|-tests/ #四個頂級資料夾之一,單元測試編寫在這個包中
		|-__init__.py
		|-test*.py
	|-venv/ #四個頂級資料夾之一,包含python flask的虛擬環境
	|-requirements.txt #列出了所有依賴包,其他電腦中能生成相同的虛擬環境
	|-config.py #儲存了配置資訊
	|-manage.py #用於啟動程式和其他的程式任務,這個指令碼先建立程式,然後定義了配置

在環境里根據requirement.txt安裝所有的依賴

編寫配置檔案:

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

class Config:   #這是基類,包含了通用設定
    MAIL_USERNAME = os.environ.get('MAIL_USERNAME') #這裡不像單一腳本里面使用app.config['MAIL_USERNAME']=...這樣的字典結構
    ...
    @staticmethod
    def init_app(app):  #這個的引數是程式例項,在這個方法中,可以執行對當前環境的配置初始化。
        pass
class DevelopmentConfig(Config): #三個子類之一
    DEBUG = True
    SQLALCHEMY_DATABASE_URI = os.environ.get('DEV_DATABASE_URL') or \
        'sqlite:///' + os.path.join(basedir, 'data-dev.sqlite')
class TestingConfig(Config):  #三個子類之一
    TESTING = True
    SQLALCHEMY_DATABASE_URI = os.environ.get('TEST_DATABASE_URL') or \
        'sqlite://'
class ProductionConfig(Config):  #三個子類之一
    SQLALCHEMY_DATABASE_URI = os.environ.get('DATABASE_URL') or \
        'sqlite:///' + os.path.join(basedir, 'data.sqlite')

config = {        #為四種模式分別指定一個配置類
    'development': DevelopmentConfig,
    'testing': TestingConfig,
    'production': ProductionConfig,
    'default': DevelopmentConfig
}

資料庫模型和電子郵件支援函式也在app這個資料夾下,models.py:

from . import db


class Role(db.Model):
    __tablename__ = 'roles'
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(64), unique=True)
    users = db.relationship('User', backref='role', lazy='dynamic')

    def __repr__(self):
        return '<Role %r>' % self.name


class User(db.Model):
    __tablename__ = 'users'
    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(64), unique=True, index=True)
    role_id = db.Column(db.Integer, db.ForeignKey('roles.id'))

    def __repr__(self):
        return '<User %r>' % self.username

email.py:

from threading import Thread
from flask import current_app, render_template
from flask_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

接下來,使用程式工廠函式:
延遲建立程式例項,把建立過程移到可顯式呼叫的工廠函式當中:

# coding:utf-8
from flask import Flask
from flask_bootstrap import Bootstrap
from flask_mail import Mail
from flask_moment import Moment
from flask_sqlalchemy import SQLAlchemy
from config import config

bootstrap = Bootstrap()
mail = Mail()
moment = Moment()
db = SQLAlchemy()

def create_app(config_name): # 程式的工廠函式
    app = Flask(__name__)
    app.config.from_object(config[config_name])  # config提供from_object()這個方法匯入配置

    config[config_name].init_app(app)  # 初始化拓展
    bootstrap.init_app(app)
    mail.init_app(app)
    moment.init_app(app)
    db.init_app(app)

    return app

在藍本中實現程式的功能:
在單指令碼的程式中,程式例項存在於全域性作用域中,路由可以直接使用app.route修飾器來定義。但是現在程式在執行的時候才建立,所以只有使用上面的create_app()之後才可以使用裝飾器,但是這個時候定義路由就太晚了。

所以我們使用藍本,藍本和程式類似也可以定義路由,但是藍本定義的路由處於休眠狀態,直到藍本註冊到程式上面。為了獲得最大的靈活性,我在程式包中建立一個子包main用於儲存藍本:app/main/init.py:

from flask import Blueprint
# 例項化一個藍本物件
main = Blueprint('main', __name__)  # 兩個引數:藍本的名字和藍本所在的包或者模組

from . import views, errors  # 末尾匯入

藍本定義的路由: app/main/views.py:

#coding:utf-8
from flask import render_template, session, redirect, url_for, current_app
from .. import db
from ..models import User
from ..email import send_email
from . import main
from .forms import NameForm


@main.route('/', methods=['GET', 'POST'])  # 路由裝飾器由藍本提供
def index():
    form = NameForm()
    if form.validate_on_submit():
        user = User.query.filter_by(username=form.name.data).first()
        if user is None:
            user = User(username=form.name.data)
            db.session.add(user)
            db.session.commit()
            session['known'] = False
            if current_app.config['FLASKY_ADMIN']:
                send_email(current_app.config['FLASKY_ADMIN'], 'New User',
                           'mail/new_user', user=user)
        else:
            session['known'] = True
        session['name'] = form.name.data
        return redirect(url_for('.index'))
    # 在程式的路由中,url_for()預設使用檢視函式的名字,比如index()檢視函式的URL可以用url_for('index)獲取
    # 但是藍本中flask會在全部斷點加上一個名稱空間即藍本的名字,所以應該使用main.index
    # .index是一種簡寫端點,是當前請求所在的藍本
    return render_template('index.html',
                           form=form, name=session.get('name'),
                           known=session.get('known', False))

表單物件也移到藍本中: app/main/forms.py:

from flask_wtf import FlaskForm
from wtforms import StringField, SubmitField
from wtforms.validators import DataRequired


class NameForm(FlaskForm):
    name = StringField('What is your name?', validators=[DataRequired()])
    submit = SubmitField('Submit')

藍本的錯誤處理程式: app/main/errors.py:

from flask import render_template
from . import main


@main.app_errorhandler(404)  # 註冊程式全域性的錯誤處理程式
def page_not_found(e):
    return render_template('404.html'), 404


@main.app_errorhandler(500)
def internal_server_error(e):
    return render_template('500.html'), 500

!!!最後一定要在工廠函式裡面將藍本註冊到程式上:

from .main import main as main_blueprint
app.register_blueprint(main_blueprint)

使用manage檔案啟動這個程式:

#coding: utf-8
#!/usr/bin/env python
import os
from app import create_app, db
from app.models import User, Role
from flask_script import Manager, Shell
from flask_migrate import Migrate, MigrateCommand

app = create_app(os.getenv('FLASK_CONFIG') or 'default')
manager = Manager(app)
migrate = Migrate(app, db)


@app.shell_context_processor
def make_shell_context():
    return dict(db=db, User=User, Role=Role)
manager.add_command("shell", Shell(make_context=make_shell_context))
# 整合python shell, 這個函式註冊了程式,資料庫例項以及模型,因此這些物件能直接匯入shell
manager.add_command('db', MigrateCommand)
# 為了匯出資料庫遷移命令

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

這裡我們直接執行的話,就使用default這個config_name

這裡我們使用Flask-Migrate來實現資料庫的遷移:

  1. 維護資料庫遷移之前,我們使用init子命令來建立遷移倉庫:
    python manage.py db init
    這個命令會建立migrations資料夾,所有的遷移指令碼都放在裡面。

  2. 使用migrate子命令來自動的建立遷移指令碼:
    python manage.py db migrate -m "initial migration"

  3. 使用upgrade來更新資料庫:
    python manage.py db upgrade
    對於第一次遷移來說,作用相當於db.create_all()
    後續就是把改動應用到資料庫

  4. 第一次之後的所有遷移重複使用3,4步:
    python manage.py db migrate
    python manage.py db upgrade