データベース

Starletteは、特定のデータベース実装に厳密にバインドされていません。

GI NOなどの非同期ORMで使用するか、通常の非同期エンドポイントを使用してSQLAlchemyと統合できます。

このドキュメントでは、さまざまなデータベースドライバのSQLAlchemyコアサポートを提供する databases パッケージとの統合方法を示します。

ここでは、テーブル定義、database.Database インスタンスの構成、およびデータベースと対話するいくつかのエンドポイントを含む完全な例を示します。

**. env **

DATABASE_URL=sqlite:///test. db

app.py

contextlibをインポート

データベースのインポート
sqlalchemyをインポート
Starlette. applicationからStarletteをインポート
from starlette. config import設定
JSONResponseをインポートします。
from starlette. routingインポートルート


環境変数または'. env'ファイルからの設定。
config=Config('. env')
DATABASE_URL=config('DATABASE_URL')


データベーステーブルの定義。
Metadata=sqlalchemy. MetaData()

ノート=sqlalchemy. Table(
    "ノート",
    メタデータ,
    sqlalchemy. Column("id",sqlalchemy.Integer,Prime_key=True),
    sqlalchemy. Column("text",sqlalchemy.String),
    sqlalchemy. Column("完了",sqlalchemy.Boolean),
)

データベース=データベース。データベース（DATABASE_URL）

@contextlib. asynccontextmanagerについて
async def lifespan(app):
    awaitデータベース. connect()
    利回り
    awaitデータベースを切断()

#メインアプリケーションコード。
async def list_notes(request):
    クエリ=notes.select()
    result=awaitデータベース. fetch_all(クエリ)
    コンテンツ=[
        {
            "text": result["text"],
            "完了": result["完了"]
        }
        結果のために
    ]
    return JSONResponse(コンテンツ)

async def add_note(request):
    data=await request. json()
    query=notes. insert().values(
       text=data["text"],
       完了=data["完了"]
    )
    データベースを待機します。
    return JSONResponse({
        "text": data["text"],
        "完了":データ["完了"]
    })

ルート=[
    Route("/notes",エンドポイント=list_notes,メソッド=["GET"]),
    Route("/notes",エンドポイント=add_note,メソッド=["POST"]),
]

アプリ=Starlette(
    ルート=ルート、
    lifespan=寿命,
)

最後に、データベーステーブルを作成する必要があります。Alembicをお勧めします。これについては、移行で簡単に説明しました。

クエリ

クエリはSQLAlchemy Coreクエリを使用して実行できます。

次の方法がサポートされています。

rows = await database.fetch_all(query)
row = await database.fetch_one(query)
async for row in database.iterate(query)
await database.execute(query)
await database.execute_many(query)

トランザクション;ビジネス;トランザクション（この単語にはさまざまな意味があり、特定の翻訳はコンテキストによって決定する必要があります）。与えられた単語「Transactions」だけを直訳すると、「取引;事務」と訳すことができます。ここでは、特定の文脈に応じて選択できるように、いくつかの一般的な解釈を提供します。タイトルの要求に応じて、直接出力の翻訳は「取引」です

データベーストランザクションは、デコレータ、コンテキストマネージャ、または低レベルのAPIとして使用できます。

エンドポイントでデコレータを使用する:

@データベーストランザクション()
async def popate_note(request):
    #このデータベースの挿入は、トランザクション内で発生します。
    それは `RuntimeError` によってロールバックされます。
    query=notes. insert().values(text="あなたは私を見ません",完了=True)
    データベースを待機します。
    RuntimeError()を上げる

コンテキストマネージャの使用:

async def popate_note(request):
    データベースと非同期のトランザクション():
        #このデータベースの挿入は、トランザクション内で発生します。
        それは `RuntimeError` によってロールバックされます。
        query=notes. insert().values(text="あなたは私を見ません",完了=True)
        awaitリクエスト。データベース。クエリを実行します。
        RuntimeError()を上げる

低レベルAPIの使用:

async def popate_note(request):
    トランザクション=awaitデータベース。トランザクション（）
    トライ:
        #このデータベースの挿入は、トランザクション内で発生します。
        それは `RuntimeError` によってロールバックされます。
        query=notes. insert().values(text="あなたは私を見ません",完了=True)
        データベースを待機します。
        RuntimeError()を上げる
    除く:
        awaitトランザクション. rollback()
        レイズ
    その他:
        awaitトランザクション. commit()

テスト分離

データベースを使用するサービスをテストする際に、確認したいことがいくつかあります。要件は次のとおりです。

別のデータベースを使用してテストします。
テストを実行するたびに、新しいテストデータベースを作成します。
各テストケース間でデータベースの状態が分離されていることを確認します。

これらの要件を満たすには、次のようにアプリケーションを構築してテストする必要があります。

Starlette. applicationからStarletteをインポート
from starlette. config import設定
データベースのインポート

config=Config(". env")

TESTING=config('TESTING', cast=bool,default=False)
DATABASE_URL=config('DATABASE_URL', cast=データベース.DatabaseURL)
TEST_DATABASE_URL=DATABASE_URL. replace(データベース='test_'+DATABASE_URL.database)

#テスト中に'force_rollback'を使用して、データベースの変更を永続化しないようにします。
各テストケースの間。
テストする場合:
    Database(TEST_DATABASE_URL, force_rollback=True)
その他:
    データベース=データベース。データベース（DATABASE_URL）

テスト実行中も、TESTING をセットアップし、テストデータベースをセットアップする必要があります。py.test を使用していると仮定すると、conftest.py は次のようになります。

pytestをインポート
starlette. configから環境をインポート
starlette. testclientからTestClientをインポート
sqlalchemyからcreate_engineをインポート
sqlalchemy_utilsからデータベース_存在、作成_データベース、ドロップ_データベースをインポート

これは `os.environ` を設定しますが、追加の保護を提供します。
#アプリケーションのインポートの下に配置すると、エラーが発生します。
'TESTING'がすでに環境から読み込まれていることを通知します。
environ['TESTING']='True'

アプリをインポート


@pytest. fix(scope="session",autouse=True)
def create_test_database():
  """
  すべてのテストケースにクリーンなデータベースを作成します。
  安全のために、データベースがすでに存在する場合は中止する必要があります。

  ここでは、一貫して数人のヘルパーに `sqlalchemy_utils` パッケージを使用しています
  データベースの作成と削除。
  """
  url=str(app. TEST_DATABASE_URL)
  エンジン=create_engine(url)
  データベースが存在しない(url)、'テストデータベースはすでに存在します。テストを中止します。'
  create_database(url)#テストデータベースを作成します。
  metadata. create_all(engine)#テーブルを作成します。
  テストを実行します。
  drop_database(url)#テストデータベースをドロップします。


@pytest. fix()
def client():
    """
    テストケースで「クライアント」フィクスチャを使用すると、完全なデータベースが得られます。
    テストケース間のロールバック:

    def test_home page(client):
        url=app. url_path_for('ホームページ')
        response=client. get(URL)
        レスポンスをアサートします。ステータスコード==200
    """
    TestClient（アプリ）をクライアントとして使用する:
        利回りクライアント

移行する

データベースへの増分的な変更を管理するには、ほぼ確実にデータベース移行を使用する必要があります。そのためには、SQLAlchemyの著者が書いたAlembicを強くお勧めします。

$pip alembicをインストール
$alembic in itマイグレーション

次に、Alembicが設定されたDATABASE_URLを参照し、テーブルメタデータを使用するように設定する必要があります。

alembic.ini で次の行を削除します。

sqlalchemy. url=driver://user:pass@localhost/db name

migrations/env.py では、'sqlalchemy.url' 設定キーと target_metadata 変数を設定する必要があります。次のようなものが必要です。

Alembic Configオブジェクト。
config=context. config

DATABASE_URLとテーブル定義を使用するようにAlembicを構成します。。。
アプリをインポート
config. set_main_option('sqlalchemy.url',str(app.DATABASE_URL))
target_metadata=アプリのメタデータ

...

次に、上記のメモの例を使用して、最初のリビジョンを作成します。

alembicリビジョン-m"ノートテーブルの作成"

新しいファイルに必要なコマンドを入力します migrations/versions 内）:

defアップグレード():
    テーブルの作成
      'ノート',
      sqlalchemy. Column("id",sqlalchemy.Integer,Prime_key=True),
      sqlalchemy. Column("text",sqlalchemy.String),
      sqlalchemy. Column("完了",sqlalchemy.Boolean),
    )

def downgrade():
    op. drop_table('notes')

そして、あなたの最初の移行を実行してください。これで私たちのノートアプリが実行可能になりました!

alembicアップグレードヘッド

テスト中に移行を実行します

テストデータベースを作成するたびに、テストスイートがデータベース移行を実行していることを確認するのは良い練習です。これにより、移行スクリプトの問題を発見し、リアルタイムデータベースの状態と一致するデータベースに対してテストが実行されていることを確認するのに役立ちます。

create_test_database フィクスチャを少し調整できます:

alembic importコマンド
from alembic. config import設定
アプリをインポート

...

@pytest. fix(scope="session",autouse=True)
def create_test_database():
    url=str(app. DATABASE_URL)
    エンジン=create_engine(url)
    データベースが存在しない(url)、'テストデータベースはすでに存在します。テストを中止します。'
    create_database(url)#テストデータベースを作成します。
    config=Config("alembic. ini")#マイグレーションを実行します。
    コマンド.アップグレード(config,"head")
    テストを実行します。
    drop_database(url)#テストデータベースをドロップします。