Google Cloud API 入門 Python編

Jun 25, 2024 · 2 min read · GCP Python ·

こんにちは、クラウドエンジニアのタケウチです。

スクウェア・エニックスに入社するまでは主にAWS環境での作業を業務としており、初めて Google Cloud API について学習する機会があったので、 Compute Engine リソースの基本操作を Python ライブラリを利用して実施する場合についてまとめていきます。

主に Compute Engine リソースの基本操作を元に Python ライブラリを利用して初めて実施する方向けの情報になります。

セットアップ

Python 自体の設定については今回は割愛しますが、私が実施した環境は wsl2 + ubuntu + pyenv + venv です。

またデフォルトの認証情報を設定するために、gcloud コマンドも別途セットアップして利用しました。

公式ドキュメントとしてPython環境のセットアップについても用意されているので、 Python 環境を最初から設定する場合は参考にすることもできます。

アプリケーションのデフォルト認証情報の設定

Google Cloud API を利用する場合にアプリケーションのデフォルトの認証情報を設定します。

今回については公式ドキュメントの通りにローカル開発環境用の認証情報を用意しました。

認証情報作成の gcloud コマンドを実行するとブラウザに遷移するので画面の指示通りに進めるとデフォルトの認証情報が$HOME/.config/gcloud/application_default_credentials.jsonとして作成されました。

ライブラリのインストール

今回については基本的な google-cloud-compute に加えて google-auth をインストールして利用します。

具体的な利用方法については後述しますが google-auth は例外処理に利用します。

また依存関係によって他のライブラリも同時にインストールされ、今回はその中の google-api-core についても例外処理に利用します。

今回必要なライブラリを以下のようにインストールします。

pip install google-cloud-compute google-auth

インスタンスの状態取得、起動、停止

認証情報、ライブラリの用意ができたら早速インスタンスの状態取得、起動、停止について見ていきます。

インスタンスの状態取得

今回は以下のフローで作成しました。

フィルターを紹介するためlistで取得するようにしていますが、 getというメソッドで単一のインスタンスを取得することも可能です。

認証情報を利用してInstancesClientを用意
必要な変数などを定義してListInstancesRequestを用意
用意したクライアントとリクエストを使用してインスタンスのリストを取得
今回はフィルターしているため1インスタンス分だけのリストを取得
インスタンスリストからインスタンス毎の名前と状態をprintする

以下は実装例です。

import logging

from google.api_core.exceptions import GoogleAPICallError
from google.auth.exceptions import DefaultCredentialsError
from google.cloud import compute_v1

def main():
    try:
        # application_default_credentials.jsonファイルが存在しないなど
        # 認証情報の用意ができていない場合はここでエラーが発生する
        instance_client = compute_v1.InstancesClient()
    except DefaultCredentialsError as err:
        logging.error(f'DefaultCredentialsError {err}')
        raise
    # リクエストを作成
    # filterについては filter_str: str ="name=instance_name OR name=instance_name2" のように複数条件の設定も可能
    # 詳しくは公式ドキュメントのListInstancesRequestを参照
    project_id: str = "project_id"
    zone: str = "zone"
    filter_str: str = "name=instance_name"
    instance_list_request = compute_v1.ListInstancesRequest(project=project_id, zone=zone, filter=filter_str)

    try:
        # インスタンスリストを取得
        instance_list = instance_client.list(request=instance_list_request)
    except GoogleAPICallError as err:
        logging.error(f'GoogleAPICallError {err}')
        raise
    # 取得したインスタンスリストからインスタンス名とインスタンスの状態をprintしていく
    for instance in instance_list:
        print(f'name: {instance.name}, status: {instance.status}')

if __name__ == '__main__':
    main()

DefaultCredentialsErrorのexceptを入れることで、デフォルトの認証情報の設定に失敗していた場合も分かりやすくしています。 google-authについてはこのためにインストールしました。

もう一つのexceptであるGoogleAPICallErrorについては正しくないproject_id指定などでエラーとなった場合に分かりやすくするために入れています。

またインスタンスの状態についてはVMインスタンスのライフサイクルに沿った結果が返ってきます。

停止状態の場合は TERMINATED 起動状態の場合は RUNNING というようになります。

API側のドキュメントでもstatusについて確認できます。

上記の実装例を動作させて、取得したインスタンスのステータスが停止状態の場合、出力される内容については以下の通りです。

> python sample.py
name: instance_name, status: TERMINATED

インスタンスの起動、停止

インスタンスの起動、停止については状態取得とも似通った実装例になるので簡易な実装例を記載します。

ここで利用されている operation.result() については簡単に説明すると、起動、停止などの長い時間が必要な処理が完了するまで待つというものになります。

詳細についてはこちらのドキュメントを確認ください。

今回は割愛していますがStartInstanceRequestやStopInstanceRequestというようにstatusの実装例にあったようなRequestを使った実装例も以下の公式ドキュメントにあります。

インスタンスの起動

instance_client = compute_v1.InstancesClient()
zone: str = "zone"
project_id: str = "project_id"
instance_name: str = "instance_name"
# インスタンスの起動
operation = instance_client.start(zone=zone, project=project_id, instance=instance_name)
# インスタンスが起動完了するまで待機
operation.result()
# インスタンスが起動しているか状態を確認
response = instance_client.get(zone=zone, project=project_id, instance=instance_name)
print(f'name: {response.name}, status: {response.status}')

インスタンスの停止

instance_client = compute_v1.InstancesClient()
zone: str = "zone"
project_id: str = "project_id"
instance_name: str = "instance_name"
# インスタンスの停止
operation = instance_client.stop(zone=zone, project=project_id, instance=instance_name)
# インスタンスが停止完了するまで待機
operation.result()
# インスタンスが停止しているか状態を確認
response = instance_client.get(zone=zone, project=project_id, instance=instance_name)
print(f'name: {response.name}, status: {response.status}')