データベース

PeeweeのDatabaseオブジェクトは、データベースへの接続を表します。Databaseクラスは、データベースへの接続を開くために必要なすべての情報でインスタンス化され、その後、次の目的で使用できます。

  • 接続を開閉する。

  • クエリを実行する。

  • トランザクション（およびセーブポイント）を管理する。

  • テーブル、列、インデックス、制約を調査する。

Peeweeは、SQLite、MySQL、MariaDB、Postgresをサポートしています。各データベースクラスは、基本的なデータベース固有の設定オプションを提供します。

from peewee import *

# SQLite database using WAL journal mode and 64MB cache.
sqlite_db = SqliteDatabase('/path/to/app.db', pragmas={
    'journal_mode': 'wal',
    'cache_size': -1024 * 64})

# Connect to a MySQL database on network.
mysql_db = MySQLDatabase('my_app', user='app', password='db_password',
                         host='10.1.0.8', port=3306)

# Connect to a Postgres database.
pg_db = PostgresqlDatabase('my_app', user='postgres', password='secret',
                           host='10.1.0.9', port=5432)

Peeweeは、データベース固有の拡張モジュールを介して、SQLite、Postgres、CockroachDBの高度なサポートを提供します。拡張機能を使用するには、適切なデータベース固有のモジュールをインポートし、提供されるデータベースクラスを使用してください。

from playhouse.sqlite_ext import SqliteExtDatabase

# Use SQLite (will register a REGEXP function and set busy timeout to 3s).
db = SqliteExtDatabase('/path/to/app.db', regexp_function=True, timeout=3,
                       pragmas={'journal_mode': 'wal'})


from playhouse.postgres_ext import PostgresqlExtDatabase

# Use Postgres (and register hstore extension).
db = PostgresqlExtDatabase('my_app', user='postgres', register_hstore=True)


from playhouse.cockroachdb import CockroachDatabase

# Use CockroachDB.
db = CockroachDatabase('my_app', user='root', port=26257, host='10.1.0.8')

# CockroachDB connections may require a number of parameters, which can
# alternatively be specified using a connection-string.
db = CockroachDatabase('postgresql://...')

データベース拡張機能の詳細については、以下を参照してください。

データベースの初期化

Database初期化メソッドは、最初のパラメーターとしてデータベースの名前を予期します。後続のキーワード引数は、接続を確立するときに基になるデータベースドライバーに渡され、ベンダー固有のパラメーターを簡単に渡すことができます。

たとえば、Postgresqlでは、接続を作成するときにhost、user、passwordを指定する必要があるのが一般的です。これらは標準のPeeweeDatabaseパラメーターではないため、接続を作成するときにpsycopg2に直接渡されます。

db = PostgresqlDatabase(
    'database_name',  # Required by Peewee.
    user='postgres',  # Will be passed directly to psycopg2.
    password='secret',  # Ditto.
    host='db.mysite.com')  # Ditto.

別の例として、pymysqlドライバーは、標準のPeeweeDatabaseパラメーターではないcharsetパラメーターを受け入れます。この値を設定するには、他の値と一緒にcharsetを渡すだけです。

db = MySQLDatabase('database_name', user='www-data', charset='utf8mb4')

使用可能なパラメーターについては、データベースドライバーのドキュメントを参照してください。

Postgresqlの使用

Postgresqlデータベースに接続するには、PostgresqlDatabaseを使用します。最初のパラメーターは常にデータベースの名前であり、その後に任意のpsycopg2パラメーターを指定できます。

psql_db = PostgresqlDatabase('my_database', user='postgres')

class BaseModel(Model):
    """A base model that will use our Postgresql database"""
    class Meta:
        database = psql_db

class User(BaseModel):
    username = CharField()

Playhouse、Peeweeの拡張機能には、次のような多くのpostgres固有の機能を提供するPostgresql拡張モジュールが含まれています。

これらの優れた機能を使用する場合は、playhouse.postgres_extモジュールのPostgresqlExtDatabaseを使用してください

from playhouse.postgres_ext import PostgresqlExtDatabase

psql_db = PostgresqlExtDatabase('my_database', user='postgres')

分離レベル

Peewee 3.9.7以降、分離レベルは、psycopg2.extensionsのシンボリック定数を使用して、初期化パラメーターとして指定できます。

from psycopg2.extensions import ISOLATION_LEVEL_SERIALIZABLE

db = PostgresqlDatabase('my_app', user='postgres', host='db-host',
                        isolation_level=ISOLATION_LEVEL_SERIALIZABLE)

注意

古いバージョンでは、基になるpsycopg2接続で分離レベルを手動で設定できます。これは、1回限りの方法で実行できます。

db = PostgresqlDatabase(...)
conn = db.connection()  # returns current connection.

from psycopg2.extensions import ISOLATION_LEVEL_SERIALIZABLE
conn.set_isolation_level(ISOLATION_LEVEL_SERIALIZABLE)

接続が作成されるたびにこれを実行するには、サブクラス化して、この目的のために設計された_initialize_database()フックを実装します。

class SerializedPostgresqlDatabase(PostgresqlDatabase):
    def _initialize_connection(self, conn):
        conn.set_isolation_level(ISOLATION_LEVEL_SERIALIZABLE)

CockroachDBの使用

playhouse.cockroachdbで定義されているCockroachDatabaseデータベースクラスを使用して、CockroachDB（CRDB）に接続します。

from playhouse.cockroachdb import CockroachDatabase

db = CockroachDatabase('my_app', user='root', port=26257, host='localhost')

Cockroach Cloudを使用している場合は、接続文字列を使用して接続パラメーターを指定する方が簡単な場合があります。

db = CockroachDatabase('postgresql://root:secret@host:26257/defaultdb...')

注意

CockroachDBには、psycopg2（postgres）Pythonドライバーが必要です。

注意

CockroachDBのインストールと入門ガイドは、こちらにあります：https://www.cockroachlabs.com/docs/stable/install-cockroachdb.html

CRDBは、特別なCockroachDatabase.run_transaction()ヘルパーメソッドを使用して利用できるクライアント側のトランザクション再試行を提供します。このメソッドは、再試行する必要がある可能性のあるトランザクションステートメントの実行を担当する呼び出し可能オブジェクトを受け入れます。

run_transaction()の最も簡単な例

def create_user(email):
    # Callable that accepts a single argument (the database instance) and
    # which is responsible for executing the transactional SQL.
    def callback(db_ref):
        return User.create(email=email)

    return db.run_transaction(callback, max_attempts=10)

huey = create_user('huey@example.com')

注意

指定された試行回数後にトランザクションをコミットできない場合、cockroachdb.ExceededMaxAttempts例外が発生します。SQLが不正な形式である場合、制約に違反している場合などは、関数は呼び出し元に例外を発生させます。

詳細については、以下を参照してください。

SQLiteの使用

SQLiteデータベースに接続するには、SqliteDatabaseを使用します。最初のパラメーターは、データベースを含むファイル名、またはインメモリデータベースを作成するための文字列':memory:'です。データベースファイル名の後には、プラグマのリストまたはその他の任意のsqlite3パラメーターを指定できます。

sqlite_db = SqliteDatabase('my_app.db', pragmas={'journal_mode': 'wal'})

class BaseModel(Model):
    """A base model that will use our Sqlite database."""
    class Meta:
        database = sqlite_db

class User(BaseModel):
    username = TextField()
    # etc, etc

Peeweeには、SQLite拡張モジュールが含まれており、全文検索、json拡張機能のサポートなど、多くのSQLite固有の機能を提供します。これらの優れた機能を使用する場合は、playhouse.sqlite_extモジュールのSqliteExtDatabaseを使用してください。

from playhouse.sqlite_ext import SqliteExtDatabase

sqlite_db = SqliteExtDatabase('my_app.db', pragmas={
    'journal_mode': 'wal',  # WAL-mode.
    'cache_size': -64 * 1000,  # 64MB cache.
    'synchronous': 0})  # Let the OS manage syncing.

PRAGMAステートメント

SQLiteでは、PRAGMAステートメント（SQLiteドキュメント）を介して、多くのパラメーターを実行時に構成できます。これらのステートメントは、通常、新しいデータベース接続が作成されたときに実行されます。新しい接続に対して1つ以上のPRAGMAステートメントを実行するには、プラグマ名と値を含む辞書または2タプルのリストとして指定できます。

db = SqliteDatabase('my_app.db', pragmas={
    'journal_mode': 'wal',
    'cache_size': 10000,  # 10000 pages, or ~40MB
    'foreign_keys': 1,  # Enforce foreign-key constraints
})

PRAGMAは、pragma()メソッドまたはSqliteDatabaseオブジェクトで公開されている特別なプロパティを使用して、動的に構成することもできます。

# Set cache size to 64MB for *current connection*.
db.pragma('cache_size', -1024 * 64)

# Same as above.
db.cache_size = -1024 * 64

# Read the value of several pragmas:
print('cache_size:', db.cache_size)
print('foreign_keys:', db.foreign_keys)
print('journal_mode:', db.journal_mode)
print('page_size:', db.page_size)

# Set foreign_keys pragma on current connection *AND* on all
# connections opened subsequently.
db.pragma('foreign_keys', 1, permanent=True)

注意

pragma()メソッドを使用して設定されたプラグマは、デフォルトでは、接続が閉じられた後は保持されません。接続が開かれるたびにプラグマが実行されるように構成するには、permanent=Trueを指定します。

注意

PRAGMA設定の完全なリスト、その意味、および受け入れられる値については、SQLiteドキュメント（http://sqlite.org/pragma.html）を参照してください。

ユーザー定義関数

SQLiteはユーザー定義のPythonコードで拡張できます。SqliteDatabaseクラスは、3種類のユーザー定義拡張機能をサポートしています。

  • 関数 - 任意の数のパラメータを受け取り、単一の値を返します。

  • 集計 - 複数の行からのパラメータを集計し、単一の値を返します。

  • 照合 - ある値のソート方法を記述します。

注意

さらに多くの拡張機能のサポートについては、playhouse.sqlite_extモジュールにあるSqliteExtDatabaseを参照してください。

ユーザー定義関数の例

db = SqliteDatabase('analytics.db')

from urllib.parse import urlparse

@db.func('hostname')
def hostname(url):
    if url is not None:
        return urlparse(url).netloc

# Call this function in our code:
# The following finds the most common hostnames of referrers by count:
query = (PageView
         .select(fn.hostname(PageView.referrer), fn.COUNT(PageView.id))
         .group_by(fn.hostname(PageView.referrer))
         .order_by(fn.COUNT(PageView.id).desc()))

ユーザー定義集計の例

from hashlib import md5

@db.aggregate('md5')
class MD5Checksum(object):
    def __init__(self):
        self.checksum = md5()

    def step(self, value):
        self.checksum.update(value.encode('utf-8'))

    def finalize(self):
        return self.checksum.hexdigest()

# Usage:
# The following computes an aggregate MD5 checksum for files broken
# up into chunks and stored in the database.
query = (FileChunk
         .select(FileChunk.filename, fn.MD5(FileChunk.data))
         .group_by(FileChunk.filename)
         .order_by(FileChunk.filename, FileChunk.sequence))

照合の例

@db.collation('ireverse')
def collate_reverse(s1, s2):
    # Case-insensitive reverse.
    s1, s2 = s1.lower(), s2.lower()
    return (s1 < s2) - (s1 > s2)  # Equivalent to -cmp(s1, s2)

# To use this collation to sort books in reverse order...
Book.select().order_by(collate_reverse.collation(Book.title))

# Or...
Book.select().order_by(Book.title.asc(collation='reverse'))

ユーザー定義テーブル値関数の例（詳細については、TableFunctionおよびtable_functionを参照してください）

from playhouse.sqlite_ext import TableFunction

db = SqliteDatabase('my_app.db')

@db.table_function('series')
class Series(TableFunction):
    columns = ['value']
    params = ['start', 'stop', 'step']

    def initialize(self, start=0, stop=None, step=1):
        """
        Table-functions declare an initialize() method, which is
        called with whatever arguments the user has called the
        function with.
        """
        self.start = self.current = start
        self.stop = stop or float('Inf')
        self.step = step

    def iterate(self, idx):
        """
        Iterate is called repeatedly by the SQLite database engine
        until the required number of rows has been read **or** the
        function raises a `StopIteration` signalling no more rows
        are available.
        """
        if self.current > self.stop:
            raise StopIteration

        ret, self.current = self.current, self.current + self.step
        return (ret,)

# Usage:
cursor = db.execute_sql('SELECT * FROM series(?, ?, ?)', (0, 5, 2))
for value, in cursor:
    print(value)

# Prints:
# 0
# 2
# 4

詳細については、以下を参照してください。

トランザクションのロックモードを設定する

SQLiteトランザクションは、3つの異なるモードで開くことができます。

  • Deferred（デフォルト） - 読み取りまたは書き込みが実行された場合にのみロックを取得します。最初の読み取りは共有ロックを作成し、最初の書き込みは予約ロックを作成します。ロックの取得は実際に必要になるまで延期されるため、別のスレッドまたはプロセスが別のトランザクションを作成し、現在のスレッドでBEGINが実行された後にデータベースに書き込む可能性があります。

  • Immediate - 予約ロックが即座に取得されます。このモードでは、他のデータベースがデータベースに書き込んだり、immediateまたはexclusiveトランザクションを開いたりすることはできません。ただし、他のプロセスは引き続きデータベースから読み取ることができます。

  • Exclusive - 排他ロックを開き、トランザクションが完了するまですべての（読み取り未コミットを除く）接続がデータベースにアクセスするのを防ぎます。

ロックモードを指定する例

db = SqliteDatabase('app.db')

with db.atomic('EXCLUSIVE'):
    do_something()


@db.atomic('IMMEDIATE')
def some_other_function():
    # This function is wrapped in an "IMMEDIATE" transaction.
    do_something_else()

詳細については、SQLiteのロックに関するドキュメントを参照してください。Peeweeでのトランザクションの詳細については、トランザクションの管理に関するドキュメントを参照してください。

APSW、高度なSQLiteドライバー

Peeweeには、apsw、高度なsqliteドライバーを使用する代替SQLiteデータベースも付属しています。APSWの詳細については、APSWプロジェクトWebサイトで入手できます。APSWは、次のような特別な機能を提供します。

  • 仮想テーブル、仮想ファイルシステム、BLOB I/O、バックアップ、およびファイル制御。

  • 接続は、追加のロックなしでスレッド間で共有できます。

  • トランザクションは、コードによって明示的に管理されます。

  • Unicodeは正しく処理されます。

  • APSWは、標準ライブラリのsqlite3モジュールよりも高速です。

  • Pythonアプリケーションに、SQLite C APIのほぼすべてを公開します。

APSWを使用する場合は、apsw_extモジュールのAPSWDatabaseを使用してください。

from playhouse.apsw_ext import APSWDatabase

apsw_db = APSWDatabase('my_app.db')

MariaDBの使用

PeeweeはMariaDBをサポートしています。MariaDBを使用するには、両者で共有されているMySQLバックエンドを使用します。詳細については、「MySQLの使用」を参照してください。

MySQLの使用

MySQLデータベースに接続するには、MySQLDatabaseを使用します。データベース名の後には、ドライバー（例：pymysqlまたはmysqlclient）に渡される任意の接続パラメータを指定できます。

mysql_db = MySQLDatabase('my_database')

class BaseModel(Model):
    """A base model that will use our MySQL database"""
    class Meta:
        database = mysql_db

class User(BaseModel):
    username = CharField()
    # etc, etc

ドライバー情報

  • pymysqlは、純粋なPythonのmysqlクライアントであり、python 2と3で動作します。Peeweeは、最初にpymysqlを使用しようとします。

  • mysqlclientは、c拡張機能を使用し、python 3をサポートしています。MySQLdbモジュールを公開します。Peeweeは、pymysqlがインストールされていない場合、このモジュールを使用しようとします。

  • mysql-pythonはMySQLdb1とも呼ばれ、レガシーであり、使用すべきではありません。これはmysqlclientと同じモジュール名を共有するため、同じことが当てはまります。

  • mysql-connector pythonは、純粋なPython（だと思います??）で、python 3をサポートしています。このドライバーを使用するには、playhouse.mysql_ext拡張機能のMySQLConnectorDatabaseを使用できます。

エラー2006：MySQLサーバーが停止しました

この特定のエラーは、MySQLがアイドル状態のデータベース接続を強制終了したときに発生する可能性があります。これは通常、データベース接続を明示的に管理しないWebアプリで発生します。アプリケーションが起動し、実行される最初のクエリを処理するために接続が開き、その接続が閉じられることがないため、開いたままになり、より多くのクエリを待機します。

これを修正するには、クエリを実行する必要があるときにデータベースに明示的に接続し、完了したら接続を閉じるようにしてください。Webアプリケーションでは、通常、リクエストが来たときに接続を開き、レスポンスを返すときに接続を閉じます。

データベース接続を管理するための一般的なWebフレームワークの設定例については、フレームワークの統合セクションを参照してください。

データベースURLを使用した接続

playhouseモジュールのデータベースURLには、データベースURLを受け取り、Databaseインスタンスを返すヘルパーconnect()関数が用意されています。

コード例

import os

from peewee import *
from playhouse.db_url import connect

# Connect to the database URL defined in the environment, falling
# back to a local Sqlite database if no database URL is specified.
db = connect(os.environ.get('DATABASE') or 'sqlite:///default.db')

class BaseModel(Model):
    class Meta:
        database = db

データベースURLの例

  • sqlite:///my_database.dbは、現在のディレクトリにあるファイルmy_database.dbのSqliteDatabaseインスタンスを作成します。

  • sqlite:///:memory:は、インメモリのSqliteDatabaseインスタンスを作成します。

  • postgresql://postgres:my_password@localhost:5432/my_databaseは、PostgresqlDatabaseインスタンスを作成します。ユーザー名とパスワード、および接続するホストとポートが指定されています。

  • mysql://user:passwd@ip:port/my_dbは、ローカルMySQLデータベースmy_dbのMySQLDatabaseインスタンスを作成します。

  • db_urlドキュメントのその他の例.

実行時のデータベース構成

データベース接続設定が、構成ファイルや環境からロードされるまで、実行時まで不明な場合があります。このような場合、database_nameにNoneを指定して、データベースの初期化を延期できます。

database = PostgresqlDatabase(None)  # Un-initialized database.

class SomeModel(Model):
    class Meta:
        database = database

データベースが初期化されていない間に接続またはクエリを発行しようとすると、例外が発生します。

>>> database.connect()
Exception: Error, database not properly initialized before opening connection

データベースを初期化するには、データベース名と追加のキーワード引数を指定して、init()メソッドを呼び出します。

database_name = input('What is the name of the db? ')
database.init(database_name, host='localhost', user='postgres')

データベースの初期化をさらに細かく制御するには、次のセクション「データベースの動的な定義」を参照してください。

データベースの動的な定義

データベースの定義/初期化方法をさらに詳細に制御するには、DatabaseProxyヘルパーを使用できます。DatabaseProxyオブジェクトはプレースホルダーとして機能し、実行時に別のオブジェクトに置き換えることができます。以下の例では、アプリの構成方法に応じてデータベースを置き換えます。

database_proxy = DatabaseProxy()  # Create a proxy for our db.

class BaseModel(Model):
    class Meta:
        database = database_proxy  # Use proxy for our DB.

class User(BaseModel):
    username = CharField()

# Based on configuration, use a different database.
if app.config['DEBUG']:
    database = SqliteDatabase('local.db')
elif app.config['TESTING']:
    database = SqliteDatabase(':memory:')
else:
    database = PostgresqlDatabase('mega_production_db')

# Configure our proxy to use the db we specified in config.
database_proxy.initialize(database)

警告

実際のデータベースドライバが実行時に変化する場合にのみ、このメソッドを使用してください。たとえば、テスト環境やローカル開発環境では SQLite を使用し、デプロイされたアプリでは PostgreSQL を使用する場合、DatabaseProxy を使用して、実行時にエンジンを切り替えることができます。

ただし、データベースファイルへのパスやデータベースホストなど、実行時に変化するのが接続値のみである場合は、代わりに Database.init() を使用する必要があります。詳細については、実行時データベース設定 を参照してください。

注意

DatabaseProxy の使用を避け、代わりに Database.bind() および関連メソッドを使用してデータベースを設定または変更する方が簡単な場合があります。詳細については、実行時にデータベースを設定する を参照してください。

実行時にデータベースを設定する

Peewee でデータベースを設定する方法は3つあります。

# The usual way:
db = SqliteDatabase('my_app.db', pragmas={'journal_mode': 'wal'})


# Specify the details at run-time:
db = SqliteDatabase(None)
...
db.init(db_filename, pragmas={'journal_mode': 'wal'})


# Or use a placeholder:
db = DatabaseProxy()
...
db.initialize(SqliteDatabase('my_app.db', pragmas={'journal_mode': 'wal'}))

Peewee は、モデルクラスのデータベースを設定または変更することもできます。この手法は、テストを実行する際に、テストモデルクラスをさまざまなデータベースインスタンスにバインドするために、Peewee テストスイートで使用されます。

相補的な2つのメソッドセットがあります。

  • Database.bind() および Model.bind() - 1つまたは複数のモデルをデータベースにバインドします。

  • Database.bind_ctx() および Model.bind_ctx() - これらは bind() と同じですが、コンテキストマネージャーを返し、データベースを一時的にのみ変更する必要がある場合に役立ちます。

例として、データベースを指定せずに2つのモデルを宣言します。

class User(Model):
    username = TextField()

class Tweet(Model):
    user = ForeignKeyField(User, backref='tweets')
    content = TextField()
    timestamp = TimestampField()

実行時にモデルをデータベースにバインドします。

postgres_db = PostgresqlDatabase('my_app', user='postgres')
sqlite_db = SqliteDatabase('my_app.db')

# At this point, the User and Tweet models are NOT bound to any database.

# Let's bind them to the Postgres database:
postgres_db.bind([User, Tweet])

# Now we will temporarily bind them to the sqlite database:
with sqlite_db.bind_ctx([User, Tweet]):
    # User and Tweet are now bound to the sqlite database.
    assert User._meta.database is sqlite_db

# User and Tweet are once again bound to the Postgres database.
assert User._meta.database is postgres_db

Model.bind() および Model.bind_ctx() メソッドは、特定のモデルクラスをバインドする点で同じように機能します。

# Bind the user model to the sqlite db. By default, Peewee will also
# bind any models that are related to User via foreign-key as well.
User.bind(sqlite_db)

assert User._meta.database is sqlite_db
assert Tweet._meta.database is sqlite_db  # Related models bound too.

# Here we will temporarily bind *just* the User model to the postgres db.
with User.bind_ctx(postgres_db, bind_backrefs=False):
    assert User._meta.database is postgres_db
    assert Tweet._meta.database is sqlite_db  # Has not changed.

# And now User is back to being bound to the sqlite_db.
assert User._meta.database is sqlite_db

このドキュメントの Peewee アプリケーションのテスト セクションにも、bind() メソッドの使用例が含まれています。

スレッドセーフと複数のデータベース

マルチスレッドアプリケーションで実行時にデータベースを変更する場合は、モデルのデータベースをスレッドローカルに保存すると、競合状態を防ぐことができます。これは、カスタムモデルの Metadata クラスで実現できます（ThreadSafeDatabaseMetadata を参照してください。これは playhouse.shortcuts に含まれています）。

from peewee import *
from playhouse.shortcuts import ThreadSafeDatabaseMetadata

class BaseModel(Model):
    class Meta:
        # Instruct peewee to use our thread-safe metadata implementation.
        model_metadata_class = ThreadSafeDatabaseMetadata

これで、使い慣れた Database.bind() メソッドまたは Database.bind_ctx() メソッドを使用して、マルチスレッド環境で安全にデータベースを切り替えることができます。

接続管理

データベースへの接続を開くには、Database.connect() メソッドを使用します。

>>> db = SqliteDatabase(':memory:')  # In-memory SQLite database.
>>> db.connect()
True

すでに開いているデータベースに対して connect() を呼び出そうとすると、OperationalError が発生します。

>>> db.connect()
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "/home/charles/pypath/peewee.py", line 2390, in connect
    raise OperationalError('Connection already opened.')
peewee.OperationalError: Connection already opened.

この例外が発生するのを防ぐために、追加の引数 reuse_if_open を指定して connect() を呼び出すことができます。

>>> db.close()  # Close connection.
True
>>> db.connect()
True
>>> db.connect(reuse_if_open=True)
False

データベース接続がすでに開いている場合、connect() の呼び出しは False を返すことに注意してください。

接続を閉じるには、Database.close() メソッドを使用します。

>>> db.close()
True

すでに閉じられている接続に対して close() を呼び出しても例外は発生しませんが、False が返されます。

>>> db.connect()  # Open connection.
True
>>> db.close()  # Close connection.
True
>>> db.close()  # Connection already closed, returns False.
False

データベースが閉じられているかどうかは、Database.is_closed() メソッドを使用してテストできます。

>>> db.is_closed()
True

自動接続の使用

データベースが autoconnect=True (デフォルト) で初期化されている場合、使用前に明示的にデータベースに接続する必要はありません。明示的に接続を管理することは**ベストプラクティス**と見なされるため、autoconnect の動作を無効にすることを検討する場合があります。

接続の有効期間を明示的にすることは非常に役立ちます。たとえば、接続に失敗した場合、クエリが実行される任意の時点ではなく、接続が開かれるときに例外がキャッチされます。さらに、接続プール を使用する場合、接続が適切にリサイクルされるように、connect() と close() を呼び出す必要があります。

正確性を最大限に保証するには、autoconnect を無効にします。

db = PostgresqlDatabase('my_app', user='postgres', autoconnect=False)

スレッドセーフ

Peewee はスレッドローカルストレージを使用して接続状態を追跡するため、Peewee の Database オブジェクトは複数のスレッドで使用しても安全です。各スレッドは独自の接続を持ち、その結果、どのスレッドも、特定の時点で開いている接続は1つだけになります。

コンテキストマネージャー

データベースオブジェクト自体はコンテキストマネージャーとして使用でき、コードのラップされたブロックの期間中に接続を開きます。さらに、トランザクションはラップされたブロックの開始時に開かれ、接続が閉じられる前にコミットされます（エラーが発生した場合、トランザクションはロールバックされます）。

>>> db.is_closed()
True
>>> with db:
...     print(db.is_closed())  # db is open inside context manager.
...
False
>>> db.is_closed()  # db is closed.
True

トランザクションを個別に管理する場合は、Database.connection_context() コンテキストマネージャーを使用できます。

>>> with db.connection_context():
...     # db connection is open.
...     pass
...
>>> db.is_closed()  # db connection is closed.
True

connection_context() メソッドはデコレーターとしても使用できます。

@db.connection_context()
def prepare_database():
    # DB connection will be managed by the decorator, which opens
    # a connection, calls function, and closes upon returning.
    db.create_tables(MODELS)  # Create schema.
    load_fixture_data(db)

DB-API 接続オブジェクト

基になる DB-API 2.0 接続への参照を取得するには、Database.connection() メソッドを使用します。このメソッドは、現在開いている接続オブジェクトが存在する場合はそれを返し、それ以外の場合は新しい接続を開きます。

>>> db.connection()
<sqlite3.Connection object at 0x7f94e9362f10>

接続プーリング

接続プーリングは、pool モジュール によって提供されます。このモジュールは、playhouse 拡張ライブラリに含まれています。プールは以下をサポートします。

  • 接続がリサイクルされるまでのタイムアウト。

  • 開いている接続数の上限。

from playhouse.pool import PooledPostgresqlExtDatabase

db = PooledPostgresqlExtDatabase(
    'my_database',
    max_connections=8,
    stale_timeout=300,
    user='postgres')

class BaseModel(Model):
    class Meta:
        database = db

次のプールされたデータベースクラスが利用可能です。

peewee の接続プールの詳細については、接続プール の playhouse ドキュメントのセクションを参照してください。

Peewee アプリケーションのテスト

Peewee を使用するアプリケーションのテストを作成する場合、テスト用に特別なデータベースを使用することが望ましい場合があります。もう1つの一般的な方法は、クリーンなデータベースに対してテストを実行することです。つまり、各テストの開始時にテーブルが空であることを確認します。

実行時にモデルをデータベースにバインドするには、次のメソッドを使用できます。

  • Database.bind_ctx()。これは、ラップされたブロックの期間中、指定されたモデルをデータベースインスタンスにバインドするコンテキストマネージャーを返します。

  • Model.bind_ctx() は、同様にコンテキストマネージャーを返します。これは、ラップされたブロックの期間中、モデル（およびオプションでその依存関係）を与えられたデータベースにバインドします。

  • Database.bind() は、モデル（およびオプションでその依存関係）を与えられたデータベースにバインドする1回限りの操作です。

  • Model.bind() は、モデル（およびオプションでその依存関係）を与えられたデータベースにバインドする1回限りの操作です。

ユースケースに応じて、これらのオプションのいずれかがより理にかなっている場合があります。以下の例では、Model.bind() を使用します。

テストケースのセットアップ例

# tests.py
import unittest
from my_app.models import EventLog, Relationship, Tweet, User

MODELS = [User, Tweet, EventLog, Relationship]

# use an in-memory SQLite for tests.
test_db = SqliteDatabase(':memory:')

class BaseTestCase(unittest.TestCase):
    def setUp(self):
        # Bind model classes to test db. Since we have a complete list of
        # all models, we do not need to recursively bind dependencies.
        test_db.bind(MODELS, bind_refs=False, bind_backrefs=False)

        test_db.connect()
        test_db.create_tables(MODELS)

    def tearDown(self):
        # Not strictly necessary since SQLite in-memory databases only live
        # for the duration of the connection, and in the next step we close
        # the connection...but a good practice all the same.
        test_db.drop_tables(MODELS)

        # Close connection to db.
        test_db.close()

        # If we wanted, we could re-bind the models to their original
        # database here. But for tests this is probably not necessary.

余談ですが、経験から言うと、互換性の問題を避けるために、本番環境で使用するのと同じデータベースバックエンドを使用してアプリケーションをテストすることをお勧めします。

Peeweeを使用してテストを実行する方法の例をさらに見たい場合は、Peewee自身のテストスイートをご覧ください。

Geventを使用した非同期処理

PostgresqlまたはMySQLで非同期I/Oを行うには、geventをお勧めします。私がgeventを好む理由

  • すべての特殊な「ループ対応」再実装は必要ありません。asyncioを使用するサードパーティライブラリは、通常、コードのレイヤーを再実装したり、プロトコル自体を再実装したりする必要があります。

  • Geventを使用すると、アプリケーションを通常の、クリーンで、慣用的なPythonで記述できます。「async」、「await」、その他のノイズをすべての行に散らかす必要はありません。コールバック、フューチャー、タスク、プロミスもありません。不要なものはありません。

  • GeventはPython 2とPython 3の両方で動作します。

  • GeventはPythonicです。Asyncioは、Pythonらしくない忌まわしいものです。

monkey-patching socketに加えて、pymysqlのような純粋なPythonドライバーでMySQLを使用している場合や、純粋なPythonモードでmysql-connectorを使用している場合は、特別な手順は必要ありません。Cで記述されたMySQLドライバーには、このドキュメントの範囲を超える特別な構成が必要になります。

PostgresとC拡張であるpsycopg2の場合は、次のコードスニペットを使用して、接続を非同期にするイベントフックを登録できます。

from gevent.socket import wait_read, wait_write
from psycopg2 import extensions

# Call this function after monkey-patching socket (etc).
def patch_psycopg2():
    extensions.set_wait_callback(_psycopg2_gevent_callback)

def _psycopg2_gevent_callback(conn, timeout=None):
    while True:
        state = conn.poll()
        if state == extensions.POLL_OK:
            break
        elif state == extensions.POLL_READ:
            wait_read(conn.fileno(), timeout=timeout)
        elif state == extensions.POLL_WRITE:
            wait_write(conn.fileno(), timeout=timeout)
        else:
            raise ValueError('poll() returned unexpected result')

SQLiteはPythonアプリケーション自体に埋め込まれているため、非ブロッキングの候補となるソケット操作は行いません。非同期処理は、SQLiteデータベースにはどちらの方向にも影響しません。

フレームワークの統合

Webアプリケーションの場合、リクエストを受信したときに接続を開き、レスポンスが配信されたときに接続を閉じるのが一般的です。このセクションでは、データベース接続が適切に処理されるように、Webアプリにフックを追加する方法について説明します。

これらの手順により、シンプルなSQLiteデータベースを使用しているか、複数のPostgres接続のプールを使用しているかに関わらず、peeweeが接続を正しく処理することが保証されます。

注意

トラフィック量の多いアプリケーションでは、リクエストごとに接続のセットアップとティアダウンにかかるコストを軽減するために、接続プールを使用するとメリットがある場合があります。

Flask

Flaskとpeeweeは相性が良く、あらゆる規模のプロジェクトで私が頼りにしている組み合わせです。Flaskは、DB接続を開閉するために使用する2つのフックを提供します。リクエストを受信したときに接続を開き、レスポンスが返されたときに接続を閉じます。

from flask import Flask
from peewee import *

database = SqliteDatabase('my_app.db')
app = Flask(__name__)

# This hook ensures that a connection is opened to handle any queries
# generated by the request.
@app.before_request
def _db_connect():
    database.connect()

# This hook ensures that the connection is closed when we've finished
# processing the request.
@app.teardown_request
def _db_close(exc):
    if not database.is_closed():
        database.close()

Django

Djangoでpeeweeを使用することはあまり一般的ではありませんが、実際には2つを非常に簡単に使用できます。Djangoでpeeweeデータベース接続を管理するには、アプリにミドルウェアを追加するのが最も簡単な方法だと私は思います。ミドルウェアは、リクエストが処理されるときに最初に実行され、レスポンスが返されるときに最後に実行されるように、ミドルウェアのリストの最初に置く必要があります。

my_blogという名前のDjangoプロジェクトがあり、peeweeデータベースがmy_blog.dbモジュールで定義されている場合は、次のミドルウェアクラスを追加できます。

# middleware.py
from my_blog.db import database  # Import the peewee database instance.


def PeeweeConnectionMiddleware(get_response):
    def middleware(request):
        database.connect()
        try:
            response = get_response(request)
        finally:
            if not database.is_closed():
                database.close()
        return response
    return middleware


# Older Django < 1.10 middleware.
class PeeweeConnectionMiddleware(object):
    def process_request(self, request):
        database.connect()

    def process_response(self, request, response):
        if not database.is_closed():
            database.close()
        return response

このミドルウェアが実行されるようにするには、settingsモジュールに追加します。

# settings.py
MIDDLEWARE_CLASSES = (
    # Our custom middleware appears first in the list.
    'my_blog.middleware.PeeweeConnectionMiddleware',

    # These are the default Django 1.7 middlewares. Yours may differ,
    # but the important this is that our Peewee middleware comes first.
    'django.middleware.common.CommonMiddleware',
    'django.contrib.sessions.middleware.SessionMiddleware',
    'django.middleware.csrf.CsrfViewMiddleware',
    'django.contrib.auth.middleware.AuthenticationMiddleware',
    'django.contrib.messages.middleware.MessageMiddleware',
)

# ... other Django settings ...

Bottle

私は自分でbottleを使用したことはありませんが、ドキュメントを見ると、次のコードでデータベース接続が適切に管理されるはずだと思います。

# app.py
from bottle import hook  #, route, etc, etc.
from peewee import *

db = SqliteDatabase('my-bottle-app.db')

@hook('before_request')
def _connect_db():
    db.connect()

@hook('after_request')
def _close_db():
    if not db.is_closed():
        db.close()

# Rest of your bottle app goes here.

Web.py

アプリケーションプロセッサのドキュメントを参照してください。

db = SqliteDatabase('my_webpy_app.db')

def connection_processor(handler):
    db.connect()
    try:
        return handler()
    finally:
        if not db.is_closed():
            db.close()

app.add_processor(connection_processor)

Tornado

TornadoのRequestHandlerクラスは、リクエストが処理されるときに接続を開閉するために使用できる2つのフックを実装しているようです。

from tornado.web import RequestHandler

db = SqliteDatabase('my_db.db')

class PeeweeRequestHandler(RequestHandler):
    def prepare(self):
        db.connect()
        return super(PeeweeRequestHandler, self).prepare()

    def on_finish(self):
        if not db.is_closed():
            db.close()
        return super(PeeweeRequestHandler, self).on_finish()

アプリでは、デフォルトのRequestHandlerを拡張する代わりに、PeeweeRequestHandlerを拡張できるようになりました。

これは、Tornadoや別のイベントループでpeeweeを非同期的に使用する方法には対応していないことに注意してください。

Wheezy.web

接続処理コードは、ミドルウェアに配置できます。

def peewee_middleware(request, following):
    db.connect()
    try:
        response = following(request)
    finally:
        if not db.is_closed():
            db.close()
    return response

app = WSGIApplication(middleware=[
    lambda x: peewee_middleware,
    # ... other middlewares ...
])

このコードを送信してくれたGitHubユーザー@tuukkamustonenに感謝します。

Falcon

接続処理コードは、ミドルウェアコンポーネントに配置できます。

import falcon
from peewee import *

database = SqliteDatabase('my_app.db')

class PeeweeConnectionMiddleware(object):
    def process_request(self, req, resp):
        database.connect()

    def process_response(self, req, resp, resource, req_succeeded):
        if not database.is_closed():
            database.close()

application = falcon.API(middleware=[
    PeeweeConnectionMiddleware(),
    # ... other middlewares ...
])

Pyramid

次のように、データベース接続のライフタイムを処理するリクエストファクトリーを設定します。

from pyramid.request import Request

db = SqliteDatabase('pyramidapp.db')

class MyRequest(Request):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        db.connect()
        self.add_finished_callback(self.finish)

    def finish(self, request):
        if not db.is_closed():
            db.close()

アプリケーションのmain()で、MyRequestがrequest_factoryとして使用されていることを確認します。

def main(global_settings, **settings):
    config = Configurator(settings=settings, ...)
    config.set_request_factory(MyRequest)

CherryPy

Publish/Subscribeパターンを参照してください。

def _db_connect():
    db.connect()

def _db_close():
    if not db.is_closed():
        db.close()

cherrypy.engine.subscribe('before_request', _db_connect)
cherrypy.engine.subscribe('after_request', _db_close)

Sanic

Sanicでは、接続処理コードをリクエストとレスポンスのミドルウェアsanicミドルウェアに配置できます。

# app.py
@app.middleware('request')
async def handle_request(request):
    db.connect()

@app.middleware('response')
async def handle_response(request, response):
    if not db.is_closed():
        db.close()

FastAPI

FastAPIはasyncio互換のフレームワークです。Peeweeは、リクエスト全体で接続状態を管理するために、スレッドローカル（これはgeventとも互換性があります）に依存しています。asyncioで使用するには、スレッドローカルの動作をasyncio互換のコンテキストローカルに置き換えるために、いくつかオーバーライドが必要です。

FastAPIでPeeweeを使用することについての完全な説明については、こちらのFastAPIドキュメントを参照してください。

https://fastapi.dokyumento.jp/advanced/sql-databases-peewee/

上記のドキュメントでは、以下について説明しています。

  • asyncioコンテキスト対応の接続状態追跡の追加

  • リクエストごとの接続処理

その他のフレームワーク

ここにフレームワークが見つかりませんか？GitHubチケットを開いてください。セクションを追加するか、さらに良いことに、ドキュメントのプルリクエストを送信してください。

クエリの実行

SQLクエリは通常、クエリビルダーAPIを使用して構築されたクエリでexecute()を呼び出すことによって（またはSelectクエリの場合は単にクエリオブジェクトを反復処理することによって）実行されます。SQLを直接実行したい場合は、Database.execute_sql()メソッドを使用できます。

db = SqliteDatabase('my_app.db')
db.connect()

# Example of executing a simple query and ignoring the results.
db.execute_sql("ATTACH DATABASE ':memory:' AS cache;")

# Example of iterating over the results of a query using the cursor.
cursor = db.execute_sql('SELECT * FROM users WHERE status = ?', (ACTIVE,))
for row in cursor.fetchall():
    # Do something with row, which is a tuple containing column data.
    pass

トランザクションの管理

Peeweeは、トランザクションを扱うためのいくつかのインターフェースを提供します。最も一般的なのは、ネストされたトランザクションもサポートするDatabase.atomic()メソッドです。atomic()ブロックは、ネストのレベルに応じて、トランザクションまたはセーブポイントで実行されます。

ラップされたブロック内で未処理の例外が発生した場合、現在のトランザクション/セーブポイントはロールバックされます。それ以外の場合、ステートメントはラップされたブロックの最後にコミットされます。

例

# Transaction will commit automatically at the end of the "with" block:
with db.atomic() as txn:
    User.create(username='u1')

# Unhandled exceptions will cause transaction to be rolled-back:
with db.atomic() as txn:
    User.create(username='huey')
    # User has been INSERTed into the database but the transaction is not
    # yet committed because we haven't left the scope of the "with" block.

    raise ValueError('uh-oh')
    # This exception is unhandled - the transaction will be rolled-back and
    # the ValueError will be raised.

注意

atomic()コンテキストマネージャーでラップされたブロック内では、Transaction.rollback()またはTransaction.commit()を呼び出すことで、いつでも明示的にロールバックまたはコミットできます。ラップされたコードブロック内でこれを行うと、新しいトランザクションが自動的に開始されます。

with db.atomic() as transaction:  # Opens new transaction.
    try:
        save_some_objects()
    except ErrorSavingData:
        # Because this block of code is wrapped with "atomic", a
        # new transaction will begin automatically after the call
        # to rollback().
        transaction.rollback()
        error_saving = True

    create_report(error_saving=error_saving)
    # Note: no need to call commit. Since this marks the end of the
    # wrapped block of code, the `atomic` context manager will
    # automatically call commit for us.

注意

atomic()は、コンテキストマネージャーまたはデコレーターのいずれかとして使用できます。

注意

Peeweeの動作は、使用している可能性があるDB-API 2.0の動作とは異なります（詳細についてはPEP-249を参照してください）。デフォルトでは、Peeweeはすべての接続を自動コミットモードにし、トランザクション管理はPeeweeによって処理されます。

コンテキストマネージャー

コンテキストマネージャーとしてatomicを使用する

db = SqliteDatabase(':memory:')

with db.atomic() as txn:
    # This is the outer-most level, so this block corresponds to
    # a transaction.
    User.create(username='charlie')

    with db.atomic() as nested_txn:
        # This block corresponds to a savepoint.
        User.create(username='huey')

        # This will roll back the above create() query.
        nested_txn.rollback()

    User.create(username='mickey')

# When the block ends, the transaction is committed (assuming no error
# occurs). At that point there will be two users, "charlie" and "mickey".

atomicメソッドを使用して、取得または作成操作を実行することもできます。

try:
    with db.atomic():
        user = User.create(username=username)
    return 'Success'
except peewee.IntegrityError:
    return 'Failure: %s is already in use.' % username

デコレーター

デコレーターとしてatomicを使用する

@db.atomic()
def create_user(username):
    # This statement will run in a transaction. If the caller is already
    # running in an `atomic` block, then a savepoint will be used instead.
    return User.create(username=username)

create_user('charlie')

トランザクションのネスト

atomic()は、トランザクションの透過的なネストを提供します。atomic()を使用する場合、最も外側の呼び出しはトランザクションでラップされ、ネストされた呼び出しはセーブポイントを使用します。

with db.atomic() as txn:
    perform_operation()

    with db.atomic() as nested_txn:
        perform_another_operation()

Peeweeは、（詳細については、savepoint()を参照してください）セーブポイントを使用して、ネストされたトランザクションをサポートします。

明示的なトランザクション

トランザクション内で明示的にコードを実行したい場合は、transaction()を使用できます。atomic()と同様に、transaction()はコンテキストマネージャーまたはデコレーターとして使用できます。

ラップされたブロック内で例外が発生した場合、トランザクションはロールバックされます。それ以外の場合、ステートメントはラップされたブロックの最後にコミットされます。

db = SqliteDatabase(':memory:')

with db.transaction() as txn:
    # Delete the user and their associated tweets.
    user.delete_instance(recursive=True)

トランザクションは、ラップされたブロック内で明示的にコミットまたはロールバックできます。この場合、新しいトランザクションが開始されます。

with db.transaction() as txn:
    User.create(username='mickey')
    txn.commit()  # Changes are saved and a new transaction begins.
    User.create(username='huey')

    # Roll back. "huey" will not be saved, but since "mickey" was already
    # committed, that row will remain in the database.
    txn.rollback()

with db.transaction() as txn:
    User.create(username='whiskers')
    # Roll back changes, which removes "whiskers".
    txn.rollback()

    # Create a new row for "mr. whiskers" which will be implicitly committed
    # at the end of the `with` block.
    User.create(username='mr. whiskers')

注意

peeweeを使用してtransaction()コンテキストマネージャーでトランザクションをネストしようとすると、最も外側のトランザクションのみが使用されます。ネストされたブロック内で例外が発生した場合、トランザクションはロールバックされません。最も外側のトランザクションにバブルアップする例外のみがロールバックをトリガーします。

これにより予期しない動作につながる可能性があるため、atomic()を使用することをお勧めします。

明示的なセーブポイント

トランザクションを明示的に作成できるのと同様に、savepoint()メソッドを使用して、セーブポイントを明示的に作成することもできます。セーブポイントはトランザクション内で発生する必要がありますが、任意の深さでネストできます。

with db.transaction() as txn:
    with db.savepoint() as sp:
        User.create(username='mickey')

    with db.savepoint() as sp2:
        User.create(username='zaizee')
        sp2.rollback()  # "zaizee" will not be saved, but "mickey" will be.

警告

セーブポイントを手動でコミットまたはロールバックすると、新しいセーブポイントが自動的に作成されることはありません。これは、手動コミット/ロールバック後に新しいトランザクションを自動的に開くtransactionの動作とは異なります。

自動コミットモード

デフォルトでは、Peeweeは自動コミットモードで動作します。したがって、トランザクション外で実行されたステートメントは、独自のトランザクションで実行されます。複数のステートメントを1つのトランザクションにグループ化するために、Peeweeはatomic()コンテキストマネージャー/デコレーターを提供します。これはすべてのユースケースをカバーする必要がありますが、万が一Peeweeのトランザクション管理を一時的に完全に無効にしたい場合は、Database.manual_commit()コンテキストマネージャー/デコレーターを使用できます。

以下は、transaction()コンテキストマネージャーの動作をエミュレートする方法です。

with db.manual_commit():
    db.begin()  # Have to begin transaction explicitly.
    try:
        user.delete_instance(recursive=True)
    except:
        db.rollback()  # Rollback! An error occurred.
        raise
    else:
        try:
            db.commit()  # Commit changes.
        except:
            db.rollback()
            raise

繰り返しますが、これが必要な人はいないと思いますが、念のためここに記載しておきます。

データベースエラー

Python DB-API 2.0仕様では、いくつかの種類の例外が記述されています。ほとんどのデータベースドライバにはこれらの例外の独自の実装があるため、Peeweeは実装固有の例外クラスをラップすることで簡略化しています。これにより、特別な例外クラスをインポートすることを心配する必要がなくなり、peeweeの例外クラスをそのまま使用できます。

  • DatabaseError

  • DataError

  • IntegrityError

  • InterfaceError

  • InternalError

  • NotSupportedError

  • OperationalError

  • ProgrammingError

注意

これらのすべてのエラークラスは、PeeweeExceptionを拡張します。

クエリのロギング

すべてのクエリは、標準ライブラリのloggingモジュールを使用して、peewee名前空間に記録されます。クエリはDEBUGレベルで記録されます。クエリを操作することに興味がある場合は、ハンドラーを登録するだけで済みます。

# Print all queries to stderr.
import logging
logger = logging.getLogger('peewee')
logger.addHandler(logging.StreamHandler())
logger.setLevel(logging.DEBUG)

新しいデータベースドライバの追加

Peeweeには、Postgres、MySQL、MariaDB、およびSQLiteの組み込みサポートが付属しています。これらのデータベースは非常に人気があり、高速で組み込み可能なデータベースから、大規模なデプロイに適したヘビー級サーバーまで、あらゆる範囲をカバーしています。とは言うものの、世の中にはたくさんのクールなデータベースがあり、ドライバがDB-API 2.0仕様をサポートしていれば、選択したデータベースのサポートを追加するのは非常に簡単です。

警告

Peeweeでは、データベース接続を自動コミットモードにする必要があります。

標準ライブラリのsqlite3ドライバ、psycopg2などを使用したことがある場合は、DB-API 2.0仕様に精通している必要があります。Peeweeは現在、いくつかの部分に依存しています。

  • Connection.commit

  • Connection.execute

  • Connection.rollback

  • Cursor.description

  • Cursor.fetchone

これらのメソッドは通常、より高レベルの抽象化にラップされ、Databaseによって公開されるため、ドライバがこれらを正確に実行していなくても、peeweeを最大限に活用できます。例としては、「playhouse」モジュールにあるapsw sqliteドライバがあります。

まず、接続を開き、接続が自動コミットモードになっていることを保証する（したがって、すべてのDB-APIトランザクションセマンティクスを無効にする）Databaseのサブクラスを提供する必要があります。

from peewee import Database
import foodb  # Our fictional DB-API 2.0 driver.


class FooDatabase(Database):
    def _connect(self, database):
        return foodb.connect(self.database, autocommit=True, **self.connect_params)

Databaseは、より高レベルのAPIを提供し、クエリの実行、テーブルとインデックスの作成、およびデータベースを調べてテーブルのリストを取得する役割を担います。上記の実装は、必要な最小限のものであり、一部の機能は動作しません。最適な結果を得るには、データベースからテーブルのテーブルとインデックスのリストを抽出するためのメソッドも追加する必要があります。FooDBはMySQLによく似ており、特別な「SHOW」ステートメントがあるとします。

class FooDatabase(Database):
    def _connect(self):
        return foodb.connect(self.database, autocommit=True, **self.connect_params)

    def get_tables(self):
        res = self.execute('SHOW TABLES;')
        return [r[0] for r in res.fetchall()]

ここに記載されていない、データベースが処理するその他の事項には、次のようなものがあります。

  • last_insert_id()およびrows_affected()

  • paramおよびquote。これらは、SQL生成コードにパラメータープレースホルダーを追加する方法とエンティティ名を引用符で囲む方法を指示します。

  • INTやTEXTなどのデータ型をベンダー固有の型名にマッピングするためのfield_types。

  • 「LIKE/ILIKE」などの操作をデータベースの同等のものにマッピングするためのoperations。

詳細については、Database APIリファレンスまたはソースコードを参照してください。

注意

ドライバがDB-API 2.0仕様に準拠している場合、起動して実行するために必要な作業はほとんどないはずです。

新しいデータベースは、他のデータベースサブクラスと同様に使用できます。

from peewee import *
from foodb_ext import FooDatabase

db = FooDatabase('my_database', user='foo', password='secret')

class BaseModel(Model):
    class Meta:
        database = db

class Blog(BaseModel):
    title = CharField()
    contents = TextField()
    pub_date = DateTimeField()