Flask基础知识总结
有一些東西是大多數網絡應用都會用到的。比如許多應用都會使用關系型數據庫和用戶 驗證,在請求之前連接數據庫并得到當前登錄用戶的信息,在請求之后關閉數據庫連接。
更多用戶貢獻的代碼片斷和方案參見?current_app
主要內容:
- 大型應用
- 應用工廠
- 應用調度
- 實現 API 異常處理
- URL 處理器
- 使用 Distribute 部署
- 使用 Fabric 部署
- 在 Flask 中使用 SQLite 3
- 在 Flask 中使用 SQLAlchemy
- 上傳文件
- 緩存
- 視圖裝飾器
- 使用 WTForms 進行表單驗證
- 模板繼承
- 消息閃現
- 通過 jQuery 使用 AJAX
- 自定義出錯頁面
- 惰性載入視圖
- 在 Flask 中使用 MongoKit
- 添加一個頁面圖標
- 流內容
- 延遲的請求回調
- 添加 HTTP 方法重載
- 請求內容校驗
- 基于 Celery 的后臺任務
大型應用
對于大型應用來說使用包代替模塊是一個好主意。使用包非常簡單。假設有一個小應用如下:
/yourapplication/yourapplication.py/static/style.css/templateslayout.htmlindex.htmllogin.html...簡單的包
要把上例中的小應用裝換為大型應用只要在現有應用中創建一個名為 yourapplication 的新文件夾,并把所有東西都移動到這個文件夾內。然后把 yourapplication.py 更名為?__init__.py?。(請首先刪除所有 .pyc 文件,否則基本上會出問題)
修改完后應該如下例:
/yourapplication/yourapplication/__init__.py/static/style.css/templateslayout.htmlindex.htmllogin.html...但是現在如何運行應用呢?原本的 python?yourapplication/__init__.py?無法運行了。因為 Python 不希望包內的模塊成為啟動文件。但是這不是一個大問題,只要在 yourapplication 文件夾旁添加一個?runserver.py?文件就可以了,其內容如下:
from yourapplication import app app.run(debug=True)我們從中學到了什么?現在我們來重構一下應用以適應多模塊。只要記住以下幾點:
Flask 應用對象必須位于?__init__.py文件中。這樣每個模塊就可以安全地導入了,且?__name__變量會解析到正確的包。
__init__.py?示例:
from flask import Flask app = Flask(__name__)import yourapplication.viewsviews.py 內容如下:
from yourapplication import app@app.route('/') def index():return 'Hello World!'最終全部內容如下:
/yourapplication/runserver.py/yourapplication/__init__.py/views.py/static/style.css/templateslayout.htmlindex.htmllogin.html...回環導入
回環導入是指兩個模塊互相導入,本例中我們添加的 views.py 就與?__init__.py?相互依賴。每個 Python 程序員都討厭回環導入。一般情況下回環導入是個壞主意,但 在這里一點問題都沒有。原因是我們沒有真正使用?__init__.py?中的視圖,只是 保證模塊被導入,并且我們在文件底部才這樣做。
但是這種方式還是有些問題,因為沒有辦法使用裝飾器。要找到解決問題的靈感請參閱大型應用一節。
使用藍圖
對于大型應用推薦把應用分隔為小塊,每個小塊使用藍圖輔助執行。關于這個主題的介紹 請參閱使用藍圖的模塊化應用一節 。
應用工廠
如果你已經在應用中使用了包和藍圖( 使用藍圖的模塊化應用 ),那么還有許多方法可以更 進一步地改進你的應用。常用的方案是導入藍圖后創建應用對象,但是如果在一個函數中創建對象,那么就可以創建多個實例。
那么這樣做有什么用呢?
用于測試。可以針對不同的情況使用不同的配置來測試應用。
那么如何做呢?
基礎工廠
方法是在一個函數中設置應用,具體如下:
def create_app(config_filename):app = Flask(__name__)app.config.from_pyfile(config_filename)from yourapplication.model import dbdb.init_app(app)from yourapplication.views.admin import adminfrom yourapplication.views.frontend import frontendapp.register_blueprint(admin)app.register_blueprint(frontend)return app這個方法的缺點是在導入時無法在藍圖中使用應用對象。但是你可以在一個請求中使用它。 如何通過配置來訪問應用?使用?<a rel="nofollow" href="http://dormousehole.readthedocs.org/en/latest/api.html#flask.current_app" "="" style="padding: 0px; margin: 0px; background-color: transparent; color: rgb(45, 133, 202);">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'])這里我們在配置中查找模板的名稱。
擴展對象初始化時不會綁定到一個應用,應用可以使用?db.init_app?來設置擴展。 擴展對象中不會儲存特定應用的狀態,因此一個擴展可以被多個應用使用。關于擴展設計的更多信息請參閱?Flask 擴展開發?。
當使用 Flask-SQLAlchemy 時,你的 model.py 可能是這樣的:
from flask.ext.sqlalchemy import SQLAlchemy # no app object passed! Instead we use use db.init_app in the factory. db = SQLAlchemy()# create some models使用應用
因此,要使用這樣的應用就必須先創建它。下面是一個運行應用的示例 run.py 文件:
from yourapplication import create_app app = create_app('/path/to/config.cfg') app.run()改進工廠
上面的工廠函數還不是足夠好,可以改進的地方主要有以下幾點:
為了單元測試,要想辦法傳入配置,這樣就不必在文件系統中創建配置文件。
當設置應用時從藍圖調用一個函數,這樣就可以有機會修改屬性(如掛接請求前/后 處理器等)。
應用調度
應用調度是在 WSGI 層面組合多個 WSGI 應用的過程。可以組合多個 Flask 應用,也可以 組合 Flask 應用和其他 WSGI 應用。通過這種組合,如果有必要的話,甚至可以在同一個 解釋器中一邊運行 Django ,一邊運行 Flask 。這種組合的好處取決于應用內部是如何 工作的。
應用調度與模塊化的最大不同在于應用調度中的每個 應用是完全獨立的,它們以各自的配置運行,并在 WSGI 層面被調度。
說明
下面所有的技術說明和舉例都歸結于一個可以運行于任何 WSGI 服務器的 application 對象。對于生產環境,參見 部署方式 。對于開發環境, Werkzeug 提供了一個內建開發服務器,它使用?werkzeug.serving.run_simple()?來運行:
from werkzeug.serving import run_simple run_simple('localhost', 5000, application, use_reloader=True)注意?werkzeug.wsgi.DispatcherMiddleware?。其原理是:每個獨立的 Flask 應用都 是一個合法的 WSGI 應用,它們通過調度中間件組合為一個基于前綴調度的大應用。
假設你的主應用運行于 / ,后臺接口位于 /backend:
from werkzeug.wsgi import DispatcherMiddleware from frontend_app import application as frontend from backend_app import application as backendapplication = DispatcherMiddleware(frontend, {'/backend': backend })根據子域調度
有時候你可能需要使用不同的配置來運行同一個應用的多個實例。可以把應用創建過程 放在一個函數中,這樣調用這個函數就可以創建一個應用的實例,具體實現參見應用工廠方案。
最常見的做法是每個子域創建一個應用,配置服務器來調度所有子域的應用請求,使用 子域來創建用戶自定義的實例。一旦你的服務器可以監聽所有子域,那么就可以使用一個很簡單的 WSGI 應用來動態創建應用了。
WSGI 層是完美的抽象層,因此可以寫一個你自己的 WSGI 應用來監視請求,并把請求分配給你的 Flask 應用。如果被分配的應用還沒有創建,那么就會動態創建應用并被登記下來:
from threading import Lockclass SubdomainDispatcher(object):def __init__(self, domain, create_app):self.domain = domainself.create_app = create_appself.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] = appreturn appdef __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 NotFounddef make_app(subdomain):user = get_user_for_subdomain(subdomain)if user is None:# 如果子域沒有對應的用戶,那么還是得返回一個 WSGI 應用# 用于處理請求。這里我們把 NotFound() 異常作為應用返回,# 它會被渲染為一個缺省的 404 頁面。然后,可能還需要把# 用戶重定向到主頁。return NotFound()# 否則為特定用戶創建應用return create_app(user)application = SubdomainDispatcher('example.com', make_app)根據路徑調度
根據 URL 的路徑調度非常簡單。上面,我們通過查找 Host 頭來判斷子域,現在只要查找請求路徑的第一個斜杠之前的路徑就可以了:
from threading import Lock from werkzeug.wsgi import pop_path_info, peek_path_infoclass PathDispatcher(object):def __init__(self, default_app, create_app):self.default_app = default_appself.create_app = create_appself.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] = appreturn appdef __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_appreturn app(environ, start_response)與根據子域調度相比最大的不同是:根據路徑調度時,如果創建函數返回 None ,那么就會回落到另一個應用:
from myapplication import create_app, default_app, get_user_for_prefixdef make_app(prefix):user = get_user_for_prefix(prefix)if user is not None:return create_app(user)application = PathDispatcher(default_app, make_app)實現 API 異常處理
在 Flask 上經常會執行 RESTful API 。開發者首先會遇到的問題之一是用于 API 的內建異常處理不給力,回饋的內容不是很有用。
對于非法使用 API ,比使用 abort 更好的解決方式是實現你自己的異常處理類型, 并安裝相應句柄,輸出符合用戶格式要求的出錯信息。
簡單的異常類
基本的思路是引入一個新的異常,回饋一個合適的可讀性高的信息、一個狀態碼和一些可選的負載,給錯誤提供更多的環境內容。
以下是一個簡單的示例:
from flask import jsonifyclass InvalidUsage(Exception):status_code = 400def __init__(self, message, status_code=None, payload=None):Exception.__init__(self)self.message = messageif status_code is not None:self.status_code = status_codeself.payload = payloaddef to_dict(self):rv = dict(self.payload or ())rv['message'] = self.messagereturn rv這樣一個視圖就可以拋出帶有出錯信息的異常了。另外,還可以通過 payload 參數以字典的形式提供一些額外的負載。
注冊一個錯誤處理句柄
現在,視圖可以拋出異常,但是會立即引發一個內部服務錯誤。這是因為沒有為這個錯誤處理類注冊句柄。句柄增加很容易,例如:
@app.errorhandler(InvalidAPIUsage) def handle_invalid_usage(error):response = jsonify(error.to_dict())response.status_code = error.status_codereturn response在視圖中的用法
以下是如何在視圖中使用該功能:
@app.route('/foo') def get_foo():raise InvalidUsage('This view is gone', status_code=410)URL 處理器
New in version 0.7.
Flask 0.7 引入了 URL 處理器,其作用是為你處理大量包含相同部分的 URL 。假設你有 許多 URL 都包含語言代碼,但是又不想在每個函數中都重復處理這個語言代碼,那么就可可以使用 URL 處理器。
在與藍圖配合使用時, URL 處理器格外有用。下面我們分別演示在應用中和藍圖中使用 URL 處理器。
國際化應用的 URL
假設有應用如下:
from flask import Flask, gapp = 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:returnif app.url_map.is_endpoint_expecting(endpoint, 'lang_code'):values['lang_code'] = g.lang_codeURL 映射的?url_value_preprocessor()?。這些函數在請求匹配后立即根據 URL 的值執行代碼。它們可以從 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, gapp = Flask(__name__)@app.url_defaults def add_language_code(endpoint, values):if 'lang_code' in values or not g.lang_code:returnif 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():...國際化的藍圖 URL
因為藍圖可以自動給所有 URL 加上一個統一的前綴,所以應用到每個函數就非常方便了。 更進一步,因為藍圖 URL 預處理器不需要檢查 URL 是否真的需要要一個 'lang_code' 參數,所以可以去除?distribute?的前身是 setuptools ,它是一個擴展庫,通常用于分發 Python 庫和 擴展。它的英文名稱的就是“分發”的意思。它擴展了 Python 自帶的一個基礎模塊安裝 系統 distutils ,支持多種更復雜的結構,方便了大型應用的分發部署。它的主要特色:
- 支持依賴 :一個庫或者應用可以聲明其所依賴的其他庫的列表。依賴庫將被自動 安裝。
- 包注冊 :可以在安裝過程中注冊包,這樣就可以通過一個包查詢其他包的信息。 這套系統最有名的功能是“切入點”,即一個包可以定義一個入口,以便于其他包掛接, 用以擴展包。
- 安裝管理 : distribute 中的 easy_install 可以為你安裝其他庫。你也可以使用早晚會替代 easy_install 的?包?。 distribute 不支持分發標準模塊,因此我們不 討論模塊的問題。關于如何把模塊轉換為包的信息參見大型應用方案。
使用 distribute 將使發布更復雜,也更加自動化。如果你想要完全自動化處理,請同時閱讀使用 Fabric 部署一節。
基礎設置腳本
因為你已經安裝了 Flask ,所以你應當已經安裝了 setuptools 或 distribute 。如果 沒有安裝,不用怕,有一個?最好使用 virtualenv?。
你的設置代碼應用放在 setup.py 文件中,這個文件應當位于應用旁邊。這個文件名只是 一個約定,但是最好不要改變,因為大家都會去找這個文件。
是的,即使你使用 distribute ,你導入的包也是 setuptools 。 distribute 完全向后兼容于 setuptools ,因此它使用相同的導入名稱。
Flask 應用的基礎 setup.py 文件示例如下:
from setuptools import setupsetup(name='Your Application',version='1.0',long_description=__doc__,packages=['yourapplication'],include_package_data=True,zip_safe=False,install_requires=['Flask'] )請記住,你必須顯式的列出子包。如果你要 distribute 自動為你搜索包,你可以使用 find_packages 函數:
from setuptools import setup, find_packagessetup(...packages=find_packages() )大多數 setup 的參數可以望文生義,但是 include_package_data 和 zip_safe 可以不容易理解。 include_package_data 告訴 distribute 要搜索一個 MANIFEST.in 文件,把文件內容所匹配的所有條目作為包數據安裝。可以通過使用這個參數分發 Python 模塊的靜態文件和模板(參見分發資源 )。 zip_safe 標志可用于強制或防止創建 zip 壓縮包。通常你不會想要把包安裝為 zip 壓縮文件,因為一些工具不支持壓縮文件,而且壓縮文件比較難以調試。
分發資源
如果你嘗試安裝上文創建的包,你會發現諸如 static 或 templates 之類的文件夾沒有被安裝。原因是 distribute 不知道要為你添加哪些文件。你要做的是:在你的 setup.py 文件旁邊創建一個 MANIFEST.in 文件。這個文件列出了所有應當添加到 tar 壓縮包的文件:
recursive-include yourapplication/templates * recursive-include yourapplication/static *不要忘了把 setup 函數的 include_package_data 參數設置為 True !否則即使把內容在 MANIFEST.in 文件中全部列出來也沒有用。
聲明依賴
依賴是在 install_requires 參數中聲明的,這個參數是一個列表。列表中的每一項都是一個需要在安裝時從 PyPI 獲得的包。缺省情況下,總是會獲得最新版本的包,但你可以指定最高版本和最低版本。示例:
install_requires=['Flask>=0.2','SQLAlchemy>=0.6','BrokenPackage>=0.7,<=1.0' ]我前面提到,依賴包都從 PyPI 獲得的。但是如果要從別的地方獲得包怎么辦呢?你只要 還是按照上述方法寫,然后提供一個可選地址列表就行了:
dependency_links=['http://example.com/yourfiles']請確保頁面上有一個目錄列表,且頁面上的鏈接指向正確的 tar 壓縮包。這樣 distribute 就會找到文件了。如果你的包在公司內部網絡上,請提供指向服務器的 URL 。
安裝 / 開發
要安裝你的應用(理想情況下是安裝到一個 virtualenv ),只要運行帶 install 參數的 setup.py 腳本就可以了。它會將你的應用安裝到 virtualenv 的 site-packages 文件夾下,同時下載并安裝依賴:
$ python setup.py install如果你正開發這個包,同時也希望相關依賴被安裝,那么可以使用 develop 來代替:
$ python setup.py develop這樣做的好處是只安裝一個指向 site-packages 的連接,而不是把數據復制到那里。這樣在開發過程中就不必每次修改以后再運行 install 了。
使用 Fabric 部署
Fabric 是一個 Python 工具,與 Makefiles 類似,但是能夠在遠程服務器上執行命令。如果與適當的 Python 包( 大型應用 )與優良的配置( 配置管理 )相結合那么 Fabric 將是在外部服務器上部署 Flask 的利器。
在下文開始之前,有幾點需要明確:
Fabric 1.0 需要要被安裝到本地。本教程假設使用的是最新版本的 Fabric 。
應用已經是一個包,且有一個可用的 setup.py 文件( 使用 Distribute 部署 )。
- 在下面的例子中,我們假設遠程服務器使用 mod_wsgi 。當然,你可以使用你自己 喜歡的服務器,但是在示例中我們選擇 Apache + mod_wsgi ,因為它們設置方便, 且在沒有 root 權限情況下可以方便的重載應用。
創建第一個 Fabfile
fabfile 是控制 Fabric 的東西,其文件名為 fabfile.py ,由 fab 命令執行。在這個文件中定義的所有函數都會被視作 fab 子命令。這些命令將會在一個或多個主機上運行。這些主機可以在 fabfile 中定義,也可以在命令行中定義。本例將在 fabfile 中 定義主機。
下面是第一個例子,比較級。它可以把當前的源代碼上傳至服務器,并安裝到一個預先存在的 virtual 環境:
from fabric.api import *# 使用遠程命令的用戶名 env.user = 'appuser' # 執行命令的服務器 env.hosts = ['server1.example.com', 'server2.example.com']def pack():# 創建一個新的分發源,格式為 tar 壓縮包local('python setup.py sdist --formats=gztar', capture=False)def deploy():# 定義分發版本的名稱和版本號dist = local('python setup.py --fullname', capture=True).strip()# 把 tar 壓縮包格式的源代碼上傳到服務器的臨時文件夾put('dist/%s.tar.gz' % dist, '/tmp/yourapplication.tar.gz')# 創建一個用于解壓縮的文件夾,并進入該文件夾run('mkdir /tmp/yourapplication')with cd('/tmp/yourapplication'):run('tar xzf /tmp/yourapplication.tar.gz')# 現在使用 virtual 環境的 Python 解釋器來安裝包run('/var/www/yourapplication/env/bin/python setup.py install')# 安裝完成,刪除文件夾run('rm -rf /tmp/yourapplication /tmp/yourapplication.tar.gz')# 最后 touch .wsgi 文件,讓 mod_wsgi 觸發應用重載run('touch /var/www/yourapplication.wsgi')上例中的注釋詳細,應當是容易理解的。以下是 fabric 提供的最常用命令的簡要說明:
- run - 在遠程服務器上執行一個命令
- local - 在本地機器上執行一個命令
- put - 上傳文件到遠程服務器上
- cd - 在服務器端改變目錄。必須與 with 語句聯合使用。
運行 Fabfile
那么如何運行 fabfile 呢?答案是使用 fab 命令。要在遠程服務器上部署當前版本的代碼可以使用這個命令:
$ fab pack deploy但是這個命令需要遠程服務器上已經創建了 /var/www/yourapplication 文件夾,且 /var/www/yourapplication/env 是一個 virtual 環境。更進一步,服務器上還沒有創建配置文件和 .wsgi 文件。那么,我們如何在一個新的服務器上創建一個基礎環境呢?
這個問題取決于你要設置多少臺服務器。如果只有一臺應用服務器(多數情況下),那么在 fabfile 中創建命令有一點多余。當然,你可以這么做。這個命令可以稱之為 setup 或 bootstrap 。在使用命令時顯式傳遞服務器名稱:
$ fab -H newserver.example.com bootstrap設置一個新服務器大致有以下幾個步驟:
- 在 /var/www 創建目錄結構: $ mkdir /var/www/yourapplication $ cd /var/www/yourapplication $ virtualenv --distribute env
上傳一個新的 application.wsgi 文件和應用配置文件(如 application.cfg ) 到服務器上。
- 創建一個新的用于 yourapplication 的 Apache 配置并激活它。要確保激活 .wsgi 文件變動監視,這樣在 touch 的時候可以自動重載應用。( 更多信息參見?flask.g?對象是綁定到請求的,而不是應用環境。
示例:
@app.route('/') def index():cur = get_db().cursor()...Note 請記住,解散請求和應用環境的函數是一定會被執行的。即使請求前處理器執行失敗或根本沒有執行,解散函數也會被執行。因此,我們必須保證在關閉數據庫連接之前 數據庫連接是存在的。
按需連接
上述方式(在第一次使用時連接數據庫)的優點是只有在真正需要時才打開數據庫連接。 如果你想要在一個請求環境之外使用數據庫連接,那么你可以手動在 Python 解釋器打開應用環境:
with app.app_context():# now you can use get_db()簡化查詢
現在,在每個請求處理函數中可以通過訪問 g.db 來得到當前打開的數據庫連接。為了簡化 SQLite 的使用,這里有一個有用的行工廠函數。該函數會轉換每次從數據庫返回的 結果。例如,為了得到字典類型而不是元組類型的返回結果,可以這樣:
def make_dicts(cursor, row):return dict((cur.description[idx][0], value)for idx, value in enumerate(row))db.row_factory = make_dicts或者更簡單的:
db.row_factory = sqlite3.Row此外,把得到游標,執行查詢和獲得結果組合成一個查詢函數不失為一個好辦法:
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 語句中,這樣會給應用帶來?SQLAlchemy?來訪問數據庫。建議在你的 Flask 應用中使用包來代替 模塊,并把模型放入一個獨立的模塊中(參見大型應用?)。雖然這不是必須的,但是很有用。
有四種 SQLAlchemy 的常用方法,下面一一道來:
Flask-SQLAlchemy 擴展
因為 SQLAlchemy 是一個常用的數據庫抽象層,并且需要一定的配置才能使用,因此我們為你做了一個處理 SQLAlchemy 的擴展。如果你需要快速的開始使用 SQLAlchemy ,那么推薦你使用這個擴展。
你可以從?Flask-SQLAlchemy?。
聲明
SQLAlchemy 中的聲明擴展是使用 SQLAlchemy 的最新方法,它允許你像 Django 一樣, 在一個地方定義表和模型然后到處使用。除了以下內容,我建議你閱讀聲明的官方文檔。
以下是示例 database.py 模塊:
from sqlalchemy import create_engine from sqlalchemy.orm import scoped_session, sessionmaker from sqlalchemy.ext.declarative import declarative_baseengine = 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():# 在這里導入定義模型所需要的所有模塊,這樣它們就會正確的注冊在# 元數據上。否則你就必須在調用 init_db() 之前導入它們。import yourapplication.modelsBase.metadata.create_all(bind=engine)要定義模型的話,只要繼承上面創建的 Base 類就可以了。你可能會奇怪這里為什么不用理會線程(就像我們在 SQLite3 的例子中一樣使用 g 對象)。 原因是 SQLAlchemy 已經用 scoped_session 為我們做好了此 類工作。
如果要在應用中以聲明方式使用 SQLAlchemy ,那么只要把下列代碼加入應用模塊就可以了。 Flask 會自動在請求結束時或者應用關閉時刪除數據庫會話:
from yourapplication.database import db_session@app.teardown_appcontext def shutdown_session(exception=None):db_session.remove()以下是一個示例模型(放入 models.py 中):
from sqlalchemy import Column, Integer, String from yourapplication.database import Baseclass 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 = nameself.email = emaildef __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', 'admin@localhost') >>> db_session.add(u) >>> db_session.commit()查詢很簡單:
>>> User.query.all() [<User u'admin'>] >>> User.query.filter(User.name == 'admin').first() <User u'admin'>人工對象關系映射
人工對象關系映射相較于上面的聲明方式有優點也有缺點。主要區別是人工對象關系映射 分別定義表和類并映射它們。這種方式更靈活,但是要多些代碼。通常,這種方式與聲明 方式一樣運行,因此請確保把你的應用在包中分為多個模塊。
示例 database.py 模塊:
from sqlalchemy import create_engine, MetaData from sqlalchemy.orm import scoped_session, sessionmakerengine = 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_sessionclass User(object):query = db_session.query_property()def __init__(self, name=None, email=None):self.name = nameself.email = emaildef __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)查詢和插入與聲明方式的一樣。
SQL 抽象層
如果你只需要使用數據庫系統(和 SQL )抽象層,那么基本上只要使用引擎:
from sqlalchemy import create_engine, MetaDataengine = create_engine('sqlite:tmp/test.db', convert_unicode=True) metadata = MetaData(bind=engine)然后你要么像前文中一樣在代碼中聲明表,要么自動載入它們:
users = Table('users', metadata, autoload=True)可以使用 insert 方法插入數據。為了使用事務,我們必須先得到一個連接:
>>> con = engine.connect() >>> con.execute(users.insert(), name='admin', email='admin@localhost')SQLAlchemy 會自動提交。
可以直接使用引擎或連接來查詢數據庫:
>>> users.select(users.c.id == 1).execute().first() (1, u'admin', u'admin@localhost')查詢結果也是類字典元組:
>>> r = users.select(users.c.id == 1).execute().first() >>> r['name'] u'admin'你也可以把 SQL 語句作為字符串傳遞給 execute() 方法:
>>> engine.execute('select * from users where id = :1', [1]).first() (1, u'admin', u'admin@localhost')關于 SQLAlchemy 的更多信息請移步其save()?方法把文件永久 地保存在文件系統中。
- JumpLoader?- Java
簡介
讓我們從一個基本的應用開始,這個應用上傳文件到一個指定目錄,并把文件顯示給用戶。 以下是應用的引導代碼:
import os from flask import Flask, request, redirect, url_for from werkzeug import secure_filenameUPLOAD_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 是允許上傳的文件擴展名的集合。接著我們給應用手動添加了一個 URL 規則。一般現在不會做這個,但是為什么這么做了呢?原因是我們需要服務器(或我們的開發服務器)為我們提供 服務。因此我們只生成這些文件的 URL 的規則。
為什么要限制文件件的擴展名呢?如果直接向客戶端發送數據,那么你可能不會想讓用戶上傳任意文件。否則,你必須確保用戶不能上傳 HTML 文件,因為 HTML 可能引起 XSS 問題(參見跨站腳本攻擊(XSS) )。如果服務器可以執行 PHP 文件,那么還必須確保不允許上傳 .php 文件。但是誰又會在服務器上安裝 PHP 呢,對不? :)
下一個函數檢查擴展名是否合法,上傳文件,把用戶重定向到已上傳文件的 URL:
def allowed_file(filename):return '.' in filename and \filename.rsplit('.', 1)[1] in ALLOWED_EXTENSIONS@app.route('/', methods=['GET', 'POST']) def upload_file():if request.method == 'POST':file = request.files['file']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 action="" method=post enctype=multipart/form-data><p><input type=file name=file><input type=submit value=Upload></form>'''那么?Plupload?- HTML5, Java, Flash
一個更簡便的方案
因為所有應用中上傳文件的方案基本相同,因此可以使用 Flask-Uploads 擴展來實現 文件上傳。這個擴展實現了完整的上傳機制,還具有白名單功能、黑名單功能以及其他 功能。
緩存
當你的應用變慢的時候,可以考慮加入緩存。至少這是最簡單的加速方法。緩存有什么用? 假設有一個函數耗時較長,但是這個函數在五分鐘前返回的結果還是正確的。那么我們就 可以考慮把這個函數的結果在緩存中存放一段時間。
Flask 本身不提供緩存,但是它的基礎庫之一 Werkzeug 有一些非常基本的緩存支持。它支持多種緩存后端,通常你需要使用的是 memcached 服務器。
設置一個緩存
與創建?SimpleCache?對象。這個對象提供簡單的緩存,它把 緩存項目保存在 Python 解釋器的內存中:
from werkzeug.contrib.cache import SimpleCache cache = SimpleCache()如果你要使用 memcached ,那么請確保有 memcache 模塊支持(你可以從 PyPI 獲得模塊)和一個正在運行的 memcached 服務器。 連接 memcached 服務器示例:
from werkzeug.contrib.cache import MemcachedCache cache = MemcachedCache(['127.0.0.1:11211'])如果你正在使用 App Engine ,那么你可以方便地連接到 App Engine memcache 服務器:
from werkzeug.contrib.cache import GAEMemcachedCache cache = GAEMemcachedCache()使用緩存
現在的問題是如何使用緩存呢?有兩個非常重要的操作:?set()?。下面展示如何使用它們:
get() 用于從緩存中獲得項目,調用時使用 一個字符串作為鍵名。如果項目存在,那么就會返回這個項目,否則返回 None:
rv = cache.get('my-item')set() 用于把項目添加到緩存中。第一個參數是鍵名,第二個參數是鍵值。還可以提供一個超時參數,當超過時間后項目會自動刪除。
下面是一個完整的例子:
def get_my_item():rv = cache.get('my-item')if rv is None:rv = calculate_value()cache.set('my-item', rv, timeout=5 * 60)return rv視圖裝飾器
Python 有一個非常有趣的功能:函數裝飾器。這個功能可以使網絡應用干凈整潔。 Flask 中的每個視圖都是一個裝飾器,它可以被注入額外的功能。你可以已經用過了?functools.wraps()?)。
下面是檢查登錄裝飾器的例子。假設登錄頁面為 'login' ,當前用戶被儲存在 g.user 中,如果還沒有登錄,其值為 None:
from functools import wraps from flask import g, request, redirect, url_fordef login_required(f):@wraps(f)def decorated_function(*args, **kwargs):if g.user is None:return redirect(url_for('login', next=request.url))return f(*args, **kwargs)return decorated_function如何使用這個裝飾器呢?把這個裝飾器放在最靠近函數的地方就行了。當使用更進一步的裝飾器時,請記住要把 route() 裝飾器放在最外面:
@app.route('/secret_page') @login_required def secret_page():pass緩存裝飾器
假設有一個視圖函數需要消耗昂貴的計算成本,因此你需要在一定時間內緩存這個視圖的 計算結果。這種情況下裝飾器是一個好的選擇。我們假設你像緩存方案中一樣設置了緩存。
下面是一個示例緩存函數。它根據一個特定的前綴(實際上是一個格式字符串)和請求的 當前路徑生成緩存鍵。注意,我們先使用了一個函數來創建裝飾器,這個裝飾器用于裝飾 函數。聽起來拗口吧,確實有一點復雜,但是下面的示例代碼還是很容易讀懂的。
被裝飾代碼按如下步驟工作
代碼:
from functools import wraps from flask import requestdef cached(timeout=5 * 60, key='view/%s'):def decorator(f):@wraps(f)def decorated_function(*args, **kwargs):cache_key = key % request.pathrv = cache.get(cache_key)if rv is not None:return rvrv = f(*args, **kwargs)cache.set(cache_key, rv, timeout=timeout)return rvreturn decorated_functionreturn decorator注意,以上代碼假設存在一個可用的實例化的 cache 對象,更多信息參見緩存方案。
模板裝飾器
不久前, TurboGear 的人發明了模板裝飾器這個通用模式。其工作原理是返回一個字典, 這個字典包含從視圖傳遞給模板的值,模板自動被渲染。以下三個例子的功能是相同的:
@app.route('/') def index():return render_template('index.html', value=42)@app.route('/') @templated('index.html') def index():return dict(value=42)@app.route('/') @templated() def index():return dict(value=42)正如你所見,如果沒有提供模板名稱,那么就會使用 URL 映射的端點(把點轉換為斜杠) 加上 '.html' 。如果提供了,那么就會使用所提供的模板名稱。當裝飾器函數返回 時,返回的字典就被傳送到模板渲染函數。如果返回的是 None ,就會使用空字典。如果 返回的不是字典,那么就會直接傳遞原封不動的返回值。這樣就可以仍然使用重定向函數或 返回簡單的字符串。
以下是裝飾器的代碼:
from functools import wraps from flask import requestdef templated(template=None):def decorator(f):@wraps(f)def decorated_function(*args, **kwargs):template_name = templateif template_name is None:template_name = request.endpoint \.replace('.', '/') + '.html'ctx = f(*args, **kwargs)if ctx is None:ctx = {}elif not isinstance(ctx, dict):return ctxreturn render_template(template_name, **ctx)return decorated_functionreturn decorator端點裝飾器
當你想要使用 werkzeug 路由系統,以便于獲得更強的靈活性時,需要和 Rule 中定義的一樣,把端點映射到視圖函數。這樣就需要 用的裝飾器了。例如:
from flask import Flask from werkzeug.routing import Ruleapp = Flask(__name__) app.url_map.add(Rule('/', endpoint='index'))@app.endpoint('index') def my_index():return "Hello world"使用 WTForms 進行表單驗證
當你必須處理瀏覽器提交的表單數據時,視圖代碼很快會變得難以閱讀。有一些庫可以簡化這個工作,其中之一便是?WTForms 官方網站?。
模板繼承
Jinja 最有力的部分就是模板繼承。模板繼承允許你創建一個基礎“骨架”模板。這個模板中包含站點的常用元素,定義可以被子模板繼承的 塊 。
聽起來很復雜其實做起來簡單,看看下面的例子就容易理解了。
基礎模板
這個模板的名稱是 layout.html ,它定義了一個簡單的 HTML 骨架,用于顯示一個簡單的兩欄頁面。“子”模板的任務是用內容填充空的塊:
<!doctype html> <html><head>{% block head %}<link rel="stylesheet" href="{{ url_for('static', filename='style.css') }}"><title>{% block title %}{% endblock %} - My Webpage</title>{% endblock %}</head><body><div id="content">{% block content %}{% endblock %}</div><div id="footer">{% block footer %}© Copyright 2010 by <a href="http://domain.invalid/">you</a>.{% endblock %}</div></body> </html>在這個例子中, {% block %} 標記定義了四個可以被子模板填充的塊。 block 標記告訴模板引擎這是一個可以被子模板重載的部分。
子模板
子模板示例:
{% extends "layout.html" %} {% block title %}Index{% endblock %} {% block head %}{{ super() }}<style type="text/css">.important { color: #336699; }</style> {% endblock %} {% block content %}<h1>Index</h1><p class="important">Welcome on my awesome homepage. {% endblock %}這里 {% extends %} 標記是關鍵,它告訴模板引擎這個模板“擴展”了另一個模板, 當模板系統評估這個模板時會先找到父模板。這個擴展標記必須是模板中的第一個標記。 如果要使用父模板中的塊內容,請使用 {{ super() }} 。
消息閃現
一個好的應用和用戶界面都需要良好的反饋。如果用戶得不到足夠的反饋,那么應用最終會被用戶唾棄。 Flask 的閃現系統提供了一個良好的反饋方式。閃現系統的基本工作方式 是:在且只在下一個請求中訪問上一個請求結束時記錄的消息。一般我們結合布局模板來使用閃現系統。
簡單的例子
以下是一個完整的示例:
from flask import Flask, flash, redirect, render_template, \request, url_forapp = Flask(__name__) app.secret_key = 'some_secret'@app.route('/') def index():return render_template('index.html')@app.route('/login', methods=['GET', 'POST']) def login():error = Noneif request.method == 'POST':if request.form['username'] != 'admin' or \request.form['password'] != 'secret':error = 'Invalid credentials'else:flash('You were successfully logged in')return redirect(url_for('index'))return render_template('login.html', error=error)if __name__ == "__main__":app.run()以下是實現閃現的 layout.html 模板:
<!doctype html> <title>My Application</title> {% with messages = get_flashed_messages() %}{% if messages %}<ul class=flashes>{% for message in messages %}<li>{{ message }}</li>{% endfor %}</ul>{% endif %} {% endwith %} {% block body %}{% endblock %}以下是 index.html 模板:
{% extends "layout.html" %} {% block body %}<h1>Overview</h1><p>Do you want to <a href="{{ url_for('login') }}">log in?</a> {% endblock %}login 模板:
{% extends "layout.html" %} {% block body %}<h1>Login</h1>{% if error %}<p class=error><strong>Error:</strong> {{ error }}{% endif %}<form action="" method=post><dl><dt>Username:<dd><input type=text name=username value="{{request.form.username }}"><dt>Password:<dd><input type=password name=password></dl><p><input type=submit value=Login></form> {% endblock %}閃現消息的類別
New in version 0.3.
閃現消息還可以指定類別,如果沒有指定,那么缺省的類別為 'message' 。不同的 類別可以給用戶提供更好的反饋。例如錯誤消息可以使用紅色背景。
使用 flash() 函數可以指定消息的類別:
flash(u'Invalid password provided', 'error')模板中的?get_flashed_messages()?的 結果。這個功能有助于在不同位置顯示不同類別的消息。
{% with errors = get_flashed_messages(category_filter=["error"]) %} {% if errors %} <div class="alert-message block-message error"><a class="close" href="#">×</a><ul>{%- for msg in errors %}<li>{{ msg }}</li>{% endfor -%}</ul> </div> {% endif %} {% endwith %}通過 jQuery 使用 AJAX
http://example.com/myapp> )上呢?在服務端,這個問題不成為問題,可以使用 url_for() 函數來得到答案。但是如果我們使用 jQuery ,那么就不能硬碼應用的路徑,只能使用動態路徑。怎么辦?
一個簡單的方法是在頁面中添加一個 script 標記,設置一個全局變量來表示應用的根 路徑。示例:
<script type=text/javascript>$SCRIPT_ROOT = {{ request.script_root|tojson|safe }}; </script>在 Flask 0.10 版本以前版本中,使用 |safe 是有必要的,是為了使 Jinja 不要轉義 JSON 編碼的字符串。通常這樣做不是必須的,但是在 script 內部我們需要這么做。
進一步說明
在 HTML 中, script 標記是用于聲明 CDATA 的,也就是說聲明的內容不會被解析。?之間的內容都會被作為腳本處理。這也意味著 在 script 標記之間不會存在任何 </ 。在這里 |tojson 會正確處理問題, 并為你轉義斜杠( {{ ""|tojson|safe }} 會被渲染為 "<\/script>" )。
在 Flask 0.10 版本中更進了一步,把所有 HTML 標記都用 unicode 轉義了,這樣使 Flask 自動把 HTML 轉換為安全標記。
JSON 視圖函數
現在讓我們來創建一個服務端函數,這個函數接收兩個 URL 參數(兩個需要相加的數字 ),然后向應用返回一個 JSON 對象。下面這個例子是非常不實用的,因為一般會在客戶端完成類似工作,但這個例子可以簡單明了地展示如何使用 jQuery 和 Flask:
from flask import Flask, jsonify, render_template, request app = Flask(__name__)@app.route('/_add_numbers') def add_numbers():a = request.args.get('a', 0, type=int)b = request.args.get('b', 0, type=int)return jsonify(result=a + b)@app.route('/') def index():return render_template('index.html')正如你所見,我還添加了一個 index 方法來渲染模板。這個模板會按前文所述載入 jQuery 。模板中有一個用于兩個數字相加的表單和一個觸發服務端函數的鏈接。
注意,這里我們使用了?示例源代碼?。
自定義出錯頁面
Flask 有一個方便的?掌握應用錯誤)。
出錯處理器
一個出錯處理器是一個函數,就像一個視圖函數一樣。與視圖函數不同之處在于出錯處理器 在出現錯誤時被調用,且傳遞錯誤。錯誤大多數是一個?errorhandler()?裝飾器注冊,注冊時應提供異常的 出代碼。請記住, Flask 不會 為你設置出錯代碼,因此請確保在返回響應時,同時提供 HTTP 狀態代碼。
以下是一個處理 “404 Page Not Found” 異常的示例:
from flask import render_template@app.errorhandler(404) def page_not_found(e):return render_template('404.html'), 404示例模板:
{% extends "layout.html" %} {% block title %}Page Not Found{% endblock %} {% block body %}<h1>Page Not Found</h1><p>What you were looking for is just not there.<p><a href="{{ url_for('index') }}">go somewhere nice</a> {% endblock %}惰性載入視圖
Flask 通常使用裝飾器。裝飾器簡單易用,只要把 URL 放在相應的函數的前面就可以了。 但是這種方式有一個缺點:使用裝飾器的代碼必須預先導入,否則 Flask 就無法真正找到你的函數。
當你必須快速導入應用時,這就會成為一個問題。在 Google App Engine 或其他系統中, 必須快速導入應用。因此,如果你的應用存在這個問題,那么必須使用集中 URL 映射。
add_url_rule()?,加上應用前綴和點符號:
def url(url_rule, import_name, **options):view = LazyView('yourapplication.' + import_name)app.add_url_rule(url_rule, view_func=view, **options)url('/', 'views.index') url('/user/<username>', 'views.user')有一件事情要牢記:請求前和請求后處理器必須在第一個請求前導入。
其余的裝飾器可以同樣用上述方法改寫。
在 Flask 中使用 MongoKit
現在使用文檔型數據庫來取代關系型數據庫已越來越常見。本方案展示如何使用 MongoKit ,它是一個用于操作 MongoDB 的文檔映射庫。
本方案需要一個運行中的 MongoDB 服務器和已安裝好的 MongoKit 庫。
使用 MongoKit 有兩種常用的方法,下面逐一說明:
聲明
聲明是 MongoKit 的缺省行為。這個思路來自于 Django 或 SQLAlchemy 的聲明。
下面是一個示例 app.py 模塊:
from flask import Flask from mongokit import Connection, Document# configuration MONGODB_HOST = 'localhost' MONGODB_PORT = 27017# create the little application object app = Flask(__name__) app.config.from_object(__name__)# connect to the database connection = Connection(app.config['MONGODB_HOST'],app.config['MONGODB_PORT'])如果要定義模型,那么只要繼承 MongoKit 的 Document 類就行了。如果你已經讀過 SQLAlchemy 方案,那么可以會奇怪這里為什么沒有使用會話,甚至沒有定義一個 init_db 函數。一方面是因為 MongoKit 沒有類似會話在東西。有時候這樣會多寫一點代碼,但會使它的速度更快。另一方面是因為 MongoDB 是無模式的。這就意味著可以在 插入數據的時候修改數據結構。 MongoKit 也是無模式的,但會執行一些驗證,以確保數據的完整性。
以下是一個示例文檔(把示例內容也放入 app.py ):
def max_length(length):def validate(value):if len(value) <= length:return Trueraise Exception('%s must be at most %s characters long' % length)return validateclass User(Document):structure = {'name': unicode,'email': unicode,}validators = {'name': max_length(50),'email': max_length(120)}use_dot_notation = Truedef __repr__(self):return '<User %r>' % (self.name)在當前連接中注冊用戶文檔
connection.register([User]) 上例展示如何定義模式(命名結構)和字符串最大長度驗證器。上例中還使用了一個 MongoKit 中特殊的 use_dot_notation 功能。缺省情況下, MongoKit 的運作方式和 Python 的字典類似。但是如果 use_dot_notation 設置為 True ,那么就可幾乎像其他 ORM 一樣使用點符號來分隔屬性。
可以像下面這樣把條目插入數據庫中:
>>> from yourapplication.database import connection >>> from yourapplication.models import User >>> collection = connection['test'].users >>> user = collection.User() >>> user['name'] = u'admin' >>> user['email'] = u'admin@localhost' >>> user.save()注意, MongoKit 對于列類型的使用是比較嚴格的。對于 name 和 email 列,你都不能使用 str 類型,應當使用 unicode 。
查詢非常簡單:
>>> list(collection.User.find()) [<User u'admin'>] >>> collection.User.find_one({'name': u'admin'}) <User u'admin'>PyMongo 兼容層
如果你只需要使用 PyMongo ,也可以使用 MongoKit 。在這種方式下可以獲得最佳的性能。注意,以下示例中,沒有 MongoKit 與 Flask 整合的內容,整合的方式參見上文:
from MongoKit import Connectionconnection = Connection()使用 insert 方法可以插入數據。但首先必須先得到一個連接。這個連接類似于 SQL 界的表。
>>> collection = connection['test'].users >>> user = {'name': u'admin', 'email': u'admin@localhost'} >>> collection.insert(user)MongoKit 會自動提交。
直接使用集合查詢數據庫:
>>> list(collection.find()) [{u'_id': ObjectId('4c271729e13823182f000000'), u'name': u'admin', u'email': u'admin@localhost'}] >>> collection.find_one({'name': u'admin'}) {u'_id': ObjectId('4c271729e13823182f000000'), u'name': u'admin', u'email': u'admin@localhost'}查詢結果為類字典對象:
>>> r = collection.find_one({'name': u'admin'}) >>> r['email'] u'admin@localhost'關于 MongoKit 的更多信息,請移步其send_from_directory()?函數來寫一個視圖:
import os from flask import send_from_directory@app.route('/favicon.ico') def favicon():return send_from_directory(os.path.join(app.root_path, 'static'),'favicon.ico', mimetype='image/vnd.microsoft.icon')上例中的 MIME 類型可以省略,瀏覽器會自動猜測類型。但是我們在例子中明確定義了, 省去了額外的猜測,反正這個類型是不變的。
上例會通過你的應用來提供圖標,如果可能的話,最好配置你的專用服務器來提供圖標, 配置方法參見網頁服務器的文檔。
另見 Wikipedia 上的update_template_context()?,以確保 更新被渲染的內容。這樣,模板遍歷流內容。由于每次產生內容后,服務器都會把內容 發送給客戶端,因此可能需要緩存來保存內容。我們使用了 rv.enable_buffering(size) 來進行緩存。 5 是一個比較明智的缺省值。
環境中的流內容
New in version 0.9.
注意,當你生成流內容時,請求環境已經在函數執行時消失了。 Flask 0.9 為你提供了 一點幫助,讓你可以在生成器運行期間保持請求環境:
from flask import stream_with_context, request, Response@app.route('/stream') def streamed_response():def generate():yield 'Hello 'yield request.args['name']yield '!'return Response(stream_with_context(generate()))如果沒有使用?RuntimeError?錯誤。
延遲的請求回調
Flask 的設計思路之一是:響應對象創建后被傳遞給一串回調函數,這些回調函數可以修改 或替換響應對象。當請求處理開始的時候,響應對象還沒有被創建。響應對象是由一個視圖函數或者系統中的其他組件按需創建的。
但是當響應對象還沒有創建時,我們如何修改響應對象呢?比如在一個請求前函數中,我們需要根據響應對象設置一個 cookie 。
通常我們選擇避開這種情形。例如可以嘗試把應用邏輯移動到請求后函數中。但是,有時候這個方法讓人不爽,或者讓代碼變得很丑陋。
變通的方法是把一堆回調函數貼到 g 對象上,并且在請求結束時調用這些回調函數。這樣你就可以在應用的任意地方延遲回調函數的執行。
裝飾器
下面的裝飾器是一個關鍵,它在 g 對象上注冊一個函數列表:
from flask import gdef after_this_request(f):if not hasattr(g, 'after_request_callbacks'):g.after_request_callbacks = []g.after_request_callbacks.append(f)return f調用延遲的回調函數
至此,通過使用 after_this_request 裝飾器,使得函數在請求結束時可以被調用。現在我們來實現這個調用過程。我們把這些函數注冊為 after_request() 回調函數:
@app.after_request def call_after_request_callbacks(response):for callback in getattr(g, 'after_request_callbacks', ()):callback(response)return response一個實例
現在我們可以方便地隨時隨地為特定請求注冊一個函數,讓這個函數在請求結束時被調用。 例如,你可以在請求前函數中把用戶的當前語言記錄到 cookie 中:
from flask import request@app.before_request def detect_user_language():language = request.cookies.get('user_lang')if language is None:language = guess_language_from_request()@after_this_requestdef remember_language(response):response.set_cookie('user_lang', language)g.language = languageLogo Table Of Contents添加 HTTP 方法重載
一些 HTTP 代理不支持所有 HTTP 方法或者不支持一些較新的 HTTP 方法(例如 PACTH )。在這種情況下,可以通過使用完全相反的協議,用一種 HTTP 方法來“代理”另一種 HTTP 方法。
實現的思路是讓客戶端發送一個 HTTP POST 請求,并設置 X-HTTP-Method-Override 頭部為需要的 HTTP 方法(例如 PATCH )。
通過 HTTP 中間件可以方便的實現:
class HTTPMethodOverrideMiddleware(object):allowed_methods = frozenset(['GET','HEAD','POST','DELETE','PUT','PATCH','OPTIONS'])bodyless_methods = frozenset(['GET', 'HEAD', 'OPTIONS', 'DELETE'])def __init__(self, app):self.app = appdef __call__(self, environ, start_response):method = environ.get('HTTP_X_HTTP_METHOD_OVERRIDE', '').upper()if method in self.allowed_methods:method = method.encode('ascii', 'replace')environ['REQUEST_METHOD'] = methodif method in self.bodyless_methods:environ['CONTENT_LENGTH'] = '0'return self.app(environ, start_response)通過以下代碼就可以與 Flask 一同工作了:
from flask import Flaskapp = Flask(__name__) app.wsgi_app = HTTPMethodOverrideMiddleware(app.wsgi_app)請求內容校驗
請求數據會由不同的代碼來處理或者預處理。例如 JSON 數據和表單數據都來源于已經 讀取并處理的請求對象,但是它們的處理代碼是不同的。這樣,當需要校驗進來的請求 數據時就會遇到麻煩。因此,有時候就有必要使用一些 API 。
幸運的是可以通過包裝輸入流來方便地改變這種情況。
下面的例子演示在 WSGI 環境下讀取和儲存輸入數據,得到數據的 SHA1 校驗:
import hashlibclass ChecksumCalcStream(object):def __init__(self, stream):self._stream = streamself._hash = hashlib.sha1()def read(self, bytes):rv = self._stream.read(bytes)self._hash.update(rv)return rvdef readline(self, size_hint):rv = self._stream.readline(size_hint)self._hash.update(rv)return rvdef generate_checksum(request):env = request.environstream = ChecksumCalcStream(env['wsgi.input'])env['wsgi.input'] = streamreturn stream._hash要使用上面的類,你只要在請求開始消耗數據之前鉤接要計算的流就可以了。(按:小心操作 request.form 或類似東西。例如 before_request_handlers 就應當小心不要操作。)
用法示例:
@app.route('/special-api', methods=['POST']) def special_api():hash = generate_checksum(request)# Accessing this parses the input streamfiles = request.files# At this point the hash is fully constructed.checksum = hash.hexdigest()return 'Hash was: %s' % checksum基于 Celery 的后臺任務
Celery 是一個 Python 編寫的是一個異步任務隊列/基于分布式消息傳遞的作業隊列。 以前它有一個 Flask 的集成,但是從版本 3 開始,它進行了一些內部的重構,已經不需要這個集成了。本文主要說明如何在 Flask 中正確使用 Celery 。本文假設你 已經閱讀過了其官方文檔中的 Celery 入門
安裝 Celery
Celery 在 Python 包索引( PyPI )上榜上有名,因此可以使用 pip 或 easy_install 之類標準的 Python 工具來安裝:
$ pip install celery配置 Celery
你首先需要有一個 Celery 實例,這個實例稱為 celery 應用。其地位就相當于 Flask 中 Flask 一樣。這個實例被用作所有 Celery 相關事務的入口,例如創建 任務、管理工人等等。因此它必須可以被其他模塊導入。
例如,你可以把它放在一個 tasks 模塊中。這樣不需要重新配置,你就可以使用 tasks 的子類,增加 Flask 應用環境的支持,并鉤接 Flask 的配置。
只要如下這樣就可以在 Falsk 中使用 Celery 了:
from celery import Celerydef make_celery(app):celery = Celery(app.import_name, broker=app.config['CELERY_BROKER_URL'])celery.conf.update(app.config)TaskBase = celery.Taskclass ContextTask(TaskBase):abstract = Truedef __call__(self, *args, **kwargs):with app.app_context():return TaskBase.__call__(self, *args, **kwargs)celery.Task = ContextTaskreturn celery這個函數創建了一個新的 Celery 對象,使用了應用配置中的 broker ,并從 Flask 配置中升級了 Celery 的其余配置。然后創建了一個任務子類,在一個應用環境中包裝了任務執行。
最小的例子
基于上文,以下是一個在 Flask 中使用 Celery 的最小例子:
from flask import Flaskapp = Flask(__name__) app.config.update(CELERY_BROKER_URL='redis://localhost:6379',CELERY_RESULT_BACKEND='redis://localhost:6379' ) celery = make_celery(app)@celery.task() def add_together(a, b):return a + b這個任務現在可以在后臺調用了:
>>> result = add_together.delay(23, 42) >>> result.wait() 65運行 Celery 工人
至此,如果你已經按上文一步一步執行,你會失望地發現你的 .wait() 不會真正返回。這是因為你還沒有運行 celery 。你可以這樣以工人方式運行 celery:
$ celery -A your_application worker把 your_application 字符串替換為你創建 celery 對像的應用包或模塊。
總結
以上是生活随笔為你收集整理的Flask基础知识总结的全部內容,希望文章能夠幫你解決所遇到的問題。
- 上一篇: 传智播客java测试题_传智播客Java
- 下一篇: ts自动编译声明文件_拥抱 TS:细数选