1 Minute Tips

カテゴリー: python

  • Poetryと依存関係の問題: 問題点、原因、対処方法

    Poetryと依存関係の問題: 問題点、原因、対処方法

    問題点

    Pythonのプロジェクトでよく使用される依存関係管理ツール、Poetryを使用している際に、ビルドプロセスが失敗し、次のエラーメッセージが表示されました。

    Warning: poetry.lock is not consistent with pyproject.toml. You may be getting improper dependencies. Run `poetry lock [--no-update]` to fix it. Because hogehoge depends on psycopg2-binary (2.9.9) which doesn't match any versions, version solving failed.

    このメッセージは、依存関係の不整合と特定のパッケージバージョンが見つからないという二つの問題を示しています。

    原因

    1. poetry.lock と pyproject.toml の不整合

    poetry.lock ファイルは、プロジェクトの依存関係を安定的に保つために使用されます。このファイルが pyproject.toml と整合していない場合、依存関係に関する不整合が発生し、ビルドプロセスに影響を与える可能性があります。

    2. psycopg2-binary のバージョン問題

    psycopg2-binary の指定されたバージョン 2.9.9 が利用可能なバージョンと一致しないため、依存関係の解決に失敗しています。

    対処方法

    1. poetry.lock と pyproject.toml の不整合の修正

    • poetry lock --no-update コマンドを実行して、poetry.lock ファイルを pyproject.toml と整合させます。

    2. psycopg2-binary のバージョン問題の解決

    • pyproject.toml ファイルを確認し、psycopg2-binary のバージョン指定を利用可能なものに更新します。例えば、より柔軟なバージョン指定（^2.9 など）を使用することができます。

    3. 依存関係の再インストール

    • 上記の変更後、poetry install を実行して依存関係を再インストールし、問題が解決されたかを確認します。

    4. テストと確認

    • ローカル環境で変更をテストし、ビルドプロセスが正常に完了することを確認します。

    この記事を通じて、Poetryを使用した依存関係管理の問題とその解決方法についての理解を深めることができます。また、Pythonプロジェクトでのビルドプロセスをよりスムーズに行うための実用的な知識も得られるはずです。

  • 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のプロジェクトに役立つことを願っています

  • Microsoft Authentication Library (MSAL) との連携時に発生する接続エラーとその回避方法

    Microsoft Authentication Library (MSAL) との連携時に発生する接続エラーとその回避方法

    PythonでMSAL (Microsoft Authentication Library) を利用してAzure AD認証を実施する際に、接続エラーに直面することがあります。この記事では、その一般的なエラーと、これを解消するための回避策について詳しく解説します。

    エラーの例

    MSALを使用してAzure ADのトークンを取得する際に、以下のようなエラーメッセージが表示されることがあります。

    requests.exceptions.ConnectionError: HTTPSConnectionPool(host='login.microsoftonline.com', port=443): Max retries exceeded with url: /yourtenant.onmicrosoft.com/v2.0/.well-known/openid-configuration (Caused by NewConnectionError('<urllib3.connection.HTTPSConnection object at 0x7f84b146ad00>: Failed to establish a new connection: [Errno -3] Temporary failure in name resolution'))

    このエラーは、プログラムがMicrosoftの認証サーバーに接続しようとしたときに発生します。

    エラーの原因

    このエラーの原因は主に以下の通りです。

    1. network接続の問題: なんらかの問題で通信がうまくいかなかった

    2. DNSの問題: DNSが一時的に名前解決できない場合があります。これは一時的な問題である場合が多く、しばらく待ってから再試行すると解決することがあります。

    3. サーバーのオーバーロード: Microsoftのサーバーに対するリクエストが多すぎて、サーバーが反応しない場合があります。

    解決策

    上記のエラーに対処としてリトライを実装しました。

    コード例

    以下は、再試行ロジックを実装したPythonのコード例です。

    import time
    import requests
    import msal
    import logging
    logger = logging.getLogger(__name__)
    def acquire_token_with_retries(client_id, private_key, thumbprint, authority_url, max_retries=3, retry_delay=5):
    retries = 0
    while retries < max_retries:
    try:
    client_app = msal.ConfidentialClientApplication(
    client_id=client_id,
    client_credential={"private_key": private_key, "thumbprint": thumbprint},
    authority=authority_url
    )
    return client_app.acquire_token_on_behalf_of(user_assertion="YOUR_USER_ASSERTION", scopes=["https://graph.microsoft.com/.default"])
    except requests.exceptions.ConnectionError as e:
    logger.warning(f"Connection error occurred: {str(e)}. Retrying in {retry_delay} seconds...")
    retries += 1
    time.sleep(retry_delay)
    logger.error(f"Failed to acquire token after {max_retries} attempts.")
    return None

    このコードでは、トークン取得を試みる前に、最大リトライ回数を指定し、それを超えた場合はエラーを記録してNoneを返します。また、リトライの間に指定された秒数だけ遅延させます。

    まとめ

    MSALを使ってAzure AD認証を行う際に、接続エラーが発生する可能性があります。その際は、インターネット接続を確認し、再試行ロジックを実装することで、これらの一時的な問題を解決できる可能性があります。エラー発生時に適切なログ情報を収集することも重要です。

  • テスト実行のトラブルシューティング: Pytest エラー「required field “lineno” missing from alias」を解決する

    テスト実行のトラブルシューティング: Pytest エラー「required field “lineno” missing from alias」を解決する

    概要

    Pythonのテストライブラリであるpytestを使用する際に、特定のエラーが発生することがある。この記事では、「required field “lineno” missing from alias」エラーの背景、原因、および解決方法について深堀りする。

    エラーの背景

    抽象構文木 (Abstract Syntax Tree)

    Pythonのコード解析には、通常、抽象構文木（AST）が使用される。ASTは、プログラムのソースコードを木構造で表現し、これにより構文解析や静的解析が行われる。このエラーはASTの構築時に関連して発生する。

    linenoとは？

    lineno は”line number”の略で、コード内の特定の行を参照する。ASTでは、特定の構造が存在する行番号を追跡するためにlineno情報が使われる。これはデバッグやエラー追跡に不可欠な情報である。

    エラーの原因

    このエラーは、PythonのパーサーまたはASTモジュールが、コードの行番号(lineno)に関する情報を必要としているが、何らかの理由で欠落している場合に発生する。

    主な原因は以下の通り：

    1. 依存ライブラリのバージョン不整合: 特定のライブラリが古いか、または互換性がない可能性がある。
    2. カスタムAST変換の問題: カスタムのAST変換コードが正しくないか、不完全な場合。
    3. Python自体のバグ: 非常にまれだが、Pythonインタープリタ自体の問題である可能性もある。

    解決策

    依存関係の更新

    依存するライブラリ、特にpytestのバージョンが古い場合、最新のバージョンに更新する。

    pip install --upgrade pytest

    プラグインの確認

    pytestのプラグインが原因の場合、プラグインを無効化または更新してみる。

    Pythonバージョンの確認

    使用しているPythonのバージョンが最新でない場合、更新を検討する。また、特定のバージョンでのみ発生するバグである可能性も考慮し、必要に応じてバージョンをダウングレードする。

    まとめ

    pytestで「required field “lineno” missing from alias」エラーに遭遇した場合、依存関係の更新、プラグインの確認、およびPythonバージョンの確認を行うことで解決する可能性がある。これにより、テストの実行がスムーズに行えるようになる。

  • Flaskにおける特定のルートのログ出力の無効化について: AWS EC2/ECS

    Flaskにおける特定のルートのログ出力の無効化について: AWS EC2/ECS

    FlaskはPythonで開発された軽量なWebフレームワークで、そのシンプルさと拡張性から多くの開発者に利用されています。

    今回は、特にAWSのEC2やECSでFlaskを使用している開発者の皆様向けに、ある一般的な問題とその解決策について共有させていただきます。

    サーバーのヘルスチェックは重要な部分であり、私たちはそのためのエンドポイント/healthzを定期的にポーリングしています。

    しかし、これがもたらすログ出力の多さが問題となり、重要な情報が見づらくなることがあります。

    そこで、ヘルスチェックのエンドポイントに関するログだけを無効にする方法について考察しました。

    解決策：特定のルートのログ出力をフィルタリングする

    この問題に対する答えの一つとして、PythonのloggingモジュールのFilterクラスを用いる方法があります。このクラスを継承し、カスタムフィルタクラスを作成します。このフィルタをロガーに追加することで、特定のログメッセージをフィルタリングできます。

    import logging

    class HealthCheckFilter(logging.Filter):
    def filter(self, record):
    return "/healthz" not in record.getMessage()

    ここで定義したHealthCheckFilterクラスは、ログメッセージに/healthzが含まれていない場合にのみTrueを返します。

    したがって、/healthzがメッセージに含まれるログエントリはフィルタリングされ、出力されません。

    そして、このフィルタを適用したいロガーに追加します：

    logger = Logger().getLogger('default')
    logger.addFilter(HealthCheckFilter())

    これにより、ヘルスチェックに関するログ出力が無効化され、ログの可読性が向上します。

    注意点

    この解決策は、特定のロガーに対してのみ適用されます。全体のロガーへの適用はされません。

    例えば、Flaskは内部的にWerkzeugサーバーを利用しており、そのサーバーは自己のロガー(‘werkzeug’)を持っています。

    そのため、もしWerkzeugサーバーからのログもフィルタリングしたい場合は、同様にフィルタを追加する必要があります。

    この記事がAWSのEC2やECSで大規模なFlaskアプリケーションを運用している皆様の一助になれば幸いです。

    特定のエンドポイントのログ出力を制御することで、ログの可読性が向上し、重要な情報へのアクセスが容易になります。

    私たちはまだ学び続けており、他の解決策や改善点があれば、ぜひ共有いただければと思います。

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

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

    クラウドストレージのOneDriveにファイルを自動的にアップロードする方法について話します。特にPythonを使ったアプローチに焦点を当てます。この記事は、開発者、データサイエンティスト、またはOneDriveを活用して効率を上げたい人向けに書かれています。

    前提条件

    このチュートリアルでは、Python 3.6以上がインストールされていることを前提とします。また、必要なPythonパッケージをインストールするためにpipも必要です。これらがまだインストールされていない場合は、公式のPythonウェブサイトからダウンロードしてください。

    必要なパッケージ

    まず最初に、このプロジェクトで必要となるPythonパッケージをインストールしましょう。今回はOffice365のREST APIクライアントライブラリを使用します。このパッケージは、OneDriveとの連携を容易にします。以下のコマンドを実行してパッケージをインストールしましょう：

    pip install Office365-REST-Python-Client

    この操作はスクリプトを実行する前に行ってください。

    Pythonスクリプトの作成

    以下に、Pythonを使用してOneDriveにファイルをアップロードするための基本的なスクリプトを提供します。

    （インデントが崩れてしまってすみません。。）

    from office365.runtime.auth.user_credential import UserCredential
from office365.runtime.client_request_exception import ClientRequestException
from office365.onedrive.driveitems.drive_item import DriveItem
from office365.graph_client import GraphClient
from office365.runtime.auth.authentication_context import AuthenticationContext
import os


def get_onedrive_context_using_user():
tenant = "xxxxxxxxxx"
authority_url = f"https://login.microsoftonline.com/{tenant}"
client_id = "xxxxxxxxxx"
client_secret = "password"

context_auth = AuthenticationContext(authority_url)
context_auth.acquire_token_for_app(client_id, client_secret)

client = GraphClient(context_auth)
return client


def create_onedrive_directory(client, dir_name: str):
"""
Creates a folder in the OneDrive directory.
"""
if dir_name:
drive_item = DriveItem(client)
drive_item.name = dir_name
drive_item.folder = {"childCount": 0}
return client.me.drive.root.children.add(drive_item)


local_path = "1G.csv"
dir_name = 'test-tsukada-dir-hoge'
chunk_size = 50000000

def print_upload_progress(range_pos):
print("{0} bytes uploaded".format(range_pos))

try:
client = get_onedrive_context_using_user()
create_onedrive_directory(client, dir_name)
remote_drive = client.me.drive.root.get_by_path(dir_name)

with open(local_path, 'rb') as f:
remote_drive.resumable_upload(f, chunk_size=chunk_size, chunk_uploaded=print_upload_progress).get().execute_query()
print(f"File {local_path} has been uploaded")
except ClientRequestException as ex:
print(ex)

    このスクリプトは OneDrive 上のルートにフォルダを作成し、指定したファイルをそのフォルダにアップロードします。アップロードの進行状況はコールバック関数print_upload_progress を用いて表示します。

    ただし、このコードは Azure AD のアプリ登録で取得した client_id と client_secret を使用していますので、それぞれ適切な値に置き換えてください。また、Azure AD のテナントIDも必要となります。

    スクリプトの実行

    スクリプトを実行するには、以下のコマンドをターミナルまたはコマンドプロンプトから実行します：

    python upload_to_onedrive.py

    これで、指定したファイルがOneDriveに自動的にアップロードされます！

    まとめ

    Pythonを使用してOneDriveにファイルを自動アップロードする方法について解説しました。この方法は、定期的なバックアップ、大量のファイルの移行、または一般的なファイルの管理など、さまざまなシナリオで役立ちます。

  • PandasとSnowflakeでNULL値を扱う

    PandasとSnowflakeでNULL値を扱う

    PandasとSnowflakeでNULL値を扱う

    PythonのPandasライブラリとSnowflakeデータベースの間でデータを移動させる場合、 NULL値の扱いはしばしば問題になります。 この記事では、Pandasのデータフレーム内でのNULL値の表現と、 その値をSnowflakeに書き込む方法を説明します。

    PandasでのNULL値

    Pandasでは、データフレーム内の欠損値は None または numpy.nan で表現されます。 これらの値は、数値データだけでなく日付/時刻データに対しても使用できます。 以下のコードスニペットは、データフレームの特定の位置に numpy.nan を設定する方法を示しています。

    import numpy as np

# Set a value in the TIMESTAMP_NTZ column to NaN
df.loc[0, 'your_timestamp_column'] = np.nan

    このコードは、データフレームの ‘your_timestamp_column’ 列の0行目の値を numpy.nan に設定します。

    NULL値をSnowflakeに書き込む

    PandasからSnowflakeへのデータの書き込みは、 Snowflake-connector-pythonの write_pandas 関数を使用して行うことができます。 この関数は、Pandasの None と numpy.nan の値をSnowflakeの NULL として解釈します。 以下のコードスニペットは、上で修正したデータフレームをSnowflakeに書き込む方法を示しています。

    from snowflake.connector.pandas_tools import write_pandas

# Then write the DataFrame to Snowflake
write_pandas(sf_conn, df, snowflake_table)

    このコードは、sf_conn（Snowflakeへの接続）を使用して、データフレーム df を snowflake_table という名前のテーブルに書き込みます。 この方法により、Pandasデータフレーム内でNULL値を簡単に扱い、その値をSnowflakeに適切に書き込むことができます。 以上、PythonとPandas、そしてSnowflakeを活用してデータをより柔軟に扱う方法についてお伝えしました。

  • No space left on device エラーへの対処法

    No space left on device エラーへの対処法

    概要

    今回は、Pythonパッケージのインストール時に遭遇する可能性のある一つの問題、 「No space left on device」エラーとその対処法について説明します。 このエラーが発生する場合、最初に疑うべきは、確かにストレージ容量が足りていない可能性です。 しかし、たとえストレージに十分な空き容量がある場合でも、特定の状況下ではこのエラーが発生します。その解決方法を説明します。

    解決策への道筋

    Pythonパッケージのダウンロードやインストール時に、一時的なファイルが作成されます。 これらの一時的なファイルは通常 /tmp ディレクトリに保存され、大規模なパッケージをインストールする際に /tmp の容量を超えてしまうことがあります。これがエラーの原因となります。 解決策としては、一時ファイルの保存場所を /tmp から別のディレクトリに変更します。これは、環境変数 TMPDIR を設定することで実現します：

    export TMPDIR=/var/tmp

    そして、再度パッケージのインストールを試みます：

    pip3 install --no-cache-dir accelerate

    注意点として、この変更は現在のシェルセッションにのみ適用されます。システム全体への恒久的な変更を行いたい場合は、適切な設定ファイル（例えば ~/.bashrc や ~/.bash_profile）に上記の export 行を追加します。

    まとめ

    本記事では、「No space left on device」エラーが発生する原因と対処法について解説しました。 この情報が皆様の問題解決の一助となり、よりスムーズにPythonを利用してビジネスを推進する助けとなれば幸いです。

  • Python の set を使って重複要素を簡単に削除する

    Python の set を使って重複要素を簡単に削除する

    本記事では、Python の set を使用して、リストや他のコレクションから重複要素を効率的に削除する方法について説明します。 また、set による重複削除の使い所や、別のやり方で重複を削除する方法についても触れます。

    Python の set とは？

    Python の set は、順序がなく、ユニークな要素のみを格納するデータ構造です。 それぞれの要素は一度だけ現れます。したがって、set() を使うことで、自動的に重複が削除されることになります。 set を使った重複要素の削除 リストや他のコレクションから重複要素を削除するには、次のように set を使います。

    original_list = [1, 2, 2, 3, 4, 4, 5]

# 重複要素を削除
unique_list = list(set(original_list))

print(unique_list) # 出力: [1, 2, 3, 4, 5]

    この例では、original_list から重複要素を削除するために、リストを set に変換し、その後再度リストに戻しています。

    使い所

    set を使った重複要素の削除は、以下のようなシチュエーションで役立ちます。 データ解析や前処理で、重複するデータポイントを削除する必要がある場合 ユーザーからの入力データに重複がある場合 データベースや CSV ファイルから取得したデータで、重複したレコードを削除する必要がある場合

    他の重複削除の方法

    set 以外にも、Python では重複要素を削除する方法がいくつかあります。例えば、次のような方法が挙げられます。

    リスト内包表記と if 文を使った方法

    リスト内包表記と if 文を使って、リストから重複要素を削除することもできます。

    original_list = [1, 2, 2, 3, 4, 4, 5]

unique_list = []
[unique_list.append(x) for x in original_list if x not in unique_list]

print(unique_list) # 出力: [1, 2, 3, 4, 5]

    ただし、この方法はリストの要素数 が大きい場合には、パフォーマンスが低下することがあります。なぜなら、リストの要素を一つずつチェックして重複を削除するため、計算量が増えるからです。

    辞書を使った方法

    辞書を使って、リストから重複要素を削除することもできます。

    original_list = [1, 2, 2, 3, 4, 4, 5]

unique_list = list(dict.fromkeys(original_list).keys())

print(unique_list) # 出力: [1, 2, 3, 4, 5]

    この方法では、リストの要素を辞書のキーとして利用し、dict.fromkeys() を使って新しい辞書を作成します。 その後、keys() メソッドで辞書のキーを取得し、リストに変換します。 ただし、これらの方法は、元のリストの要素がハッシュ可能（つまり、辞書のキーとして使用できる）であることが前提です。 ハッシュ不可能な要素が含まれる場合、set など他の方法で重複を削除する必要があります。

    まとめ

    Python の set を使って、リストや他のコレクションから重複要素を効率的に削除する方法を紹介しました。 また、set による重複削除の使い所や、別のやり方で重複を削除する方法についても触れました。 データ解析や前処理など、様々なシチュエーションで重複要素の削除が求められるため、set などの適切な手法を選択することが重要です。

  • Pythonのforループで現在のループが全体のレコードのうち何レコード目かを知る方法

    Pythonのforループで現在のループが全体のレコードのうち何レコード目かを知る方法

    Pythonのforループで、全体のレコード数が分かっている場合、現在のループが全体のレコードのうち何レコード目かを知ることができます。この記事では、PandasのDataFrameオブジェクトを例に、Pythonのforループで現在のループが全体のレコードのうち何レコード目かを知る方法を解説します。

    方法1：enumerate()を使用する方法

    PandasのDataFrameオブジェクトの場合、以下のようにenumerate()を使用して、現在のループが全体のレコードのうち何レコード目かを取得することができます。

    row_count = df.shape[0]

for i, row in enumerate(df.values):
print(i + 1) # 現在のレコード数
print(row_count) # 全体のレコード数

    上記のコードでは、enumerate()を使用してループの各回数を取得し、iという変数に代入しています。そして、ループ内で現在のループが何レコード目なのかを知りたい場合は、i + 1をすればよいです。+1をすることで、インデックス0から始まるPythonの慣習に従い、1から始めることができます。

    方法2：len()とrange()を使用する方法

    enumerate()を使用せずに、以下のようにlen()関数を使用してデータフレームの行数を取得し、ループ内でカウント変数を使用して現在のループが全体のレコードのうち何レコード目かを知ることができます。

    row_count = len(df.index)
count = 1

for row in df.values:
# 現在のレコード数と全体のレコード数を出力する
print(count, "/", row_count)

    # ループ内の処理 count += 1 上記のコードでは、len(df.index)を使用してデータフレームの行数、つまりレコード数を取得し、カウント変数countを使用して現在のループが全体のレコードのうち何レコード目かを数えています。また、ループ内では、countとrow_countを出力して、現在のループが全体のレコードのうち何レコード目かを知ることができます。

    まとめ

    Pythonのforループで現在のループが全体のレコードのうち何レコード目かを知るには、enumerate()を使用する方法と、len()とrange()を使用する方法があります。 enumerate()を使用する方法は、ループの各回数を取得して、現在のループが何レコード目かを知ることができます。 len()とrange()を使用する方法は、データフレームの行数を取得して、カウント変数を使用して現在のループが何レコード目かを数えることができます。 どちらの方法を使用するかは、データフレームの行数やループ内の処理内容によって異なります。 enumerate()を使用する場合は、行数が多い場合でも効率的に処理することができますが、len()とrange()を使用する場合は、行数が多い場合にはパフォーマンスに影響することがあります。 また、enumerate()を使用する場合は、i + 1を使用することで、Pythonの慣習に従ってループの行番号を1から始めることができます。 以上が、Pythonのforループで現在のループが全体のレコードのうち何レコード目かを知る方法の解説でした。