タグ: python

  • Google Cloud API認証の実装方法:YouTubeアップロードAPIを例に

    はじめに

    Google Cloud Platform(GCP)のAPIを利用する際、適切な認証が必要となります。特にOAuth 2.0認証フローを使用する場合、認証トークンの取得と管理は開発における重要なステップです。本記事では、YouTube Upload APIを例に、認証トークンを取得する実装方法について解説します。

    前提条件

    • Python 3.6以上の環境
    • Google Cloud Platformプロジェクトの作成済み
    • OAuth 2.0クライアントIDの取得済み
    • 必要なPythonライブラリのインストール pip install google-auth-oauthlib

    認証トークン取得の実装

    以下のコードは、Google Cloud APIの認証トークンを取得し、JSONファイルに保存する実装例です。

    from google_auth_oauthlib.flow import InstalledAppFlow
    import logging
    import json
    
    # デバッグ用にログレベルを設定
    logging.basicConfig(level=logging.DEBUG)
    
    # 必要なスコープを指定
    SCOPES = ['https://www.googleapis.com/auth/youtube.upload']
    CLIENT_SECRET_FILE = 'client_secret.json'
    
    def get_credentials():
        """
        OAuth 2.0認証フローを実行し、認証情報を取得する関数
        """
        flow = InstalledAppFlow.from_client_secrets_file(
            CLIENT_SECRET_FILE, SCOPES)
        
        # ローカルサーバーを起動して認証フローを実行
        creds = flow.run_local_server(
            port=57062,
            open_browser=True
        )
        
        # 必要な認証情報を辞書形式で返却
        return {
            'token': creds.token,
            'refresh_token': creds.refresh_token,
            'token_uri': creds.token_uri,
            'client_id': creds._client_id,
            'client_secret': creds._client_secret,
            'scopes': creds.scopes,
            'universe_domain': 'googleapis.com',
            'account': '',
            'expiry': creds.expiry.isoformat() + 'Z'  # ISO形式の日時文字列に変換
        }
    
    try:
        # 認証情報を取得
        credentials = get_credentials()
        
        # 認証情報をJSONファイルに保存
        with open('token.json', 'w') as f:
            json.dump(credentials, f, indent=4)
        
        print("認証トークン情報をtoken.jsonに保存しました。")
        
    except Exception as e:
        print(f"エラーが発生しました: {e}")

    コードの解説

    1. ライブラリのインポート

    from google_auth_oauthlib.flow import InstalledAppFlow
    import logging
    import json
    • google_auth_oauthlib:Google認証フローを扱うためのライブラリ
    • logging:デバッグ情報を出力するためのロギング機能
    • json:JSON形式のデータを扱うためのライブラリ

    2. ログレベルの設定

    logging.basicConfig(level=logging.DEBUG)

    開発中はDEBUGレベルでログを出力することで、認証プロセスの詳細な情報を確認できます。本番環境では、必要に応じてログレベルを調整しましょう。

    3. スコープと認証情報の指定

    SCOPES = ['https://www.googleapis.com/auth/youtube.upload']CLIENT_SECRET_FILE = 'client_secret.json'
    • SCOPES:アクセス権限の範囲を指定します。今回はYouTubeへの動画アップロード権限を要求しています。
    • CLIENT_SECRET_FILE:Google Cloud Consoleからダウンロードした認証情報のJSONファイル名を指定します。

    4. 認証情報取得関数

    def get_credentials():

    OAuth 2.0認証フローを実行し、認証情報を取得する関数です。

    4.1 認証フローの初期化

    flow = InstalledAppFlow.from_client_secrets_file(
        CLIENT_SECRET_FILE, SCOPES)

    クライアントシークレットファイルとスコープを指定して認証フローを初期化します。

    4.2 認証フローの実行

    creds = flow.run_local_server(
        port=57062,
        open_browser=True
    )
    • port=57062:ローカルサーバーのポート番号を指定します。
    • open_browser=True:認証フローをブラウザで自動的に開くオプションです。

    4.3 認証情報の整形

    return {
        'token': creds.token,
        'refresh_token': creds.refresh_token,
        'token_uri': creds.token_uri,
        'client_id': creds._client_id,
        'client_secret': creds._client_secret,
        'scopes': creds.scopes,
        'universe_domain': 'googleapis.com',
        'account': '',
        'expiry': creds.expiry.isoformat() + 'Z'  # ISO形式の日時文字列に変換
    }

    取得した認証情報を辞書形式で整形します。特に有効期限(expiry)はISO8601形式の文字列に変換しています。

    5. 認証情報の保存処理

    try:
        # 認証情報を取得
        credentials = get_credentials()
        
        # 認証情報をJSONファイルに保存
        with open('token.json', 'w') as f:
            json.dump(credentials, f, indent=4)
        
        print("認証トークン情報をtoken.jsonに保存しました。")
        
    except Exception as e:
        print(f"エラーが発生しました: {e}")

    認証情報を取得し、token.jsonファイルに保存します。エラーハンドリングも適切に実装されています。

    使用方法

    1. Google Cloud Consoleでプロジェクトを作成し、OAuth 2.0クライアントIDを取得します。
    2. 取得したクライアントID情報をclient_secret.jsonとして保存します。
    3. 上記のスクリプトを実行します。
    4. ブラウザが自動的に開き、Googleアカウントでの認証画面が表示されます。
    5. 認証が完了すると、トークン情報がtoken.jsonに保存されます。

    注意点

    1. セキュリティ: client_secret.jsontoken.jsonには機密情報が含まれるため、バージョン管理システムにコミットしないよう注意しましょう。.gitignoreファイルに追加することをお勧めします。
    2. トークンの有効期限: アクセストークンには有効期限があります。実際のアプリケーションでは、有効期限を確認し、必要に応じてリフレッシュトークンを使用して新しいアクセストークンを取得する処理を実装してください。
    3. 本番環境の考慮: このコードはローカル開発環境向けです。本番環境では、セキュリティを考慮した認証方法(サービスアカウントの利用など)を検討してください。

    まとめ

    本記事では、YouTube Upload APIを例に、Google Cloud APIの認証トークンを取得する方法を解説しました。OAuth 2.0認証フローを理解し、適切に実装することで、セキュアなAPIアクセスが可能になります。認証は一見複雑に思えますが、適切なライブラリを利用することで比較的簡単に実装できます。

    今回紹介したコードは基本的な実装例であり、実際のアプリケーションでは要件に応じてカスタマイズすることをお勧めします。認証情報の安全な管理と、トークンの更新処理などを適切に実装することで、より堅牢なアプリケーションを開発できるでしょう。

  • MSALを使ってOneDriveにユーザー代理でアクセスする

    MSALを使ってOneDriveにユーザー代理でアクセスする

    はじめに

    Microsoft Graph APIを使用すると、Microsoft 365のデータにアクセスできます。この記事では、Microsoft Authentication Library (MSAL) を使用してOneDriveにユーザー代理でアクセスする方法と、遭遇したいくつかの問題の解決方法を紹介します。

    OneDriveへのアクセス

    MSALを用いたアクセストークンの取得

    まず、MSALを使用してアクセストークンを取得します。以下の例では、ユーザーネームとパスワードフローを用いています。

    import msal
    
    def acquire_token_by_username_password():
        client_id = "--クライアントID--"
        tenant = "--テナント名またはID--"
    
        authority_url = f'https://login.microsoftonline.com/{tenant}'
        app = msal.PublicClientApplication(client_id=client_id, authority=authority_url)
        return app.acquire_token_by_username_password(
            username=settings.get('user_credentials', 'username'),
            password=settings.get('user_credentials', 'password'),
            scopes=["https://graph.microsoft.com/.default"])
    
    
    

    GraphClientを使用してOneDriveのデータにアクセス

    トークンを取得したら、GraphClientを使用してOneDriveにアクセスします。

    from office365.graph_client import GraphClient
    
    client = GraphClient(acquire_token_by_username_password)
    drive_items = client.me.drive.shared_with_me().execute_query()
    for item in drive_items:
        print(f"Drive Item url: {item.web_url}")
    
    
    

    遭遇した問題とその解決方法

    認証エラーの解決

    認証時にinvalid_clientというエラーが発生した場合、クライアント秘密キーまたは証明書が正しくない可能性があります。ConfidentialClientApplicationクラスを使用して、証明書を用いた認証を試みます。

    import os
    import msal
    
    cert_path = '/path/to/your/certificate.pem'
    with open(cert_path, 'r') as f:
        private_key = f.read().strip()
    
    authority_url = f'https://login.microsoftonline.com/{tenant}'
    app = msal.ConfidentialClientApplication(
        client_id=client_id,
        client_credential={"private_key": private_key, "thumbprint": thumbprint},
        authority=authority_url)
    
    

    タイプエラーの解決

    TypeError: 'dict' object is not callable というエラーに遭遇した場合、GraphClient コンストラクタがコールバック引数を期待しているため、辞書を直接渡すのではなく、トークンを返す関数を渡す必要があります。

    def acquire_token():
        # トークン取得のコード
        return msal_app.acquire_token_on_behalf_of(user_assertion=id_token, scopes=["https://graph.microsoft.com/.default"])
    
    graph_client = GraphClient(acquire_token)
    

    まとめ

    この記事では、MSALを使用してOneDriveにユーザー代理でアクセスする方法と、遭遇した問題の解決策を紹介しました。Microsoft Graph APIは強力であり、これを使用することでMicrosoft 365のさまざまなサービスと連携できます。

  • FlaskとQueueを用いたタスク管理の実装

    FlaskとQueueを用いたタスク管理の実装

    PythonのFlaskフレームワークは、その軽量さと拡張性から、さまざまなウェブアプリケーションで用いられています。今回はFlaskを使用したシステムで、Pythonの組み込みモジュールであるQueueを利用して、タスク管理を効率的に行う方法について解説します。

    この記事で取り扱う主要なテーマは以下の通りです:

    • PythonのQueueを使ったタスク管理
    • Queueとスレッドを組み合わせた非同期処理

    PythonのQueueを使ったタスク管理

    Pythonのqueue モジュールは、FIFO(First-In, First-Out)方式のキューを実装しており、スレッド間でデータの受け渡しを行う際に特に有用です。以下に具体的なコードを示します。

    import queue
    import threading
    # グローバルにキューを作成
    task_queue = queue.Queue()
    def worker():
    while True:
    task = task_queue.get()
    if task is None:
    break
    # ここでタスクを処理
    task_queue.task_done()
    # ワーカースレッドを作成し、実行
    worker_thread = threading.Thread(target=worker)
    worker_thread.start()
    @app.route('/task', methods=['POST'])
    def add_task():
    task_data = request.get_json()
    task_queue.put(task_data)
    return jsonify({'status': 'queued'})

    このコードでは、新たなタスクがPOSTリクエストとして/taskエンドポイントに送信されると、そのタスクはキューに追加されます。一方、バックグラウンドで実行されているワーカースレッドは、キューからタスクを取り出して処理を行います。

    Queueとスレッドを組み合わせた非同期処理

    Queueとスレッドを組み合わせることで、Flaskアプリケーションでの非同期タスク管理を実現できます。これにより、リクエストを迅速に受け付けつつ、時間のかかる処理をバックグラウンドで並行して実行することができます。

    この記事が皆さんのFlaskとPythonのプロジェクトに役立つことを願っています

  • FlaskとThreadを使った並行処理と排他制御の実装

    FlaskとThreadを使った並行処理と排他制御の実装

    PythonのFlaskフレームワークは、小規模なアプリケーションから大規模なエンタープライズアプリケーションまで幅広い用途で使用されています。今回は、Flaskを用いたウェブアプリケーションで、複数のリクエストが同時に来た場合でも、裏側で動く処理を1つに制限する方法を解説します。

    この記事で取り扱う主要なテーマは以下の通りです:

    • Pythonのスレッドを使った並行処理
    • 排他制御によるリソースへの同時アクセス制限

    スレッドを使った並行処理

    Python の threadingモジュールを用いることで、スレッドを利用した並行処理を実装することができます。以下に具体的なコードを示します。

    import threading
    # スレッドを使う処理は関数にまとめます
    def process_request(request_data):
    # リクエストに対する処理を記述
    pass
    @app.route('/exec', methods=['POST'])
    def exec():
    # POSTリクエストのデータを取得
    request_data = request.get_json()
    # 新たなスレッドを作成して関数を実行
    thread = threading.Thread(target=process_request, args=(request_data,))
    thread.start()
    # レスポンスを即時に返す
    return jsonify({'status': 'processing'})

    このコードでは、POSTリクエストが/execエンドポイントに送信されると、新たなスレッドが作成され、process_request関数がそのスレッド上で実行されます。この関数はリクエストのデータを処理し、その間、メインスレッドは次のリクエストを受け付ける準備をします。

    排他制御による同時アクセス制限

    一方、裏側で動く処理を1つに保つためには排他制御が必要です。PythonのthreadingモジュールのLockを用いることで、排他制御を実現できます。以下に具体的なコードを示します。

    lock = threading.Lock()
    def process_request(request_data):
    with lock:
    # リクエストに対する処理を記述
    pass

    このコードでは、lockオブジェクトを用いてwithステートメントの中の処理が排他的に実行されるようになります。すなわち、一度に一つのスレッドだけがこの処理を実行でき、他のスレッドはロックが解放されるまで待機する必要があります。

    これらの方法を組み合わせることで、Flaskでの並行リクエストの受け付けと裏側での処理の排他制御を実現することができます。これにより、アプリケーションのパフォーマンスと安定性を向上させることが可能になります。

    この記事が皆様のPythonとFlaskのプロジェクトに役立つことを願っています。

  • Office365-REST-Python-Clientを使用したSharePointへの1GBファイルのアップロード方法

    概要

    sharepoint に 大容量(1G)のファイルをアップロードする方法を紹介

     

    CURL コマンドでやると、以下の記事で紹介のあるように認証周りに骨が折れる。

    https://rougeref.hatenablog.com/entry/2020/04/15/000000

    そこで、認証周りを包括しつつ非常に簡単にアップロードできるライブラリを紹介。

    インストール方法

    pip install Office365-REST-Python-Client

     

    サンプルコード

    以下のコードはサンプルとしてユーザー認証をしている。

    アプリケーションを登録して利用することが望ましい。

    sharepoint_url と、UserCredential の部分を適宜修正することでそのまま利用可能。

    from office365.runtime.auth.user_credential import UserCredential
    from office365.sharepoint.client_context import ClientContext
    from office365.sharepoint.files.creation_information import FileCreationInformation
    import os
    
    def get_sharepoint_context_using_user():
    
    sharepoint_url = 'https://xxxxxxxxxx.sharepoint.com/sites/upload-test'
    user_credentials = UserCredential('xxxxxxxxxx@sumito.jp', 'password')
    ctx = ClientContext(sharepoint_url).with_credentials(user_credentials)
    return ctx
    
    def create_sharepoint_directory(dir_name: str):
    """
    Creates a folder in the sharepoint directory.
    """
    if dir_name:
    
    ctx = get_sharepoint_context_using_user()
    
    result = ctx.web.folders.add(f'Shared Documents/{dir_name}').execute_query()
    
    if result:
    # documents is titled as Shared Documents for relative URL in SP
    relative_url = f'Shared Documents/{dir_name}'
    return relative_url
    
    
    target_url = "/sites/upload-test/Shared Documents/test-tsukada-dir-hoge"
    ctx = get_sharepoint_context_using_user()
    target_folder = ctx.web.get_folder_by_server_relative_url(target_url)
    size_chunk = 1000000
    local_path = "1G.csv"
    
    def print_upload_progress(offset):
    file_size = os.path.getsize(local_path)
    print("Uploaded '{0}' bytes from '{1}'...[{2}%]".format(offset, file_size, round(offset / file_size * 100, 2)))
    
    create_sharepoint_directory('test-tsukada-dir-hoge')
    
    with open(local_path, 'rb') as f:
    uploaded_file = target_folder.files.create_upload_session(f, size_chunk,print_upload_progress).execute_query()

     

    動作確認

     

     

     

     

  • Python でjsonの配列の変数から value を抜き出す

    Python で jsonの変数を持たせたいことがあり、読み込む上でのメモ

    import json
    
    list = [{"name":"Worker","no":"123"}, {"name":"User","no":"456"}, {"name":"Organization","no":"789"}, {"name": "Position","no":"123"}, {"name": "PositionTranslation","no":"123"}]
    
    for d in list:
    print(d.get('name'))
    print(d.get('no'))
    print("-----")

     

    list の中に json の配列を格納

    loop の中で .get(key名)を行うことで value を取り出すことができる

    結果

    Worker
    123
    -----
    User
    456
    -----
    Organization
    789
    -----
    Position
    123
    -----
    PositionTranslation
    123
    -----

     

  • AWS lambda(Python 3.8)でimagemagickを使う

    AWS lambda(Python 3.8)でimagemagickを使う

    tl;dr;

    AWS lambda(Python 3.8)で画像処理ツールのimagemagickを動かした。

    install方法からlambda内でimagemagickの動かす際のコードまでを紹介。

    imagemagickとは

    imagemagick は動的に画像を編集する古典的なツール。こちらの記事によると、なんと1987年から開発されている。wikipediaを確認する限り、Photoshopのリリースが1990年だから、その古さに驚かされる。

    しかしこのツール、度重なる脆弱性は報告されているものの、画像編集の機能としてまだまだ現役として使っているプロジェクトも存在する。

    今回はそのimagemagickをlambdaで。Python 3.8のランタイムから使う方法を紹介する。

    Serverless.Pub

    ありがたい事にServerlessを支援し積極的に情報発信をしてくれている団体がいる。

    彼らが作ったlambdaの`binally layer`をCloudFormationでdeployする機能も提供してくれている。

    https://serverless.pub/lambda-utility-layers/

    GitHubはこちら

    https://github.com/serverlesspub/imagemagick-aws-lambda-2

    こちらを`git clone`して、`make all` すればimagemagickのlambdaの`binally layer`ができる。

    その方法を使ってもよいのだが、
    実は以下のページからDeployボタンを押せば、それだけで自動でCloudFormationが自分のAWSで動き、lambdaの中に`binally layer`が作成されることもできる。

    https://serverlessrepo.aws.amazon.com/applications/arn:aws:serverlessrepo:us-east-1:145266761615:applications~image-magick-lambda-layer

    しかし、注意すべきはここの部分。

    要約すると、nodejs10.xランタイムのようなAmazon Linux 2で使えるよ。とのこと。PythonのPの字も見当たらない。
    対象外かと思い当初諦めていた。

    そんな中、最近、`Python 3.8`ランタイムがlambdaで使えるようになった事を知り、公式ドキュメントを読んでみると、今までのPythonで使っていたOSとは異なり、`Amazon linux2`がベースとなっていることがわかった。

    node.jsに絞っているのはAmazonLinux2がベースのように見受けられたため、これはlambdaの中でPythonでもimagemagickを動かせるかもと思い今回試す事にした。

    やってみた

    https://serverlessrepo.aws.amazon.com/applications/arn:aws:serverlessrepo:us-east-1:145266761615:applications~image-magick-lambda-layer

    を開きDeployボタンを押す。
    しかし、この時点でいきなりDeployされるわけではない。

    DeployはCloudFormationによって行われる。CloudFormationのスタック名確定させたら(デフォルトでもよい)デプロイを押下。

    しばらくすると、AWS LambdaのLayersに新たに`image-magick`が登録される。

    Lambda関数から、Layersを選択する

    レイヤーの追加ボタンを押し、imagemagickを選択する。

    レイヤーに表示されない場合は、以下のようにARNを直接入力する。

    これで下準備は完了

    AWS lambda(Python 3.8)でimagemagickを使う

     ここからがPythonからimagemagickを使えるかの確認。

    今回はシンプルにversionを表示させることで使える事にしたい。

    コードはこちら。

    import json
    import subprocess
    
    def lambda_handler(event, context):
        subprocess.check_call(['convert', '-version'])
      
        return {
            'body': json.dumps('Hello from Lambda!')
        }
    

    subprocessを使って、Pythonから、lambdaが動いているOS側のコマンドを実行する。

    imagemagickをinstallすると使えるようになる`convert`コマンドを叩く。

     

    結果は以下の通り

    無事imagemagickのversionを確認できた。

    最後に

    私が実施した際はレイヤーの追加ボタンを押しても選択肢としてimagemagickが表示されなかった。

    これは、CloudFormationによって作られるimagemagickのレイヤーがnode.jsに限定しているため。不便なのでpull reqを出してmergeしてもらった。

    https://github.com/serverlesspub/imagemagick-aws-lambda-2/pull/18/files

    今はレイヤー選択画面から互換性のあるレイヤーとして選択できるはずだ。