dbos-python

DBOS Python のベストプラクティス

DBOS の耐久性のあるワークフローを使用して、信頼性と耐障害性に優れた Python アプリケーションを構築するためのガイドです。対象は DBOS Python 3.x です。

適用する場面

次の場合に、これらのガイドラインを参照してください。

• 既存の Python コードに DBOS を追加する場合
• ワークフローとステップを作成する場合
• 同時実行制御にキューを使用する場合
• ワークフロー間の通信（イベント、メッセージ、ストリーム）を実装する場合
• DBOS アプリケーションを設定して起動する場合
• 外部アプリケーションから DBOSClient を使用する場合
• DBOS アプリケーションをテストする場合
• DBOS Python 2.x のコードを 3.x にアップグレードする場合

優先度別のルールカテゴリ

優先度	カテゴリ	影響	接頭辞
1	ライフサイクル	重大	lifecycle-
2	ワークフロー	重大	workflow-
3	ステップ	高	step-
4	キュー	高	queue-
5	通信	中	comm-
6	パターン	中	pattern-
7	テスト	低〜中	test-
8	クライアント	中	client-
9	高度な機能	低	advanced-

重要なルール

DBOS の設定と起動

DBOS アプリケーションでは、メイン関数内で DBOS を設定して起動する必要があります。

import os
from dbos import DBOS, DBOSConfig

@DBOS.workflow()
def my_workflow():
    pass

if __name__ == "__main__":
    config: DBOSConfig = {
        "name": "my-app",
        "application_version": "0.1.0",
        "system_database_url": os.environ.get("DBOS_SYSTEM_DATABASE_URL"),
    }
    DBOS(config=config)
    DBOS.launch()

新しいアプリケーションを作成する場合は、application_version を "0.1.0" に設定します。省略した場合、DBOS はワークフローのソースコードから不透明なハッシュ値を生成します。既存のアプリケーションを編集する場合は、設定済みのバージョンを変更しないでください。変更はデプロイ上の判断事項です（references/advanced-versioning.md を参照）。

ワークフローとステップの構成

ワークフローはステップで構成されます。複雑な処理を行う関数や、外部サービスにアクセスする関数は、すべてステップにする必要があります。

@DBOS.step()
def call_external_api():
    return requests.get("https://api.example.com").json()

@DBOS.workflow()
def my_workflow():
    result = call_external_api()
    return result

主な制約

• ステップから DBOS.start_workflow または DBOS.recv を呼び出さないでください
• ワークフローを開始するためにスレッドを使用しないでください。DBOS.start_workflow またはキューを使用してください
• ワークフローは決定論的である必要があります。非決定的な処理はステップ内で行ってください
• ワークフローまたはステップからグローバル変数を作成または更新しないでください
• async def のコードでは、DBOS メソッドの _async 版を使用してください（await DBOS.start_workflow_async(...)、send_async、recv_async、sleep_async、register_queue_async など）。同期 DBOS メソッドの多く（DBOS.sleep、DBOS.recv、DBOS.send、DBOS.set_event、DBOS.get_event、DBOS.register_queue など）は、イベントループの実行中に呼び出すと RuntimeError を発生させます
• キューの登録とスケジュールの作成は DBOS.launch() の後に行い、データソースはその前に作成してください
• DBOS Conductor への接続（conductor_key の使用、または DBOS Cloud へのデプロイ）には、pip install "dbos[enterprise]" が必要です

DBOS 3.0 で削除された機能（生成しないでください）

@DBOS.transaction / DBOS.sql_session / application_database_url（データソースを使用してください）、Queue(...)（DBOS.register_queue を使用してください）、partition_queue / priority_enabled、@DBOS.scheduled（DBOS.apply_schedules を使用してください）、および DBOS(fastapi=...) / DBOS(flask=...)。既存の 2.x コードを移行する方法については、references/advanced-upgrading-v3.md を参照してください。

使用方法

詳しい説明と例については、各ルールファイルを参照してください。

references/lifecycle-config.md
references/workflow-determinism.md
references/queue-concurrency.md
references/advanced-upgrading-v3.md
references/advanced-shared-database.md

複数のアプリケーションで 1 つのシステムデータベースを共有する場合は、references/advanced-shared-database.md を参照してください。

参考資料

• https://docs.dbos.dev/
• https://github.com/dbos-inc/dbos-transact-py