Linear のデータをPythonのpetlでETL処理する方法｜CData Connect AI
CData Connect AI の Python SDK と petl フレームワークを使うと、Linear のデータをローカルにドライバーをインストールすることなく、Pythonで直接抽出・変換・ロードするETLパイプラインを構築できます。パーソナルアクセストークンで接続し、数行のコードで設定完了です。
Python は豊富なモジュールのエコシステムを備えているため、素早く作業に取りかかり、システムをより効果的に連携できます。CData Connect AI Python SDK と petl フレームワークを使えば、Linear のデータを抽出・変換して CSV などに出力する Linear 連携のアプリケーションやパイプラインを構築できます。この記事では、Connect AI に接続し、petl を使ってLinear のデータを抽出・変換し、CSV ファイルに書き出す方法をご紹介します。
Connect AI Python SDK（cdata-connect-ai）は DB-API 2.0（PEP 249）に準拠したクライアントなので、petl は etl.fromdb を使って SDK の接続から直接読み取れます。ソースごとにドライバーをインストールする必要はありません。パーソナルアクセストークンで接続して、パイプラインを構築しましょう。
Python + petl でできること
Linear データの定期抽出パイプライン構築
SQL クエリで取得した Linear のレコードを petl の etl.fromdb で抽出し、etl.sort などの変換処理をはさんでから CSV に書き出す、軽量な ETL パイプラインを構築できます。
複数データソースとの統合分析
DB-API 2.0 準拠のクライアントなので、同じコードパターンで Linear 以外のデータソースにも接続でき、petl 上で結合・整形してから任意の出力先にロードできます。
Linear への書き戻し処理の自動化
書き込みに対応した権限を持つ PAT で接続していれば、cursor.executemany を使ったバッチ INSERT により、変換済みデータを Linear に自動で書き戻す処理を組み込めます。
Connect AI で Linear に接続する方法は？
CData Connect AI では、直感的なクリック操作ベースのインターフェースでデータソースに接続できます。
- Connect AI にログインし、Sources をクリックして、 Add Connection をクリックします
- 「Add Connection」パネルから「Linear」を選択します
-
Linear に接続するために必要な認証プロパティを入力します。
Linear への認証には、個人用 API キーまたは OAuth 2.0 を使用できます。ご自身の Linear アカウントで接続する場合は、API キーを使うのが最も手軽です。
API キーによる認証
以下の接続プロパティを設定します。
- AuthScheme：APIKey に設定します。
- APIKey：Linear の個人用 API キー。
個人用 API キーを作成するには、Linear にログインして Settings > Security & access > Personal API keys を開き、New API key を選択して作成します。キーは一度しか表示されないため、すぐにコピーしておきましょう。
OAuth による認証
OAuth を使用するには、Linear にカスタム OAuth アプリケーションを登録する必要があります（Settings > API > OAuth applications）。登録すると OAuthClientId と OAuthClientSecret が発行されます。対応しているフローは以下の 2 つです。
- Authorization code：AuthScheme を OAuth、InitiateOAuth を GETANDREFRESH に設定し、OAuthClientId、OAuthClientSecret、およびアプリケーションで定義した CallbackURL（例：http://localhost:33333）を指定します。ドライバーがブラウザで Linear を開くので、アクセスを許可してください。
- Client credentials：AuthScheme を OAuthClient に設定し、OAuthClientId と OAuthClientSecret を指定します。アプリケーション自体を認証するためブラウザ操作は不要で、マシン間（M2M）連携に適しています。
デフォルトでは、ドライバーは read,write スコープをリクエストします。アクセストークンは有効期限が切れると自動で更新されます。
- 「Save & Test」をクリックします
- 「Permissions」タブに移動し、ユーザーベースの権限を更新します。
パーソナルアクセストークン（PAT）を生成する
Python SDK は、アカウントのメールアドレスとパーソナルアクセストークン（PAT）を使って Connect AI に認証します。アクセスの粒度を保つために、アプリケーションごとに個別の PAT を作成することをおすすめします。
- Connect AI アプリの右上にある歯車アイコン（）をクリックして、設定ページを開きます。
- 設定ページの Access Tokens セクションに移動し、 Create PAT をクリックします。
- PAT に名前を付けて Create をクリックします。
- パーソナルアクセストークンは作成時にのみ表示されます。必ずコピーして安全な場所に保管してください。
必要なモジュールをインストールする方法は？
pip ユーティリティを使って、SDK と petl フレームワークをインストールします。
pip install cdata-connect-ai pip install petl
Python で Linear のデータの ETL アプリを構築する方法は？
必要なモジュールをインストールしたら、ETL アプリを構築する準備は完了です。以下にコードスニペットを紹介しますが、完全なソースコードは記事の末尾に掲載しています。
まずモジュールをインポートし、アカウントのメールアドレスと PAT で Connect AI に接続します。
import petl as etl
import cdata_connect_ai
conn = cdata_connect_ai.connect(
username="[email protected]",
password="<your_pat>",
)
Linear をクエリする SQL ステートメントを作成する方法は？
SQL を使って、Linear をクエリするステートメントを作成します。この記事では、Team エンティティからデータを読み取ります。識別子は <Connection>.<Schema>.<Table> という3つの部分で構成されており、接続名はデフォルトでソース名（例：Linear1）になります。
sql = (
"SELECT id, name "
"FROM [Linear1].[Linear].[Team] "
"WHERE key = 'ENG'"
)
Linear のデータの抽出・変換・ロードを実装する方法は？
接続とクエリが準備できたら、petl を使ってLinear のデータを抽出・変換・ロードします。この例では、Linear のデータを抽出し、name カラムでデータを並べ替えて、CSV ファイルにロードします。
table1 = etl.fromdb(conn, sql) table2 = etl.sort(table1, 'name') etl.tocsv(table2, 'team_data.csv')
Linear は Connect AI では読み取り専用のソースのため、このパイプラインではLinear のデータの抽出と変換はできますが、行の書き戻しはできません。抽出が完了したら接続を閉じましょう。
conn.close()
CData Connect AI Python SDK を使えば、petl のような ETL パッケージでのデータへの直接アクセスも含め、通常のデータベースと同じようにLinear のデータを扱えます。
関連情報と無料トライアル
これで、CData Connect AI Python SDK を使って、petl でリアルタイムのLinear のデータをパイプライン処理できるようになりました。Linear（および数百種類のその他のデータソース）への接続について詳しくは、Connect AI のページをご覧ください。
CData Connect AI の詳細、または無料トライアルにお申し込みください：
完全なソースコード
import petl as etl
import cdata_connect_ai
conn = cdata_connect_ai.connect(
username="[email protected]",
password="<your_pat>",
)
sql = (
"SELECT id, name "
"FROM [Linear1].[Linear].[Team] "
"WHERE key = 'ENG'"
)
table1 = etl.fromdb(conn, sql)
table2 = etl.sort(table1, 'name')
etl.tocsv(table2, 'team_data.csv')
conn.close()