タグ: python

  • 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

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