19:使用SQLAlchemy的数据库

使用sqlite数据库上的sqlAlchemy ORM存储和检索数据。

背景

我们基于金字塔的维基应用程序现在需要数据库支持的页面存储。这通常意味着一个SQL数据库。金字塔社区强烈支持 SQLAlchemy 项目及其发展趋势 object-relational mapper (ORM) 作为连接到数据库的一种便捷的、毕达式的方式。

在这个步骤中，我们将sqlachemy连接到一个sqlite数据库表，在前面的步骤中为wiki页面提供存储和检索。

备注

Pyramidcookiecutter pyramid-cookiecutter-starter 对于启动一个SQLAlchemy项目非常有帮助，包括生成控制台脚本。因为我们希望看到所有的决定，所以我们将放弃本教程中的便利性，并自己连接起来。

目标

  • 使用sqlAlchemy模型将页面存储在sqlite中。

  • 使用SQLAlchemy查询列出/添加/查看/编辑页面。

  • 通过编写棱锥体提供数据库初始化命令 控制台脚本 可以从命令行运行。

步骤

  1. 我们将使用表单步骤作为起点：

    cd ..; cp -r forms databases; cd databases

  2. 我们需要在 databases/setup.py 以及一个 entry point 对于命令行脚本：

     1from setuptools import setup
 2
 3# List of dependencies installed via `pip install -e .`
 4# by virtue of the Setuptools `install_requires` value below.
 5requires = [
 6    'deform',
 7    'pyramid',
 8    'pyramid_chameleon',
 9    'pyramid_tm',
10    'sqlalchemy',
11    'waitress',
12    'zope.sqlalchemy',
13]
14
15# List of dependencies installed via `pip install -e ".[dev]"`
16# by virtue of the Setuptools `extras_require` value in the Python
17# dictionary below.
18dev_requires = [
19    'pyramid_debugtoolbar',
20    'pytest',
21    'webtest',
22]
23
24setup(
25    name='tutorial',
26    install_requires=requires,
27    extras_require={
28        'dev': dev_requires,
29    },
30    entry_points={
31        'paste.app_factory': [
32            'main = tutorial:main'
33        ],
34        'console_scripts': [
35            'initialize_tutorial_db = tutorial.initialize_db:main'
36        ],
37    },
38)

    备注

    我们还没做呢 $VENV/bin/pip install -e . 因为我们需要先编写一个脚本并更新配置。

  3. 我们的配置文件位于 databases/development.ini wires together some new pieces:

    [app:main]
use = egg:tutorial
pyramid.reload_templates = true
pyramid.includes =
    pyramid_debugtoolbar
    pyramid_tm

sqlalchemy.url = sqlite:///%(here)s/sqltutorial.sqlite

[server:main]
use = egg:waitress#main
listen = localhost:6543

# Begin logging configuration

[loggers]
keys = root, tutorial, sqlalchemy.engine.Engine

[logger_tutorial]
level = DEBUG
handlers =
qualname = tutorial

[handlers]
keys = console

[formatters]
keys = generic

[logger_root]
level = INFO
handlers = console

[logger_sqlalchemy.engine.Engine]
level = INFO
handlers =
qualname = sqlalchemy.engine.Engine

[handler_console]
class = StreamHandler
args = (sys.stderr,)
level = NOTSET
formatter = generic

[formatter_generic]
format = %(asctime)s %(levelname)-5.5s [%(name)s][%(threadName)s] %(message)s

# End logging configuration

  4. 现在需要通过以下更改将此引擎配置读取到应用程序中 databases/tutorial/__init__.py ：

     1from pyramid.config import Configurator
 2
 3from sqlalchemy import engine_from_config
 4
 5from .models import DBSession, Base
 6
 7def main(global_config, **settings):
 8    engine = engine_from_config(settings, 'sqlalchemy.')
 9    DBSession.configure(bind=engine)
10    Base.metadata.bind = engine
11
12    config = Configurator(settings=settings,
13                          root_factory='tutorial.models.Root')
14    config.include('pyramid_chameleon')
15    config.add_route('wiki_view', '/')
16    config.add_route('wikipage_add', '/add')
17    config.add_route('wikipage_view', '/{uid}')
18    config.add_route('wikipage_edit', '/{uid}/edit')
19    config.add_static_view('deform_static', 'deform:static/')
20    config.scan('.views')
21    return config.make_wsgi_app()

  5. 在处生成命令行脚本 databases/tutorial/initialize_db.py 初始化数据库：

     1import os
 2import sys
 3import transaction
 4
 5from sqlalchemy import engine_from_config
 6
 7from pyramid.paster import (
 8    get_appsettings,
 9    setup_logging,
10    )
11
12from .models import (
13    DBSession,
14    Page,
15    Base,
16    )
17
18
19def usage(argv):
20    cmd = os.path.basename(argv[0])
21    print('usage: %s <config_uri>\n'
22          '(example: "%s development.ini")' % (cmd, cmd))
23    sys.exit(1)
24
25
26def main(argv=sys.argv):
27    if len(argv) != 2:
28        usage(argv)
29    config_uri = argv[1]
30    setup_logging(config_uri)
31    settings = get_appsettings(config_uri)
32    engine = engine_from_config(settings, 'sqlalchemy.')
33    DBSession.configure(bind=engine)
34    Base.metadata.create_all(engine)
35    with transaction.manager:
36        model = Page(title='Root', body='<p>Root</p>')
37        DBSession.add(model)

  6. 现在我们已经准备好了所有的部件，因为我们改变了 setup.py ，我们现在安装所有产品：

    $VENV/bin/pip install -e .

  7. 脚本引用了 databases/tutorial/models.py ：

     1from pyramid.authorization import Allow, Everyone
 2
 3from sqlalchemy import (
 4    Column,
 5    Integer,
 6    Text,
 7    )
 8
 9from sqlalchemy.ext.declarative import declarative_base
10
11from sqlalchemy.orm import (
12    scoped_session,
13    sessionmaker,
14    )
15
16from zope.sqlalchemy import register
17
18DBSession = scoped_session(sessionmaker())
19register(DBSession)
20Base = declarative_base()
21
22
23class Page(Base):
24    __tablename__ = 'wikipages'
25    uid = Column(Integer, primary_key=True)
26    title = Column(Text, unique=True)
27    body = Column(Text)
28
29
30class Root:
31    __acl__ = [(Allow, Everyone, 'view'),
32               (Allow, 'group:editors', 'edit')]
33
34    def __init__(self, request):
35        pass

  8. 让我们运行这个控制台脚本，从而生成我们的数据库和表：

    $VENV/bin/initialize_tutorial_db development.ini

2016-04-16 13:01:33,055 INFO  [sqlalchemy.engine.Engine][MainThread] SELECT CAST('test plain returns' AS VARCHAR(60)) AS anon_1
2016-04-16 13:01:33,055 INFO  [sqlalchemy.engine.Engine][MainThread] ()
2016-04-16 13:01:33,056 INFO  [sqlalchemy.engine.Engine][MainThread] SELECT CAST('test unicode returns' AS VARCHAR(60)) AS anon_1
2016-04-16 13:01:33,056 INFO  [sqlalchemy.engine.Engine][MainThread] ()
2016-04-16 13:01:33,057 INFO  [sqlalchemy.engine.Engine][MainThread] PRAGMA table_info("wikipages")
2016-04-16 13:01:33,057 INFO  [sqlalchemy.engine.Engine][MainThread] ()
2016-04-16 13:01:33,058 INFO  [sqlalchemy.engine.Engine][MainThread]
CREATE TABLE wikipages (
       uid INTEGER NOT NULL,
       title TEXT,
       body TEXT,
       PRIMARY KEY (uid),
       UNIQUE (title)
)


2016-04-16 13:01:33,058 INFO  [sqlalchemy.engine.Engine][MainThread] ()
2016-04-16 13:01:33,059 INFO  [sqlalchemy.engine.Engine][MainThread] COMMIT
2016-04-16 13:01:33,062 INFO  [sqlalchemy.engine.Engine][MainThread] BEGIN (implicit)
2016-04-16 13:01:33,062 INFO  [sqlalchemy.engine.Engine][MainThread] INSERT INTO wikipages (title, body) VALUES (?, ?)
2016-04-16 13:01:33,063 INFO  [sqlalchemy.engine.Engine][MainThread] ('Root', '<p>Root</p>')
2016-04-16 13:01:33,063 INFO  [sqlalchemy.engine.Engine][MainThread] COMMIT

  9. 现在我们的数据由SQLAlchemy查询驱动，我们需要更新 databases/tutorial/views.py ：

     1import colander
 2import deform.widget
 3
 4from pyramid.httpexceptions import HTTPFound
 5from pyramid.view import view_config
 6
 7from .models import DBSession, Page
 8
 9
10class WikiPage(colander.MappingSchema):
11    title = colander.SchemaNode(colander.String())
12    body = colander.SchemaNode(
13        colander.String(),
14        widget=deform.widget.RichTextWidget()
15    )
16
17
18class WikiViews:
19    def __init__(self, request):
20        self.request = request
21
22    @property
23    def wiki_form(self):
24        schema = WikiPage()
25        return deform.Form(schema, buttons=('submit',))
26
27    @property
28    def reqts(self):
29        return self.wiki_form.get_widget_resources()
30
31    @view_config(route_name='wiki_view', renderer='wiki_view.pt')
32    def wiki_view(self):
33        pages = DBSession.query(Page).order_by(Page.title)
34        return dict(title='Wiki View', pages=pages)
35
36    @view_config(route_name='wikipage_add',
37                 renderer='wikipage_addedit.pt')
38    def wikipage_add(self):
39        form = self.wiki_form.render()
40
41        if 'submit' in self.request.params:
42            controls = self.request.POST.items()
43            try:
44                appstruct = self.wiki_form.validate(controls)
45            except deform.ValidationFailure as e:
46                # Form is NOT valid
47                return dict(form=e.render())
48
49            # Add a new page to the database
50            new_title = appstruct['title']
51            new_body = appstruct['body']
52            DBSession.add(Page(title=new_title, body=new_body))
53
54            # Get the new ID and redirect
55            page = DBSession.query(Page).filter_by(title=new_title).one()
56            new_uid = page.uid
57
58            url = self.request.route_url('wikipage_view', uid=new_uid)
59            return HTTPFound(url)
60
61        return dict(form=form)
62
63
64    @view_config(route_name='wikipage_view', renderer='wikipage_view.pt')
65    def wikipage_view(self):
66        uid = int(self.request.matchdict['uid'])
67        page = DBSession.query(Page).filter_by(uid=uid).one()
68        return dict(page=page)
69
70
71    @view_config(route_name='wikipage_edit',
72                 renderer='wikipage_addedit.pt')
73    def wikipage_edit(self):
74        uid = int(self.request.matchdict['uid'])
75        page = DBSession.query(Page).filter_by(uid=uid).one()
76
77        wiki_form = self.wiki_form
78
79        if 'submit' in self.request.params:
80            controls = self.request.POST.items()
81            try:
82                appstruct = wiki_form.validate(controls)
83            except deform.ValidationFailure as e:
84                return dict(page=page, form=e.render())
85
86            # Change the content and redirect to the view
87            page.title = appstruct['title']
88            page.body = appstruct['body']
89            url = self.request.route_url('wikipage_view', uid=uid)
90            return HTTPFound(url)
91
92        form = self.wiki_form.render(dict(
93            uid=page.uid, title=page.title, body=page.body)
94        )
95
96        return dict(page=page, form=form)

  10. 我们的测试 databases/tutorial/tests.py 已更改为包括SQLAlchemy引导：

     1import unittest
 2import transaction
 3
 4from pyramid import testing
 5
 6
 7def _initTestingDB():
 8    from sqlalchemy import create_engine
 9    from .models import (
10        DBSession,
11        Page,
12        Base
13        )
14    engine = create_engine('sqlite://')
15    Base.metadata.create_all(engine)
16    DBSession.configure(bind=engine)
17    with transaction.manager:
18        model = Page(title='FrontPage', body='This is the front page')
19        DBSession.add(model)
20    return DBSession
21
22
23class WikiViewTests(unittest.TestCase):
24    def setUp(self):
25        self.session = _initTestingDB()
26        self.config = testing.setUp()
27
28    def tearDown(self):
29        self.session.remove()
30        testing.tearDown()
31
32    def test_wiki_view(self):
33        from tutorial.views import WikiViews
34
35        request = testing.DummyRequest()
36        inst = WikiViews(request)
37        response = inst.wiki_view()
38        self.assertEqual(response['title'], 'Wiki View')
39
40
41class WikiFunctionalTests(unittest.TestCase):
42    def setUp(self):
43        from pyramid.paster import get_app
44        app = get_app('development.ini')
45        from webtest import TestApp
46        self.testapp = TestApp(app)
47
48    def tearDown(self):
49        from .models import DBSession
50        DBSession.remove()
51
52    def test_it(self):
53        res = self.testapp.get('/', status=200)
54        self.assertIn(b'Wiki: View', res.body)
55        res = self.testapp.get('/add', status=200)
56        self.assertIn(b'Add/Edit', res.body)

  11. 使用运行包中的测试 pytest ：

    $VENV/bin/pytest tutorial/tests.py -q
..
2 passed in 1.41 seconds

  12. 运行 Pyramid 应用程序时使用：

    $VENV/bin/pserve development.ini --reload

  13. 在浏览器中打开http://localhost:6543/。

分析

让我们从依赖关系开始。我们决定使用 SQLAlchemy 与我们的数据库交谈。不过，我们也安装了 pyramid_tm 和 zope.sqlalchemy . 为什么？

Pyramid对支持 transactions . 具体来说，您可以将事务管理器作为中间件或Pyramid“中间层”安装到应用程序中。然后，在返回响应之前，将执行应用程序的所有事务感知部分。

这意味着Pyramid视图代码通常不管理事务。如果视图代码或模板生成错误，事务管理器将中止事务。这是一种非常自由的代码编写方式。

这个 pyramid_tm 软件包提供一个“tween”，它在 development.ini 配置文件。安装它。然后，我们需要一个使sqlacalchemy，从而RDBMS事务管理器与Pyramid事务管理器集成的包。就是这样 zope.sqlalchemy 做。

我们应该指向磁盘上用于sqlite文件的位置？在配置文件中。这使我们的包的消费者可以以安全（非代码）的方式更改位置。也就是说，在配置中。在Pyramid中不需要这种面向配置的方法；您仍然可以在 __init__.py 或者一些配套的模块。

这个 initialize_tutorial_db 是一个很好的框架支持示例。您将设置指向 [console_scripts] ，这些将生成到虚拟环境的 bin 目录。我们的控制台脚本遵循的模式是使用所有引导向配置文件提供数据。然后，它打开sqlAlchemy并创建wiki的根目录，wiki还生成sqlite文件。注意 with transaction.manager 将工作放在事务范围内的部分，因为我们不在自动完成此操作的Web请求内。

这个 models.py 做一些额外的工作，将SQLAlchemy连接到Pyramid事务管理器中。然后它为 Page .

我们的视图主要围绕着用适当的数据库支持替换字典数据的虚拟字典进行了更改：列出行、添加行、编辑行和删除行。

额外credit

  1. 为什么要这么做？为什么我不能只打两行，然后就有魔法？

  2. 尝试删除wiki页面的按钮。