2

私は、SQLAlchemy 足場に基づいて、動作する Pyramid Web アプリを持っています。アプリケーション内に、メールを送信し、SQLAlchemy を介してデータベースにテーブルを挿入/更新する機能があります。Web では、ビューを介してこの関数を呼び出します。つまり、Web 上のボタンがビューに送信され、ビュー内で関数が呼び出されます。

これと同じ関数を定期的に呼び出すコンソール スクリプトを作成したいと考えています。私はPyramid コンソール スクリプトをセットアップするためのサンプル ドキュメントに取り組んでいます。完璧な世界では、Web アプリで使用しているすべてのモデルと関数にアクセスできるようにしたいと考えていますが、それらをコンソールから使用できるようにしたいと考えています。試行錯誤の結果、モデル オブジェクトの 1 つをクエリしてコンソールに出力できるという点で、何かを機能させるための基本事項を含めることができました。必要な関数を呼び出すことさえできます。

ただし、関数内ではデータベースに行を書き込み、電子メールを送信します。コンソールから関数を呼び出すと、すべての作業が行われ (少なくともコンソールに出力されます)、メールが送信されます。必要な場所に「INSERT」ステートメントを出力します。しかし、実際にはINSERTを実行しているのかコミットしているのかはわかりません。ピラミッド アプリの残りの部分が使用する models.py パッケージから DBSession をインポートしていますが、知っておく必要があるトリックや何かがありますか? 新しい DBSession を宣言し、コンソール スクリプトに合わせて作成しようとしましたが、何らかの「マッパーが見つかりません」というエラーが発生しました。

以下の例では、SendEmail がレコードごとに呼び出されます。関数自体は基本的に、対応するレコードを検索し、別のモデル オブジェクトのデータベースに行を挿入してから、電子メールを送信します。Web アプリの一部として機能します。ここのコンソール側では、実行すべきことをすべて実行していることが出力され、電子メールが送信されますが、データベース レコードは実際には挿入されません。

これが私のコンソールスクリプトです：

# 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()

また、私が実行しているが、今はそれほど重要ではないもう 1 つのこと: log.blah が (作成した LogSession を使用して) データベースに送られるログ ハンドラーがあります。これも Web アプリからは正常に機能しますが、実行するとデータベースに書き込まれません。それが同じ問題なのか、それとも私の設定でハンドラーがコンソールなどに正しく設定されていないのかはわかりません。わかりませんが、上記の主な問題は私が探しているものです。ありがとう！

EDIT：私はそれをもう少し突っ込んで、SQLAlchemyのセットアップについて話しているチュートリアルを見つけ、 initializeddb.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)

これはまさに私が望むことをしているようです。または、少なくとも実際にはコミットしてデータベースに書き込みます。呼び出された関数内で DB エラーが発生した場合、全体がロールバックされた場合、その一部がコミットされた場合など、何が起こるか正確にはわからないので、もう少し作業する必要があります。関数自体の内部には例外処理とクリーンアップがありますが、通常は裏で処理するために pyramid_tm と Zope に依存していると思います。

4

1 に答える 1

0

私が言える限り、決して変更をコミットしないでください。

DBSession.commit()

関数の前に呼び出しますcloser。使い方に応じて動作するはずですDBSession。

于 2013-10-07T03:39:02.630 に答える