1. 程式人生 > >Flask1.0.2系列(二十) Flask相關的模式

Flask1.0.2系列(二十) Flask相關的模式

英文原文地址:http://flask.pocoo.org/docs/1.0/patterns/

若有翻譯錯誤或者不盡人意之處,請指出,謝謝~


        某些東西很常見,你可能在大多數Web應用程式中都能發現它們。舉個栗子,相當多的應用程式使用了關係型資料庫,以及使用者驗證。在這種情形下,它們很可能在開始請求的時候開啟一個數據庫連線,並且獲取當前登入使用者的資訊。在請求的最後,資料庫連線將被再次關閉。

        在Flask Snippet Archives文章中有更多使用者貢獻的程式碼片段和模式。


1. 大型應用程式

        對於大型應用程式來說,最好的設計思路是使用包而不是模組。想象一個小的應用程式時這樣的:

/yourapplication
    yourapplication.py
    /static
        style.css
    /templates
        layout.html
        index.html
        login.html
        ...

1.1 簡單的包

        為了將其轉換到一個大的專案中,僅需要在大的專案中建立一個新的資料夾yourapplication,並將所有東西拷貝到這個資料夾下。然後將yourapplication.py更名為__init__.py。(確保首先刪除所有.pyc檔案,否則可能會有異常出現)

        你隨後建立的目錄結構應該像下面這樣:

/yourapplication
    /yourapplication
        __init__.py
        /static
            style.css
        /templates
            layout.html
            index.html
            login.html
            ...

        但是你現在該如何執行你的應用程式呢?不要天真地認為

python yourapplication/__init__.py

就能工作。Python不希望包中的模組成為啟動檔案。但是這不是什麼大問題,僅需要新增一個叫做setup.py的新檔案,這個檔案放在yourapplication資料夾所在的資料夾下,並且這個檔案的內容如下:

from setuptools import setup


setup(
    name='yourapplication',
    packages=['yourapplication'],
    include_package_data=True,
    install_requires=[
        'flask',
    ],
)

        為了執行這個應用程式,你需要匯出一個環境變數,用來告訴Flask在哪裡可以找到應用程式例項:

export FLASK_APP=yourapplication

        為了安裝和執行應用程式,你需要傳送下面的命令:

pip install -e .
flask run

        我們能從這裡知道什麼呢?現在,我們可以重構應用程式,將其分解到多個模組中。你只需要記住下面的快速檢查清單:

        1. Flask應用程式物件的建立是在__init__.py檔案中。這種方式能讓每個模組安全地匯入它,並且__name__變數將解析為正確的包。

        2. 所有檢視方法(即在頂部標記了route()裝飾器的方法)必須在__init__.py檔案中被匯入。不僅是物件本身,而且模組也在裡面。在應用程式物件生成之後匯入檢視模組。

        這裡有一個__init__.py檔案的示例:

from flask import Flask


app = Flask(__name__)


import yourapplication.views

        並且view.py程式碼如下:

from yourapplication import app


@app.route('/')
def index():
    return 'Hello World!'

        你隨後建立的目錄結構應該像下面這樣:

/yourapplication
    setup.py
    /yourapplication
        __init__.py
        views.py
        /static
            style.css
        /templates
            layout.html
            index.html
            login.html
            ...

        迴圈匯入:

        所有Python開發者都討厭這個,實際上我們在這裡已經這樣做過了:迴圈匯入(表示兩個模組互相依賴於對方。在這裡view.py依賴於__init__.py)。請注意,通常這是一個不好的想法,但是這裡實際上是正確的。因為我們實際上在__init__.py中並沒有使用views,並且我們確保了在__init__.py檔案的底部我們才匯入了這個模組。

        這種方法依然有一些問題,但是如果你想要在這裡使用裝飾器,那就沒有其他辦法了。後面的部分,我們會講解有關如何處理這種情況的靈感。

1.2 使用藍圖

        如果你有一個很大的應用程式,這裡推薦將這個程式分解成很多小組,而每個小組的實現都依賴於使用藍圖。更多關於藍圖的介紹,請查閱前面的章節,使用藍圖將應用程式模組化。


2. 應用程式工廠

        如果你已經在你的應用程式中使用了包和藍圖,有一些不錯的方法可以進一步改善這種體驗。一種常用的模式是,在藍圖被匯入的時候建立應用程式物件。但是如果你將建立這個物件的程式碼移動到一個方法內的話,你可以在這個app物件之後再建立其他你所需要的例項。

        那麼,為什麼你需要這樣做呢?

        1. 為了測試。你可以通過不同的配置來建立這個應用程式例項,這樣可以在任務你需要的場景下進行測試。

        2. 多個例項。想象一下你需要執行不同版本的應用程式。當然,你可以在建立你的web服務時,根據不同的配置來建立多個例項。但是,如果你使用了工廠,你可以在同一個應用程式程序中運行同一個應用程式的多個例項,這是很方便的。

        那麼我們該如何實現這種方式呢?

2.1 基礎工廠

        基礎工廠的思想是在一個方法中建立應用程式。就像這樣:

def create_app(config_filename):
    app = Flask(__name__)
    app.config.from_pyfile(config_filename)

    from yourapplication.model import db
    db.init_app(app)

    from yourapplication.views.admin import admin
    from yourapplication.views.frontend import frontend
    app.register_blueprint(admin)
    app.register_blueprint(frontend)

    return app

        這種方法的缺點是,在匯入時,你無法在藍圖中使用應用程式物件。然而,你可以通過一個請求使用它。怎樣才能訪問應用程式的配置項呢?使用current_app就可以:

from flask import current_app, Blueprint, render_template


admin = Blueprint('admin', __name__, url_prefix='/admin')


@admin.route('/')
def index():
    return render_template(current_app.config['INDEX_TEMPLATE'])

        在這裡,我們在配置中尋找指定名稱的模板。

2.2 工廠 & 擴充套件

        最好是建立你的擴充套件和應用程式工廠,以便擴充套件物件最初不會被繫結到應用程式。

        舉個栗子,使用Flask-SQLAlchemy,你不應該像下面這樣使用:

def create_app(config_filename):
    app = Flask(__name__)
    app.config.from_pyfile(config_filename)

    db = SQLAlchemy(app)

        你應該在model.py(或者等價的模組)中建立物件:

db = SQLAlchemy()

        然後在你的application.py(或者等價的模組)中初始化:

def create_app(config_filename):
    app = Flask(__name__)
    app.config.from_pyfile(config_filename)

    from yourapplication.model import db
    db.init_app(app)

        使用這種設計模式,沒有特定的應用程式狀態被儲存到擴充套件物件上,因此一個擴充套件物件可以被用到多個應用程式物件上。更多資訊請參考Flask Extension Development

2.3 使用應用程式

        為了執行一個應用程式,你可以使用flask命令:

export FLASK_APP=myapp
flask run

        Flask會自動在myapp中檢測工廠(create_app或者make_app)方法。你也可以傳遞引數到這個工廠方法中:

export FLASK_APP="myapp:create_app('dev')"
flask run

        然後在myapp中的create_app工廠會被呼叫,並且帶有字串引數‘dev’。

2.4 工廠改進

        上面描述的工廠方法並不是很智慧,但是你可以改進它。下面的更改很容易實現:

        1. 可以在單元測試中傳遞配置值,這樣就不必在檔案系統上建立配置檔案了。

        2. 當應用程式建立時,從藍圖中呼叫一個方法,這樣你就有一個可以修改應用程式屬性的地方了(就像在before/after請求處理函式的鉤子函式一樣)。

        3. 如果有需要,可以在應用程式被建立時,新增到WSGI中介軟體。


3. 應用程式的排程

        應用程式的排程,即在WSGI層級上組合多個Flask應用程式的程序。你不僅僅可以組合Flask應用程式,還可以組合任意的WSGI應用程式。如果你願意的話,這裡還能允許你在同一個直譯器中,執行一個Django應用程式和一個Flask應用程式。這是否有用取決於應用程式內部是如何工作的。

        應用程式的排程與模組方法的根據區別在於,在這種情況下,你執行的是相同或不同的Flask應用程式,它們彼此間是完全隔離的。它們執行在不同的配置,並且在WSGI層進行排程。

3.1 如何使用此文件

        下面所有技術和示例最終會得到一個應用程式物件,這個物件能被執行在任何WSGI伺服器上。在生產環境下,請參考Deployment Options。在開發環境下,Werkzeug提供了一個內建的開發伺服器,可以使用werkzeug.serving.run_simple()來啟用它:

from werkzeug.serving import run_simple


run_simple('localhost', 5000, application, use_reloader=True)

        注意run_simple不能用於生產環境。釋出應用時可以使用full-blown WSGI server

        為了使用互動式偵錯程式,必須在應用層序和這個簡單伺服器上開啟除錯。下面是一個帶有除錯功能的“hello world”的示例:

from flask import Flask
from werkzeug.serving import run_simple


app = Flask(__name__)
app.debug = True


@app.route('/')
def hello_world():
    return 'Hello World!'


if __name__ == '__main__':
    run_simple('localhost', 5000, app,
               use_reloader=True, use_debugger=True, use_evalex=True)

3.2 合併應用程式

        如果你擁有一些完全獨立的應用程式,並且你希望它們能在同一個的Python直譯器中彼此共同工作,你可以利用werkzeug.wsgi.DispatcherMiddleware來實現。在這裡,每個Flask應用物件都是一個有效的WSGI應用物件,排程器中介軟體會將它們合併到一個規模更大的應用中,並通過字首來實現排程。

        舉個栗子,你可以使你的主程式執行在‘/’上,使你的後端介面執行在‘/backend’上:

from werkzeug.wsgi import DispatcherMiddleware
from frontend_app import application as frontend
from backend_app import application as backend


application = DispatcherMiddleware(frontend, {
    '/backend':     backend
})

3.3 通過子域來排程

        有時候你可能希望通過配置的方式來使用多個同樣應用程式的例項。假定在一個方法內建立了應用程式,並且你可以呼叫這個方法來進行初始化操作,這是非常容易實現的。為了開發你的應用程式來支援在方法中建立新的例項,請檢視應用程式工廠模式一節。

        一個非常常見的例子是,為每個子域名建立對應的應用程式物件。舉個栗子,你可以通過配置你的web服務的方式,將你所有子域名的請求分發到你的應用程式,並且你可以在稍後使用這子域名的資訊來建立使用者指定的例項。一旦你的服務是為了監聽所有子域名而建立的,你可以使用一個非常簡單的WSGI應用程式來實現動態應用程式的建立。

        實現此功能最完美的抽象層是WSGI層。你可以編寫你自己的WSGI應用程式,它觀察接收到的請求,並將請求委託到你的Flask應用程式。如果這個應用程式不存在,它會自動地建立並記住這個建立的應用程式:

from threading import Lock


class SubdomainDispatcher(object):

    def __init__(self, domain, create_app):
        self.domain = domain
        self.create_app = create_app
        self.lock = Lock()
        self.instances = {}

    def get_application(self, host):
        host = host.split(':')[0]
        assert host.endswith(self.domain), 'Configuration error'
        subdomain = host[:-len(self.domain)].rstrip('.')
        with self.lock:
            app = self.instances.get(subdomain)
            if app is None:
                app = self.create_app(subdomain)
                self.instances[subdomain] = app
            return app

    def __call__(self, environ, start_response):
        app = self.get_application(environ['HTTP_HOST'])
        return app(environ, start_response)

        可以這樣實現一個排程方:

from myapplication import create_app, get_user_for_subdomain
from werkzeug.exceptions import NotFound


def make_app(subdomain):
    user = get_user_for_subdomain(subdomain)
    if user is None:
        # if there is no user for that subdomain we still have
        # to return a WSGI application that handles that request.
        # We can then just return the NotFound() exception as
        # application which will render a default 404 page.
        # You might also redirect the user to the main page then
        return NotFound()

    # otherwise create the application for the specific user
    return create_app(user)


application = SubdomainDispatcher('example.com', make_app)

3.4 使用路徑來排程

        通過URL路徑分發請求,跟之前的方法很相似。不是通過檢查用來確定子域名的HOST頭資訊,而是簡單檢查請求路徑中到第一個斜槓之前的部分:

from threading import Lock
from werkzeug.wsgi import pop_path_info, peek_path_info


class PathDispatcher(object):

    def __init__(self, default_app, create_app):
        self.default_app = default_app
        self.create_app = create_app
        self.lock = Lock()
        self.instances = {}

    def get_application(self, prefix):
        with self.lock:
            app = self.instances.get(prefix)
            if app is None:
                app = self.create_app(prefix)
                if app is not None:
                    self.instances[prefix] = app
            return app

    def __call__(self, environ, start_response):
        app = self.get_application(peek_path_info(environ))
        if app is not None:
            pop_path_info(environ)
        else:
            app = self.default_app
        return app(environ, start_response)

        這種方式與子域名的方式最大的區別在於,如果這裡建立應用程式物件的函式放回了None,那麼請求就被降級回推到另一個應用當中:

from myapplication import create_app, default_app, get_user_for_prefix


def make_app(prefix):
    user = get_user_for_prefix(prefix)
    if user is not None:
        return create_app(user)


application = PathDispatcher(default_app, make_app)


4 實現API異常

        在Flask頂部實現RESTful API是非常常見的。開發人員首先要做的一件事是,認識到對於API來說,內建異常是表達不充分的,而且它們所發出的test/html型別的內容對於API呼叫者來說,並不是很有用的。

        較於使用abort來發送一個非法API呼叫的訊號,更好的方法是實現你自己的異常型別並且為這個異常型別設定對應的錯誤處理程式,而這個錯誤處理程式中就可以將錯誤轉換成使用者期望的格式。

4.1 簡單的異常類

        最基本的思想是,引入一個新的異常,這個異常可以獲取適當的人類可讀的資訊、錯誤的狀態碼以及一些可選的有效負載,從而為錯誤提供更多的內容。

        簡單示例如下:

from flask import jsonify


class InvalidUsage(Exception):
    status_code = 400

    def __init__(self, message, status_code=None, payload=None):
        Exception.__init__(self)
        self.message = message
        if status_code is not None:
            self.status_code = status_code
        self.payload = payload

    def to_dict(self):
        rv = dict(self.payload or ())
        rv['message'] = self.message
        return rv

        現在,試圖可以丟擲這個異常,並伴隨一個錯誤資訊。此外,通過payload引數可以提供一些像字典一樣的額外的有效負載。

4.2 註冊一個錯誤處理程式

        按照上述內容,雖然檢視可以丟擲這個異常,但是在實際執行時只能看到一個內部服務錯誤的資訊。這是因為,這裡並沒有為這個錯誤類註冊對應的處理程式。然而這是很容易新增的:

@app.errorhandler(InvalidUsage)
def handle_invalid_usage(error):
    response = jsonify(error.to_dict())
    response.status_code = error.status_code
    return response

4.3 在檢視中使用

        這裡簡單演示了一個檢視如何使用上述的異常類功能:

@app.route('/foo')
def get_foo():
    raise InvalidUsage('This view is gone', status_code=410)


5. 使用URL處理器

        (新增於版本0.7。)

        Flask0.7引進了URL處理器的概念。你可能有一堆URL,這些URL有一部分相同的部分,有一部分是不能明確提供的。舉個例子,有一堆URL,這些URL中分別有各自的語言程式碼,但是你不想在每個單獨的方法中去處理對應語言程式碼的URL。

        當於藍圖組合使用的時候,URL處理器特別有用。我們將在這裡處理應用程式特定的URL處理器,以及藍圖特定的URL處理器。

5.1 國際化的應用程式URL

        考慮一個應用程式如下:

from flask import Flask, g


app = Flask(__name__)


@app.route('/<lang_code>/')
def index(lang_code):
    g.lang_code = lang_code
    ...


@app.route('/<lang_code>/about')
def about(lang_code):
    g.lang_code = lang_code
    ...

        在每個方法中,你必須將語言程式碼設定到g物件上,這是一個很糟糕的設計,因為需要大量重複的實現程式碼。雖然一個裝飾器可以用來簡化這種方式,但是如果你希望從一個方法到另一個方法生成url,那麼你仍然需要顯示地提供語言程式碼,這是很讓人厭煩的。

        這裡需要使用url_defaults()方法。它們可以自動將值注入到url_for()呼叫中。下面的程式碼檢查了,語言程式碼是否還在URL值的字典中,並且端點是否想要一個叫做“lang_code”的值:

@app.url_defaults
def add_language_code(endpoint, values):
    if 'lang_code' in values or not g.lang_code:
        return
    if app.url_map.is_endpoint_expecting(endpoint, 'lang_code'):
        values['lang_code'] = g.lang_code

        url_map物件的is_endpoint_expecting()方法可以用來斷定為給定的端點提供語言程式碼是否有意義。

        與這個方法相對的是url_value_preprocessor()。它們在請求匹配之後執行,並且可以根據URL值來執行程式碼。其想法是,將資訊從值字典中提取出來放到別的地方:

@app.url_value_preprocessor
def pull_lang_code(endpoint, values):
    g.lang_code = values.pop('lang_code', None)

        這樣,你就不用再在每個函式中將lang_code賦值到g物件中了。你可以通過編寫你自己的裝飾器來進一步完善這個功能,即使用語言程式碼作為URL的字首,但是更漂亮的解決方案是使用一個藍圖。一旦“lang_code”從值字典中彈出,它將不能再被轉發到檢視函式中,從而程式碼簡化如下:

from flask import Flask, g


app = Flask(__name__)


@app.url_defaults
def add_language_code(endpoint, values):
    if 'lang_code' in values or not g.lang_code:
        return
    if app.url_map.is_endpoint_expecting(endpoint, 'lang_code'):
        values['lang_code'] = g.lang_code


@app.url_value_preprocessor
def pull_lang_code(endpoint, values):
    g.lang_code = values.pop('lang_code', None)


@app.route('/<lang_code>/')
def index():
    ...


@app.route('/<lang_code>/about')
def about():
    ...

5.2 國際化的藍圖URL

        因為藍圖能自動地使用一個通常的字串作為所有URL的字首,因此它能很容易自動地對每個方法實現這個。此外,藍圖可以有每個藍圖的URL處理器,它從url_defaults()方法中刪除了大量的邏輯,因為它不需要檢查URL是否真正對“lang_code”引數感興趣:

from flask import Blueprint, g


bp = Blueprint('frontend', __name__, url_prefix='/<lang_code>')


@bp.url_defaults
def add_language_code(endpoint, values):
    values.setdefault('lang_code', g.lang_code)


@bp.url_value_preprocessor
def pull_lang_code(endpoint, values):
    g.lang_code = values.pop('lang_code')


@bp.route('/')
def index():
    ...


@bp.route('/about')
def about():
    ...


6. 使用Setuptools進行部署

        Setuptools是一個擴充套件庫,它通常被用於分發Python庫和擴充套件。它繼承於distutiles,一個使用Python附帶的基本模組安裝系統,也支援各種更復雜的結構,使更大的應用程式更容易分發:

        略,原文地址


7. 使用Fabric進行部署

        略,原文地址


8. Flask中使用SQLite3

        在Flask中,你可以很容易地實現,當需要的時候開啟資料庫連線,以及當上下文完蛋的時候(通常實在請求的最後)關閉資料庫連線。

        這裡有一個簡單的示例展示了在Flask中如何使用SQLite3:

import sqlite3
from flask import g


DATABASE = '/path/to/database.db'


def get_db():
    db = getattr(g, '_database', None)
    if db is None:
        db = g._database = sqlite3.connect(DATABASE)
    return db


@app.teardown_appcontext
def close_connection(exception):
    db = getattr(g, '_database', None)
    if db is not None:
        db.close()

        現在,為了使用這個資料庫,應用程式必須擁有一個活躍的應用程式上下文(如果一個請求正在努力奮鬥那麼應用程式上下文一直都是活躍的),或者應用程式自己建立一個應用程式上下文。這個時候,get_db方法就可以被用來獲取當前資料庫連線。無論何時上下文被銷燬,這個資料庫連線也會被終止。

        注意:如果你使用Flask0.9及其以上版本,你需要使用flask._app_ctx_stack.top,而不是繫結到請求和不是應用程式上下文的flask.g物件。

        示例:

@app.route('/')
def index():
    cur = get_db().cursor()
    ...

        注意:

        請記住,teardown_request和appcontext方法總是會被執行,即使一個before_request處理程式失敗了或者從來不被執行。正因如此,我們必須在我們關閉資料庫之前確保這個資料庫在這裡是存在的。

8.1 在有需求的時候才連線

        這種方法的好處(在第一次使用時才進行連線)是,只有在真正需要的時候才會開啟連線。如果你希望在一個請求上下文之外使用這個程式碼,你可以在一個Python shell中通過手動開啟應用程式上下文來使用它:

with app.app_context():
    # now you can use get_db()

8.2 簡單的查詢

        現在,在每個請求處理程式方法中,你可以訪問get_db()來獲取當前開啟的資料庫連線。為了簡化與SQLite的工作,行工廠(row_factory)方法是很有用的。它將執行每一個從資料庫返回的結果,以便能轉換結果。舉個栗子,為了獲取字典型別而不是元組型別,下面的程式碼可以增加到我們之前建立的get_db方法中:

def make_dicts(cursor, row):
    return dict((cursor.description[idx][0], value)
                for idx, value in enumerate(row))


db.row_factory = make_dicts

        這樣可以使sqlite3模組針對這個資料庫連線返回字典型別,以便更加容易處理。若要更加簡化,我們可以在get_db中使用如下方法:

db.row_factory = sqlite3.Row

        這樣將使用Row物件而不是字典來返回查詢結果。這些都是namedtuple型別,因此我們可以通過索引或鍵來訪問它們。舉個栗子,假設我們有一個叫做r的sqlite3.Row物件,其欄位包括id,FirstName,LastName以及MiddleInitial:

>>> # You can get values based on the row's name
>>> r['FirstName']
John
>>> # Or, you can get them based on index
>>> r[1]
John
# Row objects are also iterable:
>>> for value in r:
...     print(value)
1
John
Doe
M

        此外,通過結合獲取遊標、執行和獲取結果的方式來提供一個查詢方法,這是一個不錯的想法:

def query_db(query, args=(), one=False):
    cur = get_db().execute(query, args)
    rv = cur.fetchall()
    cur.close()
    return (rv[0] if rv else None) if one else rv

        這個方便的小函式結合了行工廠,使與資料庫的工作比使用原始遊標和連線物件更加方便。

        這裡展示如何使用它:

for user in query_db('select * from users'):
    print user['username'], 'has the id', user['user_id']

        或者,如果你僅僅希望單個結果:

user = query_db('select * from users where username = ?',
                [the_username], one=True)
if user is None:
    print 'No such user'
else:
    print the_username, 'has the id', user['user_id']

        為了給SQL語句傳遞變數,在語句中使用一個“?”標記,並且將這個引數作為一個列表進行傳遞。永遠不要直接使用字串格式化的方式來新增它們到SQL語句中,因為這樣做有可能會因為SQL注入而讓應用程式遭到攻擊。

8.3 初始化框架

        關係型資料庫需要框架,因此應用程式經常會尋找一個schema.sql檔案來建立資料庫。基於這個框架檔案,在一個方法中進行資料庫的建立工作,這是一個不錯的想法。下面的方法展示瞭如何實現這個想法:

def init_db():
    with app.app_context():
        db = get_db()
        with app.open_resource('schema.sql', mode='r') as f:
            db.cursor().executescript(f.read())
        db.commit()

        隨後,你就可以在Python shell中建立這樣的一個數據庫了:

>>> from yourapplication import init_db
>>> init_db()


9. Flask中使用SQLAlchemy

        很多人更喜歡使用SQLAlchemy來訪問資料庫。在這種情況下,我們鼓勵使用包而不是Flask應用程式的模組,並將模型放入到單獨的模組(更大型的應用程式)中。雖然這不是必須的,但是這樣做很有意義。

        通常使用SQLAlchemy,僅需四個步驟。

9.1 Flask-SQLAlchmy擴充套件

        因為SQLAlchemy是一個常用的資料庫抽象層和物件關係對映器,它需要少量的配置工作,所以這裡有一個Flask擴充套件可以為你處理這些問題。如果你想快速入門,那麼這是最推薦的方式。

        你可以在PypI上下載Flask-SQLAlchemy

9.2 宣告

        SQLAlchemy中的宣告式擴充套件是使用SQLAlchemy最新的方法。它允許你同時定義表和模型,就像Django的工作方式一樣。除了下面要講的內容外,我還推薦了關於宣告擴充套件的官方文件。

        這裡有個示例,假設你的應用程式中有一個database.py模組:

from sqlalchemy import create_engine
from sqlalchemy.orm import scoped_session, sessionmaker
from sqlalchemy.ext.declarative import declarative_base


engine = create_engine('sqlite:////tmp/test.db', convert_unicode=True)
db_session = scoped_session(sessionmaker(autocommit=False,
                                         autoflush=False,
                                         bind=engine))
Base = declarative_base()
Base.query = db_session.query_property()


def init_db():
    # import all modules here that might define models so that
    # they will be registered properly on the metadata.  Otherwise
    # you will have to import them first before calling init_db()
    import yourapplication.models
    Base.metadata.create_all(bind=engine)

        為了定義你的模型,你需要定義一個類,這個類繼承於上面建立的Base類。如果你很驚奇為什麼我們不需要關心執行緒(就像我們在上面的SQLite3示例中使用g物件一樣):這是因為在呼叫scoped_session的時候,SQLAlchemy就已經為我們做了這些工作了。

        為了使用在你的應用程式中以一種宣告式的方式使用SQLAlchemy,你僅需要將下面的程式碼放到你的應用程式模組中。Flask會在請求結束或者應用程式關閉的時候自動地移除資料庫會話。

from yourapplication.database import db_session


@app.teardown_appcontext
def shutdown_session(exception=None):
    db_session.remove()

        這裡有一個模型的示例(比如,放到model.py模組中):

from sqlalchemy import Column, Integer, String
from yourapplication.database import Base


class User(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True)
    name = Column(String(50), unique=True)
    email = Column(String(120), unique=True)

    def __init__(self, name=None, email=None):
        self.name = name
        self.email = email

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

        為了建立資料庫,你可以使用init_db方法:

>>> from yourapplication.database import init_db
>>> init_db()

        你可以像下面這樣給資料庫新增資料:

>>> from yourapplication.database import db_session
>>> from yourapplication.models import User
>>> u = User('admin', '[email protected]')
>>> db_session.add(u)
>>> db_session.commit()

        查詢資料操作如下:

>>> User.query.all()
[<User u'admin'>]
>>> User.query.filter(User.name == 'admin').first()
<User u'admin'>

9.3 手動實現物件關係對映

        手動的物件關係對映與上面的宣告式方法相比各有千秋。兩者最主要的區別在於,手動的物件關係對映需要你單獨地定義表和類,然後將兩者對映起來。它很靈活,但是需要鍵入的內容更多。通常上,它執行方式就像宣告式方法,因此一定要將你的應用程式分解為一個包中的多個模組。

        這裡有一個關於你應用程式的database.py模組的示例:

from sqlalchemy import create_engine, MetaData
from sqlalchemy.orm import scoped_session, sessionmaker


engine = create_engine('sqlite:////tmp/test.db', convert_unicode=True)
metadata = MetaData()
db_session = scoped_session(sessionmaker(autocommit=False,
                                         autoflush=False,
                                         bind=engine))


def init_db():
    metadata.create_all(bind=engine)

        就像在宣告式方法中一樣,你需要在請求結束後或者應用程式上下文關閉時,關閉這個會話。將下面的程式碼放到你的應用程式模組中:

from yourapplication.database import db_session


@app.teardown_appcontext
def shutdown_session(exception=None):
    db_session.remove()

        這裡展示一個數據表和模型(程式碼放置到models.py中):

from sqlalchemy import Table, Column, Integer, String
from sqlalchemy.orm import mapper
from yourapplication.database import metadata, db_session


class User(object):
    query = db_session.query_property()

    def __init__(self, name=None, email=None):
        self.name = name
        self.email = email

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


users = Table('users', metadata,
    Column('id', Integer, primary_key=True),
    Column('name', String(50), unique=True),
    Column('email', String(120), unique=True)
)
mapper(User, users)

        查詢和新增工作就跟前一節的示例一樣。

9.4 SQL抽象層

        如果你僅僅希望使用資料庫系統(以及SQL)抽象層,基本上你僅需要這個引擎即可:

>>> con = engine.connect()
>>> con.execute(users.insert(), name='admin', email='[email protected]')

from sqlalchemy import create_engine, MetaData, Table


engine = create_engine('sqlite:////tmp/test.db', convert_unicode=True)
metadata = MetaData(bind=engine)

        然後你既可以像前面的示例那樣在你的程式碼中宣告資料表,也可以自動地載入它們:

from sqlalchemy import Table


users = Table('users', metadata, autoload=True)

        為了新增資料,你可以使用insert函式。我們必須首先獲取一個連線,然後我們就可以使用一個事務了:

>>> con = engine.connect()
>>> con.execute(users.insert(), name='admin', email='[email protected]')

        SQLAlchemy會為我們自動提交這個操作的。

        為了查詢你的資料庫,你可以直接使用這個引擎或者使用一個連線:

>>> users.select(users.c.id == 1).execute().first()
(1, u'admin', u'[email protected]')

        這些結果是元組型別:

>>> r = users.select(users.c.id == 1).execute().first()
>>> r['name']
u'admin'

        你也可以為execute()函式傳遞SQL命令的字串:

>>> engine.execute('select * from users where id = :1', [1]).first()
(1, u'admin', u'[email protected]')

        關於SQLAlchemy的更多資訊,請前往其官網


10. 上傳檔案

        正如你所知的,上傳檔案是一個老生常談的問題。檔案上傳的基礎思想是非常簡單的,它的主要工作內容如下:

        1. 一個<form>標籤被標記成enctype=multipart/form-data,並且在這個表單中放置了一個<input type=file>。

        2. 應用程式從請求物件上的files字典進行訪問檔案。

        3. 使用檔案的save()函式將檔案永久儲存在檔案系統的某個地方。

10.1 相關介紹

        讓我們從一個非常基礎的應用程式開始,它將一個檔案上傳到一個特定的上傳資料夾,並向用戶顯示一個檔案。讓我們看一下我們應用程式的自引導程式碼(bootstrapping):

import os
from flask import Flask, flash, request, redirect, url_for
from werkzeug.utils import secure_filename


UPLOAD_FOLDER = '/path/to/the/uploads'
ALLOWED_EXTENSIONS = set(['txt', 'pdf', 'png', 'jpg', 'jpeg', 'gif'])

app = Flask(__name__)
app.config['UPLOAD_FOLDER'] = UPLOAD_FOLDER

        首先,我們需要進行一些匯入操作。大多數匯入內容是可以直接理解的,werkzeug.secure_filename()稍後會進行解釋。UPLOAD_FOLDER是我們儲存上傳檔案的地方,ALLOWED_EXTENSIONS是一組允許上傳檔案的字尾。

        為什麼我們要限制上傳檔案的字尾?如果伺服器失直接傳送資料到客戶端的,那麼你可能不希望你的使用者能夠上傳所有東西。這樣做,你可以確保使用者不能上傳導致XSS問題的HTML檔案(參見Cross-Site Scripting(XSS))。如果服務能執行.php檔案,那麼也可以確保不會上傳這類檔案,但是誰在伺服器上安裝了PHP呢(意思是,一般python+flask不會用到PHP這個世界上最好的語言偷笑)?

        接下來,編寫檢查擴充套件是否有效的函式,以及上傳檔案並將使用者重定向到上傳檔案的URL:

def allowed_file(filename):
    return '.' in filename and \
           filename.rsplit('.', 1)[1].lower() in ALLOWED_EXTENSIONS


@app.route('/', methods=['GET', 'POST'])
def upload_file():
    if request.method == 'POST':
        # check if the post request has the file part
        if 'file' not in request.files:
            flash('No file part')
            return redirect(request.url)
        file = request.files['file']
        # if user does not select file, browser also
        # submit an empty part without filename
        if file.filename == '':
            flash('No selected file')
            return redirect(request.url)
        if file and allowed_file(file.filename):
            filename = secure_filename(file.filename)
            file.save(os.path.join(app.config['UPLOAD_FOLDER'], filename))
            return redirect(url_for('uploaded_file',
                                    filename=filename))
    return '''
    <!doctype html>
    <title>Upload new File</title>
    <h1>Upload new File</h1>
    <form method=post enctype=multipart/form-data>
      <input type=file name=file>
      <input type=submit value=Upload>
    </form>
    '''

        那麼secure_filename()方法實際的作用是什麼呢?這裡有一個宗旨叫做,永遠不要信任使用者輸入的東西。在這裡,對於上傳檔案的檔名來說也同樣不要去信任它。所有提交的表單都可能被遺忘,並且檔名也可能是不安全的。因此,你需要記住:在直接儲存一個檔案到檔案系統上時,一定要使用這個函式來確保檔名是安全的。

        專業資訊:

        所以,你感興趣的是secure_filename()方法能做什麼,並且如果你不使用它的話會帶來什麼問題?想象一下,如果某人將下面的內容作為檔名傳送給你的應用程式:

        filename = "../../../../home/username/.bashrc"

        假設這幾個../是正確的,你可以將這個路徑新增到UPLOAD_FOLDER中,這將導致使用者可能有能力修改伺服器檔案系統上的檔案,但實際上他應該不能修改。這確實是需要了解應用程式才能知道,但是相信我,黑客是很有耐心來破解這個路徑的。

        現在,讓我們看看這個方法是如何工作的:

>>> secure_filename('../../../../home/username/.bashrc')

'home_username_.bashrc'

        還有最後一件需要做的事:檔案上傳的服務。在upload_file()中,我們將使用者重定向到url_for('uploaded_file', filename=filename),即/uploads/filename。因此,我們要編寫uploaded_file()方法來返回上傳路徑下,上傳成功的這個檔案。在Flask0.5開始,我們可以使用一個方法來為我們實現這個功能:

from flask import send_from_directory


@app.route('/uploads/<filename>')
def uploaded_file(filename):
    return send_from_directory(app.config['UPLOAD_FOLDER'], filename)

        你可以為uploaded_file註冊build_obly規則,也可以使用SharedDataMiddleware來實現下載服務。這也能在其他舊版的Flask上工作:

from werkzeug import SharedDataMiddleware


app.add_url_rule('/uploads/<filename>', 'uploaded_file', build_only=True)
app.wsgi_app = SharedDataMiddleware(app.wsgi_app, {
    'uploads': app.config['UPLOAD_FOLDER']
})

        如果你現在執行應用程式,你會發現所有功能都符合預期。

10.2 改進上傳

        (新增於版本0.6。)

        Flask到底是如何處理上傳的呢?如果檔案很小,它會將檔案儲存到web服務的記憶體中,否則會儲存到本地臨時位置(就如tempfile.gettempdir()返回的內容一樣)。但是在上傳被中止後,你該人如何指定最大檔案的大小呢?預設情況下,Flask會很開心地接受檔案上傳到一個無限制的記憶體中,但是實際上我們買不起這麼誇張的硬體,因此正確的方法是在配置中設定配置項MAX_CONTENT_LENGTH來限制最大上傳檔案的大小:

from flask import Flask, Request


app = Flask(__name__)
app.config['MAX_CONTENT_LENGTH'] = 16 * 1024 * 1024

        上面的程式碼將把允許的最大有效負載限制為16Mb。如果一個更大的檔案要進行傳輸,Flask會丟擲一個RequestEntityTooLarge的異常。

        連線重置問題:

        當使用本地開發伺服器時,你可能獲取一個連線重置錯誤(connection reset error)而不是一個413響應。在使用生產WSGI伺服器執行應用程式時,你可以得到正確的狀態響應。

        這個特性是於Flask0.6新增的,但是在以前的版本中通過繼承請求物件也可以實現。有關資訊可以檢視Werkzeug文件的檔案處理(file handling)。

10.3 上傳進度條

        一段時間以前,許多開發人員都有這樣的想法:將傳輸來的檔案一小塊一小塊地讀取,並將上傳進度儲存在資料庫中,然後通過客戶端的JavaScript程式碼來讀取進度。簡單來說,客戶端每5秒就會查詢一次當前傳輸了多少。你感覺到這種諷刺了嗎?客戶端詢問一些它已經知道的事情。

10.4 一個更簡單的解決方案

        現在,這裡有更好的解決方案,它執行更快,並且更加可靠。有一些像JQuery這樣的JavaScript庫,它們應用表單外掛來簡化進度條的構建。

        因為在處理上傳的所有應用程式中,檔案上傳的常見模式幾乎沒有變化,因此Flask也提供了一個叫做Flask-Uploads的擴充套件來實現一個完整的上傳機制,還提供了包括檔案型別白名單、黑名單等多種功能。


11. 快取

        當你的應用程式執行緩慢的時候,試著為其增加一些快取。至少這是一種提高速度的最簡單方式。那麼快取能做什麼呢?比如,你有一個需要一段時間才能完成的方法,但是這個方法的返回結果可能在5分鐘內都是有效的,因此你可以將這個結果放到快取中一段時間,而不是每次需要這個值的時候都去執行一次這個方法。

        Flask本身並不提供快取功能,但是作為Flask基礎的Werkzeug庫,有一些非常基礎的快取支援。Werkzeug支援多種快取後端,其中最常見的是Memcached伺服器。

11.1 建立一個快取

        就像建立Flask物件那樣,你先建立一個快取物件,然後讓它一直存在。如果你使用了開發伺服器,你可以建立一個SimpleCache物件,這物件是一個簡單的快取,用於將元素快取到Python直譯器的記憶體中:

from werkzeug.contrib.cache import SimpleCache


cache = SimpleCache()

        如果你想要使用Memcached,請確保你有一個支援Memcache的模組(可以從PyPI上獲取),並且有一個可用的Memcached伺服器正在執行。然後你就可以像下面這樣連線到快取伺服器:

from werkzeug.contrib.cache import MemcachedCache


cache = MemcachedCache(['127.0.0.1:11211'])

        如果你正在使用App Engine,你可以通過下面的程式碼輕鬆連線到App Engine的快取伺服器:

from werkzeug.contrib.cache import GAEMemcachedCache


cache = GAEMemcachedCache()

11.2 使用一個快取

        現在怎樣使用一個快取呢?這裡有兩個非常重要的操作:get()set()

        為了從快取獲取一項資料,你可以呼叫get(),併為其傳遞一個字串作為鍵名。如果在這個快取中存在這個鍵的話,則返回對應的值,否則這個方法將返回None:

rv = cache.get('my-item')

        為了新增一項資料到快取中,使用set()函式。這個函式的第一個引數是鍵,第二個引數是需要被設定的值。同時,也可以提供一個timeout關鍵字引數,用於在達到這個引數規定的時