AWS RedShift

README.md

AWS RedShift

django-redshift-backend

pip install django-redshift-backend

alembic.md

Alembic

psycopg2とSQLAlchemyを使ってRedshiftのデータベーススキーマの修正（マイグレーション）を管理する標準的で強力なORMツールは、PostgreSQLデータベースで広く使われているものと同じです。

それは、Alembic (アレンビック) です。

AlembicはSQLAlchemyの開発チームによって作られた公式なマイグレーションツールであり、SQLAlchemyのセッションやモデル定義と深く連携します。

---

1. Alembic のご紹介

概要

• Alembic (アレンビック) は、SQLAlchemyを基盤とするデータベースのための軽量なマイグレーションツールです。
• Pythonコードでスキーマ変更（DDL: CREATE TABLE, ALTER TABLE など）を定義し、それをバージョン管理することができます。

Redshiftでの利用

RedshiftはPostgreSQL 8.0.2をベースにしているため、psycopg2とSQLAlchemyを通じて接続し、Alembicを使ってマイグレーションを行うことが可能です。

📌 主な機能

1. リビジョンファイルの作成: スキーマ変更を記述するPythonスクリプト（リビジョンファイル）を自動または手動で作成します。
   • オートジェネレート: SQLAlchemyのモデル定義の変更を検知し、自動でupgrade()とdowngrade()のSQLを生成できます。
2. アップグレード / ダウングレード:
   • upgradeコマンドで、リビジョンファイルを適用し、スキーマを新しい状態に更新します。
   • downgradeコマンドで、スキーマを前の状態に戻すことができます。
3. バージョン管理: データベース内にマイグレーション履歴を記録する専用のテーブル（通常alembic_version）を作成し、現在のデータベースの状態を管理します。

構成要素

項目	説明	Redshiftでの注意点
env.py	Alembicの環境設定ファイル。ここでSQLAlchemyエンジン（Redshiftへの接続）を設定します。	接続設定 (psycopg2経由) を正確に行う必要があります。
versions/	マイグレーションスクリプト（リビジョン）を格納するディレクトリ。
script.py.mako	リビジョンファイルを作成する際のテンプレート。

---

2. RedshiftでAlembicを使う上での注意点

Redshiftは分析データベースであり、通常のPostgreSQLとはDDLのサポート範囲や挙動が異なる点に注意が必要です。

1. トランザクションとDDL:

   • 標準のPostgreSQLでは、ALTER TABLEなどのDDL操作をトランザクション内で実行し、エラー時にロールバックできます。
   • Redshiftでは、ほとんどのDDL操作は自動的にコミット（Auto-commit）され、トランザクション内のロールバックができません。
   • このため、Alembicのマイグレーションスクリプトは、失敗した場合に手動で修正が必要になるリスクが高くなります。

2. サポートされていない操作:

   • FOREIGN KEY (外部キー): Redshiftは外部キー制約をサポートしていますが、参照整合性チェックは行いません（ヒントとして使われるのみ）。Alembicが外部キーを生成しても、その動作はPostgreSQLとは異なります。
   • UNIQUE (ユニーク制約): 同様に、ユニーク制約もRedshiftでは強制されません。

3. スキーマの最適化:

   • Redshiftは、テーブルの**DISTKEY (分散キー)** や SORTKEY (ソートキー) の設定がパフォーマンスに極めて重要です。
   • Alembicの標準的なオートジェネレート機能は、これらのRedshift固有の最適化設定に対応していないため、リビジョンファイル内で手動でSQLを記述するか、Redshiftに対応したSQLAlchemy Dialectの拡張機能を利用する必要があります。

---

💡 まとめと次のステップ

Redshiftのスキーマ管理には、psycopg2 + SQLAlchemy + Alembic の組み合わせが最適です。

ただし、Redshiftの特性（DDLのロールバック不可、キー設定の重要性）を理解し、特に重要なテーブル変更の際は、Alembicのオートジェネレートに頼らず、手動で最適化されたSQLを記述することを推奨します。

Alembicの導入方法について、さらに具体的な手順が必要でしたら、お気軽にお申し付けください。

---

Alembicを使用してRedshiftのORM（SQLAlchemy）を管理する際、テーブル作成時に**DISTKEY（分散キー）やSORTKEY**（ソートキー）といったRedshift固有のパラメータをPythonコードから指定するには、主に以下の2つの方法があります。

1. SQLAlchemyの拡張機能/方言 (Dialect) を利用する (推奨される方法)
2. Alembicのマイグレーション内で生のSQLを記述する

最もPythonicで、モデル定義とマイグレーションの連携を保ちやすいのは、1つ目のSQLAlchemyの拡張機能を利用する方法です。

---

1. SQLAlchemyの拡張機能（方言）を利用する方法 (推奨)

Redshift固有のパラメータに対応するためには、通常、既存のPostgreSQL用のSQLAlchemyの方言（Dialect）を拡張したライブラリを使用します。

A. sqlalchemy-redshift の利用

最も一般的に使用されるのは、Redshift特有のパラメータをSQLAlchemyのテーブル定義に追加できるように設計されたサードパーティライブラリ**sqlalchemy-redshift**です。

インストール

pip install sqlalchemy-redshift

モデル定義での記述方法

sqlalchemy-redshiftをインポートし、Tableの引数としてdiststyle、distkey、sortkeyを指定します。

from sqlalchemy import Table, Column, Integer, String, MetaData
from sqlalchemy.ext.declarative import declarative_base

# sqlalchemy-redshift から Redshift 固有の拡張をインポート
# DistStyle、DistKey、SortKey などは、このパッケージを通じて利用可能になります。
# ただし、最新のSQLAlchemyでは直接ColumnやTableの引数として渡すことが多いです。

Base = declarative_base()

class User(Base):
    __tablename__ = 'users'

    # 1. テーブルオプションとしてRedshift固有のパラメータを指定

    # 分散スタイル (DISTSTYLE) を指定
    __table_args__ = (
        # diststyle='KEY' または 'ALL' または 'EVEN' を指定
        {'redshift_diststyle': 'KEY'}, 
        
        # 分散キー (DISTKEY) を指定
        {'redshift_distkey': 'user_id'},
        
        # ソートキー (SORTKEY) を指定（単一キー）
        {'redshift_sortkey': 'created_at'}, 
        
        # 複合ソートキー (COMPOUND SORTKEY) の場合
        # {'redshift_sortkey': ('created_at', 'status')}, 
        
        # インターリーブドソートキー (INTERLEAVED SORTKEY) の場合
        # {'redshift_interleaved_sortkey': ('created_at', 'status')}, 
    )

    id = Column(Integer, primary_key=True)
    user_id = Column(String, index=True) # DISTKEY
    name = Column(String)
    created_at = Column(Integer) # SORTKEY

Alembicでの適用

上記のようにモデル (BaseまたはMetaData) を定義しておけば、Alembicのオートジェネレート機能 (alembic revision --autogenerate) を実行した際に、これらのRedshift固有のパラメータを考慮したDDL（CREATE TABLE文）が生成されます。

---

2. Alembicのマイグレーション内で生のSQLを記述する方法

拡張ライブラリを使わず、既存のPostgreSQL Dialect（psycopg2 + SQLAlchemy）をそのまま使用したい場合は、マイグレーションファイル内で直接SQLを記述します。

この方法は、特にRedshift固有の複雑な処理や、オートジェネレートでは対応できない変更（例：ALTER TABLEの制限を回避する処理）を行う場合に有効です。

Alembicリビジョンファイル (upgrade関数内) での記述方法

from alembic import op
import sqlalchemy as sa

# revision identifiers, used by Alembic.
revision = '...'
down_revision = '...'
branch_labels = None
depends_on = None

def upgrade():
    # 既存のテーブル作成DDL
    op.create_table(
        'sales_data',
        sa.Column('order_id', sa.Integer, primary_key=True),
        sa.Column('region', sa.String),
        sa.Column('sale_date', sa.Date)
        # ここではDISTKEY/SORTKEYは指定しない（通常のSQLAlchemy定義のみ）
    )

    # **生のSQLでDISTKEY/SORTKEYを追加する**
    # Redshiftでは、DISTKEY/SORTKEYはテーブル作成時に指定するのが最も効率的です。
    # ここでは便宜上 op.execute() を使ってテーブル作成後のSQLとして実行する例を示しますが、
    # 理想は op.create_table() の時点で実現するか、手動で op.create_table() を丸ごと生SQLに置き換えることです。

    op.execute(
        """
        -- Alembicの op.create_table() を使わずに、生のCREATE TABLE文全体を実行する例
        CREATE TABLE sales_data_optimized (
            order_id INTEGER NOT NULL,
            region VARCHAR,
            sale_date DATE
        )
        DISTSTYLE KEY
        DISTKEY (region)
        COMPOUND SORTKEY (sale_date, order_id);
        """
    )


def downgrade():
    op.drop_table('sales_data_optimized')
    # op.drop_table('sales_data')

💡 補足：生のSQLでのベストプラクティス

Redshiftでは、ALTER TABLEでDISTKEYやSORTKEYを変更することができないため、実務では以下のパターンを適用することが多いです。

1. 新しいテーブルの作成: CREATE TABLE AS SELECT... や CREATE TABLE (カラム定義...) で新しいテーブルを作成し、DISTKEY/SORTKEYを指定する。
2. データ転送: 既存のテーブルから新テーブルにデータを挿入する (INSERT INTO new_table SELECT * FROM old_table;)。
3. テーブル名の交換: 古いテーブルを削除/リネームし、新しいテーブルを元の名前にリネームする。

Alembic内でこの一連の流れをop.execute()を使って記述することで、Redshiftの制約に対応したマイグレーション管理が可能です。

---

Alembicでも、Djangoのマイグレーションと同様に、Pythonのモデル定義の変更から、必要なマイグレーションスクリプトを自動生成する機能があります。

これはオートジェネレート (Autogenerate) 機能と呼ばれています。

---

Alembicのオートジェネレート機能

Alembicは基本的に手動でスクリプトを作成することもできますが、開発の大部分ではこのオートジェネレート機能を使います。

1. 動作の仕組み

Alembicは、以下の2つの状態を比較することで、必要なスキーマ変更を自動的に判断します。

1. 現在のデータベースの状態: psycopg2とSQLAlchemyを通じてRedshiftに接続し、現在のテーブル、カラムなどのスキーマ情報を取得します（インスペクション）。
2. 現在のPythonモデルの状態: プロジェクト内のSQLAlchemyのBase（またはMetaData）に定義されている、現在のテーブルクラス定義を読み込みます。

この2つの状態を比較し、差分（例: カラムの追加、型の変更など）があれば、それを解消するためのDDL操作（op.add_column(), op.drop_table()など）をPythonコードとして自動的にマイグレーションスクリプトに書き出します。

2. コマンド

オートジェネレートを行う基本的なコマンドは以下の通りです。

alembic revision --autogenerate -m "descriptive message for this change"

このコマンドを実行すると、versions/ディレクトリ内に新しいリビジョンファイルが作成され、そのファイルのupgrade()関数とdowngrade()関数内に、自動で生成されたスキーマ変更の操作コードが記述されます。

3. Redshiftでの注意点 (手動修正の必要性)

オートジェネレート機能は非常に便利ですが、Redshiftのような特殊なデータベースでは、自動生成されたスクリプトをそのまま実行するのではなく、必ず手動で確認・修正することが強く推奨されます。

項目	自動生成の問題点	手動修正が必要な理由
DISTKEY/SORTKEY	単純なオートジェネレートでは、Redshift固有のキー定義は考慮されないことが多い。（特にsqlalchemy-redshiftのような拡張を使っていない場合）	パフォーマンス上極めて重要な情報であり、手動で生のSQL (op.execute()) や拡張機能を用いて追記する必要があります。
ALTER TABLE	PostgreSQLでは可能なDDL（例：ALTER TABLE ... ALTER COLUMN TYPE）が、Redshiftでは実行できない場合があります。	実行できない操作を検知したら、手動で「新しいテーブル作成 $\rightarrow$ データ移行 $\rightarrow$ テーブルリネーム」という回避策のSQLに書き直す必要があります。
データの欠損リスク	カラム名の変更や型の縮小（例: VARCHAR(256)からVARCHAR(10)）など、データ損失やデータ切り詰めが発生する可能性のある変更は、手動で入念にレビューすべきです。

まとめ

• Alembicは自動生成が可能です (alembic revision --autogenerate)。
• ただし、DjangoがOLTP（トランザクション処理）DBを前提としているのに対し、RedshiftはOLAP（分析処理）DBであり、DDLの制約が厳しいです。
• そのため、Alembicを使う場合、自動生成はたたき台として利用し、Redshiftのパフォーマンスと制限を考慮して、DISTKEY/SORTKEYの追加や実行不可能なDDLの代替処理を手動で追記・修正するのが、Redshiftにおける標準的な運用となります。

---

Pydantic と SQLModel を使用して定義したデータモデルから、SQLAlchemyモデルへの変換と、Alembicのオートジェネレート機能を使ったマイグレーションスクリプトの作成は可能です。

これは、SQLModelの設計思想と、AlembicがSQLAlchemyのモデル定義をそのまま利用できる点に基づいています。

---

1. SQLModelの役割

SQLModelは、PydanticとSQLAlchemyを統合するために開発されたライブラリです。

• Pydanticの継承: SQLModelで定義されたクラスは、自動的にPydanticのBaseModelを継承します。これにより、データ検証、型ヒント、スキーマ生成（JSON Schemaなど）の機能が組み込まれます。
• SQLAlchemyの継承: また、SQLModelのクラスは、データベーステーブルを表現するために必要なSQLAlchemyの宣言的ベースクラス（DeclarativeBase）も継承します。

つまり、SQLModelのクラスは、そのまま有効なSQLAlchemyのモデルクラスとして機能するように設計されています。

2. Alembicとの連携

Alembicは、SQLAlchemyのモデル定義が格納されているMetaDataオブジェクトを参照して、オートジェネレートを実行します。

必要な手順

AlembicでSQLModelの定義を利用してオートジェネレートを行うための手順は以下の通りです。

1. SQLModelのクラス定義

モデルを定義します。これは通常のSQLModelの定義と同じです。

from typing import Optional
from sqlmodel import Field, SQLModel, create_engine

# Baseクラスとして機能する
class Hero(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str
    secret_name: str
    age: Optional[int] = None

2. Alembicのenv.pyの設定

Alembicの設定ファイルであるenv.py内で、Alembicがモデル定義を参照できるように、SQLModelのMetaDataを読み込むように設定します。

env.pyの一部（変更が必要な箇所）:

# target_metadata の設定を変更します

from sqlmodel import SQLModel as Base  # <-- SQLModelのBaseをインポート

# ... (中略) ...

# Alembicがスキーマの比較に使用するメタデータオブジェクトを設定
# 通常は SQLAlchemy の Base.metadata ですが、SQLModelでは SQLModel がメタデータを持っています。
target_metadata = Base.metadata

# ... (後略) ...

# run_migrations_online の中も通常通り、データベース接続を設定します。

3. Alembicの実行

上記の設定が完了すれば、以下のコマンドでオートジェネレートが可能です。

alembic revision --autogenerate -m "Add initial Hero table"

これにより、Heroクラスの定義に基づいてop.create_table('hero', ...)を含むマイグレーションスクリプトが自動生成されます。

3. Redshift固有のキーの指定について

先述の通り、RedshiftでDISTKEYやSORTKEYを指定する場合、単純なSQLModelの定義では不足します。

この場合は、以下のいずれかの方法で対応します。

1. sqlalchemy-redshift の利用: SQLModelのクラス定義時に、__table_args__を使ってRedshift固有のパラメータを渡します。（この方法が最もPythonicです。）

   # sqlalchemy-redshift がインストールされている前提
   class Product(SQLModel, table=True):
       __tablename__ = 'products'
       
       # Redshift固有のキー指定を __table_args__ に含める
       __table_args__ = (
           {'redshift_diststyle': 'KEY'}, 
           {'redshift_distkey': 'category_id'},
           {'redshift_sortkey': ('sale_date', 'product_id')},
       )
       
       # ... (カラム定義) ...

2. 自動生成後の手動修正: オートジェネレートで生成されたマイグレーションスクリプト内のop.create_table()呼び出しの直後に、op.execute()を使って生のSQLでキー定義のDDLを追加します。

この仕組みにより、Pydanticの持つシンプルさとSQLAlchemy/Alembicの持つデータベース管理能力を両立させることができます。

CDC.md

CDC

AWS環境で実現するには、主に以下の2つの方法が主流です。

1. AWS DMS (Database Migration Service) の利用 (実績が豊富で柔軟性が高い)
2. Amazon RDS ゼロ ETL 統合 (最もシンプルで最新の選択肢)

それぞれの構成と特徴を詳しく解説します。

---

1. AWS DMS (Database Migration Service) を利用した構成

AWS DMSは、データベース間のデータ移行や継続的なレプリケーション（CDC）を行うための専用サービスです。

構成の概要

1. RDS for MySQLの設定:
   • MySQLの**バイナリログ（Binlog）**を有効にし、フォーマットをROWに設定します。DMSはこれを読み取って変更を追跡します。（CDCの必須設定）
2. AWS DMS コンポーネント:
   • レプリケーションインスタンス: データ移行（レプリケーション）を実行する専用のEC2インスタンスです。ソースとターゲットの間でデータを読み書きし、マッピングや変換を行います。
   • ソースエンドポイント: RDS for MySQLへの接続情報を定義します。
   • ターゲットエンドポイント: Amazon Redshiftへの接続情報を定義します。
   • 移行タスク: CDC（継続的レプリケーション）を定義するコア設定です。どのテーブルを移行するか、フルロード後にCDCを継続するかなどを指定します。
3. データフロー:
   • RDS for MySQLで変更（UPDATE/INSERT/DELETE）が発生すると、その変更がBinlogに記録されます。
   • DMSのレプリケーションインスタンスがBinlogを継続的に読み取ります。
   • DMSは変更データをRedshiftに適した形式に変換し、Redshiftクラスターに書き込みます（通常はS3経由でCOPYコマンドを使用）。

メリット・デメリット

項目	メリット	デメリット
柔軟性	非常に高く、多種多様なデータベースに対応。テーブルやスキーマのフィルタリング、データ変換（トランスフォーメーション）も可能。
コスト		レプリケーションインスタンスの料金が継続的に発生する。
運用		インスタンスの管理（サイズ選定、冗長性など）や、Binlogの保持期間の管理が必要。
安定性	実績が豊富で安定しているが、タスク設定やインスタンスサイズによってはチューニングが必要になる場合がある。

---

2. Amazon RDS ゼロ ETL 統合 (推奨)

これは、2023年以降に登場した新しい機能で、最もシンプルかつ管理負担の少ないCDCの方法です。現時点ではAurora MySQLからRedshiftへの統合が中心ですが、RDS for MySQLへの対応も進んでいます。

構成の概要

1. RDS for MySQL (または Aurora MySQL) の設定:
   • 特別なCDC設定は不要で、通常のBinlogを有効にするだけで済みます。
2. ゼロ ETL 統合の作成:
   • AWSコンソールで、ソースのRDS for MySQL/AuroraとターゲットのRedshiftクラスターを選択し、「統合」を作成します。
3. データフロー:
   • RDS側で発生したデータ変更が、AWSのインフラストラクチャによって自動的にRedshiftにレプリケートされます。
   • ユーザーは、中間でDMSインスタンスを管理したり、タスクを監視したりする必要がありません。

メリット・デメリット

項目	メリット	デメリット
シンプルさ	圧倒的に簡単。DMSのようなレプリケーションインスタンスやタスク設定が不要で、運用管理負担が最小限。
コスト	DMSの専用インスタンス料金が不要（ただし、統合に伴うデータ転送やストレージの料金は発生）。
パフォーマンス	AWSネイティブな連携により、低レイテンシで一貫性の高いレプリケーションが期待できる。
制限		対応リージョンやデータベースバージョン、機能に制限がある場合がある（利用前に公式ドキュメントで確認が必要）。
柔軟性		DMSほど詳細なデータ変換やフィルタリングの柔軟性はない。

どちらを選ぶべきか？

ユースケース	推奨構成	理由
最新の機能と最小限の運用負担を求める場合	2. ゼロ ETL 統合	設定がシンプルで、AWSがスケーリングと管理を担ってくれるため。
特殊なデータ変換・フィルタリングが必要な場合	1. AWS DMS	DMSの「移行タスク」で、スキーマやカラムレベルの詳細な制御が可能なため。
Aurora MySQLを使用している場合	2. ゼロ ETL 統合	Aurora MySQLからの統合は既に広く利用可能で、性能と管理面で優位性があるため。

もし、RDS for MySQLからRedshiftへのゼロ ETL統合が利用可能であれば、まずそちらを検討されることを強く推奨します。利用できない、または特殊な要件がある場合は、AWS DMSが標準的な選択肢となります。

dango-sqlmodel.md

Django モデルクラスから　SQLModelへ

Djangoのモデルクラスの定義を、直接SQLModelのクラス定義に完全に自動で変換する公式のツールや、広く知られたサードパーティ製ツールは、現時点では存在しません。

しかし、この変換作業は、ほとんどの部分を自動化し、手動で微調整を行うという形で実現できます。

変換が必要な理由と課題

項目	Djangoモデル (django.db.models)	SQLModel (sqlmodel)	必要な変換
基底クラス	models.Model	SQLModel, table=True	基底クラスの変更。
カラム定義	models.CharField, models.IntegerField など	Field(...) の型指定と引数	データ型の対応付け（例: CharField(max_length=255) $\rightarrow$ str = Field(max_length=255)）。
主キー	primary_key=True	Optional[int] = Field(default=None, primary_key=True)	Pythonの型ヒントとPydantic/SQLModelのデフォルト値の適用。
リレーション	models.ForeignKey など	Relationship / Link の定義	ORMの抽象度が異なり、手動での書き直しが必要。

推奨される変換アプローチ (半自動化)

完全な自動化が難しいため、「モデルのスケルトン生成」と「手動によるキー・リレーションの修正」を組み合わせるのが最も効率的です。

---

1. モデルのスケルトンを生成する (自動化)

Djangoモデルのソースコードを読み取り、対応するSQLModelのクラスの枠組みを自動生成するカスタムスクリプトを作成します。

💡 簡単なカスタムスクリプトの例（概念）

Djangoモデルのフィールド定義をパースして、SQLModelのフィールドと型ヒントに変換する処理を実装します。

# 例: Djangoの settings.INSTALLED_APPS からモデルをインポートし、
# モデルを動的に読み込む（実際のコードは複雑になるため概念のみ）

# 変換ロジックの概要:
# class MyDjangoModel(models.Model):
#     name = models.CharField(max_length=100)
#     is_active = models.BooleanField(default=True)
#
# ↓ スクリプトが生成する
#
# class MySQLModel(SQLModel, table=True):
#     name: str = Field(max_length=100)
#     is_active: bool = Field(default=True)

この方法で、CharField $\rightarrow$ str, IntegerField $\rightarrow$ int などの単純なフィールドは自動変換できます。

2. Pydanticの自動生成機能の活用 (逆方向のヒント)

Redshiftのマイグレーション管理でも触れたように、SQLAlchemyのツールは「既存のDBスキーマ」からモデル定義を生成する機能を持っています。

sqlacodegen や sqla_model_parser の利用

1. Djangoモデル $\rightarrow$ データベーススキーマ: Djangoのマイグレーションを実行し、MySQL（またはRedshift）にテーブルを先に作成します。
2. データベーススキーマ $\rightarrow$ SQLAlchemyモデル:
   • sqlacodegen などのツールをMySQLデータベースに対して実行し、既存のテーブル定義から生のSQLAlchemyモデルのコードを生成させます。
3. SQLAlchemyモデル $\rightarrow$ SQLModel:
   • 生成されたSQLAlchemyモデルのコードをベースに、SQLModel, table=Trueを継承させ、Pydanticの型ヒントを追加することで、SQLModelの形式に手動で修正します。

この方法は、Djangoモデルの定義を無視し、データベースの実態からモデルを生成するため、最も安全で確実です。

3. CDCの文脈でのベストプラクティス

CDCの文脈では、DjangoのモデルとSQLModelのモデルが厳密に同期している必要があります。

• 単一の情報源 (SSOT): もし可能であれば、**SQLModelの定義をSSOT（Single Source of Truth）**としてください。
  • モデル定義: SQLModelでモデルを定義します。
  • Django連携:
    • Django側では、django-model-utilsのAbstractBaseModelや抽象クラスを使い、SQLModelと同じフィールド定義を手動で再定義します。
    • または、Django側のORMを使わず、生のSQLで対応するなど、Django側の依存度を下げます。
  • CDC/バッチ: SQLModelの定義をそのまま利用し、AlembicでRedshiftのスキーマを管理します。

既存のDjangoプロジェクトが大規模である場合、このアプローチは難しいかもしれませんが、SQLModelがPydanticとSQLAlchemyを統合しているため、今後のCDCバッチや分析基盤の構築においては、SQLModelを起点とすることが最も効率的でバグが少ない方法となります。

MERGE.md

MERGE

MERGE INTO ターゲットテーブル AS T
USING ソーステーブル AS S
ON T.結合キー = S.結合キー
WHEN MATCHED THEN
    UPDATE SET 列1 = S.列1, 列2 = S.列2, ... -- 一致した場合の更新処理
WHEN NOT MATCHED THEN
    INSERT (列1, 列2, ...) VALUES (S.列1, S.列2, ...); -- 一致しない場合の挿入処理

• AWS: マージの例