2

我有一个基于 SQLAlchemy 脚手架的工作 Pyramid Web 应用程序。在我的应用程序中，我有一个函数可以通过 SQLAlchemy 发送电子邮件并在我的数据库中插入/更新表。在 web 上，我通过视图调用这个函数，即 web 上的一个按钮提交给一个视图，并在视图内调用该函数。

我想创建一个调用这个相同函数的控制台脚本，但是是在预定的基础上。我正在研究设置 Pyramid 控制台脚本的示例文档。在一个完美的世界中，我希望能够访问我在 Web 应用程序中使用的所有模型和功能，但能够从控制台使用它们。通过反复试验，我设法包含了一些基础知识以使某些东西正常工作，因为我能够查询我的模型对象之一并将其打印到控制台。我什至可以调用我想要的函数。

但是，在函数内部，它将一行写入数据库并发送一封电子邮件。当我从控制台调用该函数时，它会完成所有工作（至少打印到控制台）并发送电子邮件。它在应该打印的地方打印“插入”语句。但它实际上并没有执行插入或提交它们，我不确定是哪个。我正在从金字塔应用程序的其余部分使用的 models.py 包中导入 DBSession，但是有什么技巧或我需要知道的吗？我尝试声明一个新的 DBSession 并将其自定义到控制台脚本，但这会引发某种“找不到映射器”错误。

在下面的示例中，每条记录都会调用 SendEmail；该函数本身基本上是查找相应的记录，将一行插入另一个模型对象的数据库中，然后发送电子邮件。它作为网络应用程序的一部分非常有用。在这里的控制台端，它打印出它正在做它应该做的所有事情并发送电子邮件，但实际上并没有插入数据库记录。

这是我的控制台脚本：

# describe the script here

import datetime
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
import logging
import optparse
import smtplib
from smtplib import SMTPException
import sys
import textwrap

import pyramid.paster
from pyramid.paster import bootstrap
from pyramid.request import Request

from sqlalchemy.exc import DBAPIError
from sqlalchemy import (
    or_,
    and_,
    not_,
    asc,
    desc,
    func
    )

from functions import SendEmail
from models import DBSession, LogSession, groupfinder

from models import MyObject

from pyramid.session import UnencryptedCookieSessionFactoryConfig

my_session_factory = UnencryptedCookieSessionFactoryConfig('itsaseekreet')

from pyramid.config import Configurator

from sqlalchemy import engine_from_config

from pyramid.authentication import AuthTktAuthenticationPolicy
from pyramid.authorization import ACLAuthorizationPolicy

from zope.sqlalchemy import ZopeTransactionExtension


def main():
    description = """\
        Print the deployment settings for a Pyramid application.  Example:
        'show_settings deployment.ini'
        """
    usage = "usage: %prog config_uri"
    parser = optparse.OptionParser(
        usage=usage,
        description=textwrap.dedent(description)
    )
    parser.add_option(
        '-o', '--omit',
        dest='omit',
        metavar='PREFIX',
        type='string',
        action='append',
        help=("Omit settings which start with PREFIX (you can use this "
              "option multiple times)")
    )
    options, args = parser.parse_args(sys.argv[1:])
    if not len(args) >= 1:
        print('You must provide at least one argument')
        return 2
    config_uri = args[0]
    omit = options.omit
    if omit is None:
        omit = []

    request = Request.blank('/', base_url='http://localhost:13715/')
    env = bootstrap(config_uri, request=request)
    settings = env['registry'].settings
    pyramid.paster.setup_logging(config_uri)

    engine = engine_from_config(settings, 'sqlalchemy.')
    DBSession.configure(bind=engine)
    LogSession.configure(bind=engine)
    authn_policy = AuthTktAuthenticationPolicy(
        'itsaseekreet', callback=groupfinder)
    authz_policy = ACLAuthorizationPolicy()
    config = Configurator(settings=settings,
                          root_factory='myapp.models.RootFactory',
                          session_factory=my_session_factory)
    config.set_authentication_policy(authn_policy)
    config.set_authorization_policy(authz_policy)
    config.add_static_view('static', 'static', cache_max_age=3600)

    log = logging.getLogger(__name__)

    log.info('Starting EmailSender...')

    init_time = datetime.datetime.utcnow()

    log.info('Current datetime (UTC): {0}'.format(str(init_time)))

    items_to_process = DBSession.query(MyObject). \
        filter(and_(MyObject.startdate <= init_time,
                    MyObject.enddate >= init_time,
                    MyObject.manual_send_only == False)).all()


    for item in items_to_process:
        log.info('{0}: runtime: {1}'.format(item.description, item.send_time))
        item_url = request.route_url('itemresponse', responseid='XXXXXX')

        rtn = SendEmail(item.id, item_url)

    env['closer']()


if __name__ == '__main__':
    main()

另外，我现在遇到的另一件事但不是那么重要：我有日志处理程序，我的 log.blah 进入数据库（使用我创建的 LogSession）。这在网络应用程序中也可以正常工作，但在我运行它时没有写入数据库。我不知道这是否是同一个问题，或者在我的配置中，处理程序是否没有为控制台或其他东西设置正确。我不知道，但上面的主要问题是我正在寻找的。谢谢！

编辑：我又戳了一下，发现教程谈论 SQLAlchemy 设置，并正在查看 initializedb.py 脚本，因为它会像我想要的那样修改数据库并连接到模型。我做了import transaction并用

with transaction.manager:
    items_to_process = DBSession.query(MyObject). \
        filter(and_(MyObject.startdate <= init_time,
                    MyObject.enddate >= init_time,
                    MyObject.manual_send_only == False)).all()


    for item in items_to_process:
        log.info('{0}: runtime: {1}'.format(item.description, item.send_time))
        item_url = request.route_url('itemresponse', responseid='XXXXXX')

        rtn = SendEmail(item.id, item_url)

这似乎完全符合我的要求，或者至少实际确实提交并写入数据库。我将不得不更多地使用它，因为我不确定如果被调用函数中存在数据库错误，如果它回滚整个事情，如果它提交它的一部分，或者什么，会发生什么。在函数本身内有异常处理和清理，但我认为它通常取决于pyramid_tm 和Zope 来处理幕后的东西。

4

1 回答 1

0

据我所知，永远不要提交更改。

DBSession.commit()

closer在你的函数之前调用它。它应该根据您的使用方式工作DBSession。

于 2013-10-07T03:39:02.630 回答