single_file_tasksチュートリアルで学んだことを書き留める。
download
* 完了したタスク一覧表示、完了したタスクを差し戻し機能をつけてみた。
ファイル構成
ファイル | 説明 |
---|---|
tasks.py | python code |
schema.sql | 初期DB設定sql |
tasks.db | タスクデータを格納 |
static/style.css | css |
templates/layout.mako | layout |
templates/new.mako | タスク作成 |
templates/list.mako | タスク一覧 |
templates/notfound.mako | 404 |
main処理
# 宣言
settings = {}
settings['reload_all'] = True
settings['debug_all'] = True
settings['mako.directories'] = os.path.join(here, 'templates')
settings['db'] = os.path.join(here, 'tasks.db')
# session作成。sessionはcookieベース。(document.cookieで確認できます。)
session_factory = UnencryptedCookieSessionFactoryConfig('itsaseekreet')
# pyramidに設定を流し込む
config = Configurator(settings=settings, session_factory=session_factory)
# パスの設定
config.add_route('list', '/')
config.add_route('new', '/new')
config.add_route('close', '/close/{id}')
# 静的パスの指定
config.add_static_view('static', os.path.join(here, 'static'))
# デコレーター(@view_config と @subscriber)を探して設定に加える
config.scan()
# webサーバー設定
app = config.make_wsgi_app()
server = make_server('0.0.0.0', 8080, app)
server.serve_forever()
pyramid.events.subscriber
- @subscriber(NewRequest)
リクエスト要求に対し、一番最初に実行されます。(synfonyで言うところのpreExecute())
DB接続処理をしています。リクエスト終了時にDB接続を切断する処理を入れています。(postExecute)
- @subscriber(ApplicationCreated)
webサーバーが起動した直後に1度だけ呼ばれます。
pyramid.httpexceptions.HTTPFound
return HTTPFound(location=request.route_url('list'))
ページリロードによる2重登録抑止のため特定のページにリダイレクトさせます。
logging
import logging
logging.basicConfig()
log = logging.getLogger(__file__)
log.warn('NewRequest')
- WARNING:tasks.py:NewRequest というメッセージが表示されます。
- debug, info, warning, error, critical とあるが、開発中のみdebugメッセージを表示させるなどできる。
@reify
関数の結果をキャッシュして、初回以降は処理を行わず結果だけを返します。
pyramid.session
session_factory = UnencryptedCookieSessionFactoryConfig('itsaseekreet')
config = Configurator(settings=settings, session_factory=session_factory)
request.session.flash('Task was successfully reopen!') #セッションに値をセット
#template側:
% if request.session.peek_flash(): #セッションに値があるか確認
request.session.pop_flash() #セッション値を表示
% endif
sqlite3
select
rs = request.db.execute("select id, name from tasks where closed = 0")
tasks = [dict(id=row[0], name=row[1]) for row in rs.fetchall()]
1レコードだけ取得
row = rs.fetchone()
tasks = [dict(id=row[0], name=row[1])]
selectの結果をカラム名で参照
request.db = sqlite3.connect(settings['db'])
+ request.db.row_factory = sqlite3.Row
- tasks = [dict(id=row[0], name=row[1]) for row in rs.fetchall()]
+ tasks = [dict(id=row["id"], name=row["name"]) for row in rs.fetchall()]
update
task_id = int(request.matchdict['id']) #型変換してセット
#placeholder
request.db.execute("update tasks set closed = ? where id = ?", (0, task_id))
request.db.commit() #commitしないと保存されません。
自動コミットON
request.db = sqlite3.connect(settings['db'])
request.db.isolation_level = None
# もしくは
request.db = sqlite3.connect(settings['db'], isolation_level = None)
rollback
request.db.rollback()
pyramid.config.Configurator
Configuratorに値を格納
mysettings['val'] = "hello"
config = Configurator(settings=mysettings)
Configuratorに格納した値を参照(@subscribers)
request = event.request
settings = request.registry.settings
log.warn(settings['val'])
Configuratorに格納した値を参照(@view)
settings = request.registry.settings
log.warn(settings['val'])
development.ini、production.iniの値をrequest.registry.settingsに格納
{'debug_routematch': True,
'pyramid.default_locale_name': 'en',
'db': '/home/user/local/virtualenv/tasks/tasks.db',
'pyramid.reload_templates': True,
'debug_templates': True,
'debug_all': True,
'reload_templates': True,
'mako.directories': '/home/user/local/virtualenv/tasks/templates',
'pyramid.debug_routematch': True,
'reload_resources': True,
'default_locale_name': 'en',
'pyramid.reload_assets': True,
'reload_all': True,
'debug_authorization': True,
'pyramid.debug_authorization': True,
'pyramid.reload_resources': True,
'reload_assets': True,
'pyramid.debug_notfound': True,
'pyramid.debug_templates': True,
'prevent_http_cache': False,
'debug_notfound': True,
'pyramid.prevent_http_cache': False}
404ページ
@view_config(context='pyramid.exceptions.NotFound', renderer='notfound.mako')
def notfound_view(request):
request.response.status = '404 Not Found'
return {}
どんな例外もキャッチ
- 例:makoで存在しないurlを指定した時。
${request.route_url('forbidden_page')}
- 参考
@view_config(context=Exception, renderer='exception.mako')
def error_view(exception, request):
return {'message':exception}
その他
ファイルのディレクトリ(絶対パス)を取得
here = os.path.dirname(os.path.abspath(__file__))
使用例
os.path.join(here, 'schema.sql')
settings['mako.directories'] = os.path.join(here, 'templates')