pythonでclass無いの全ての関数が経由する関数を作成したい

Question

#　前提
###開発言語
・python3
###使用
・自動ツールで、処理のパラメータなどをスプレッドシートから読み取り、結果をスプレッドシートに記入しています。
・ツールは複数実行される
・ツール数は20以上で、同時には実行されないように時間をずらしていますが、どうしても複数重なるタイミングがあります。その時にエラーが発生

困っていること

pythonのコードの書き方で、例外処理をどうすれば綺麗に書けるかわかりません。
pythonのgspreadを使っていてアクセス制限が生じます。
それは、100秒間に100アクセスしかできない制約で、このエラー自体は起きるものだと想定しています。
そのエラーの場合100秒間待機してから再呼び出しをしたいと考えています。

現状

class Spread_Sheet:
    def __init__(self):
        # シートのオブジェクトを作成
        #2つのAPIを記述しないとリフレッシュトークンを3600秒毎に発行し続けなければならない
        scope = ['https://spreadsheets.google.com/feeds','https://www.googleapis.com/auth/drive']
        #認証情報設定
        json_string = gas_json.json_string
        credentials = ServiceAccountCredentials._from_parsed_json_keyfile(json_string, scope)
        #OAuth2の資格情報を使用してGoogle APIにログインします。
        gc = gspread.authorize(credentials)
        
        self.workbook = gc.open_by_key(SPREADSHEET_KEYS)


    # シートが存在するのかチェック
    def exists_vsheet_check(self,s_name):
        is_Exists = False
        worksheets = self.workbook.worksheets() # シート一覧を取得
        worksheets_list = [ss.title for ss in worksheets]
        
        if s_name in worksheets_list:
            is_Exists = True
        return is_Exists

    # パラメータの取得
    def get_parameter(self,s_name):
        pass

    # データの更新
    def set_date(self,s_name):
        pass

    # ・
    # ・
    # ・

この様に、スプレッドシートを操作するクラスを用意して、全てのアクセスをそのクラスからするようにしています。
基本的にself.workbookを使うと1回のカウントとなり、100秒間に100回を超えるとエラーがでます。

そのエラーがgspread.exceptions.APIErrorです。これに例外処理したら次はrequests.exceptions.connectionerrorのエラーが出たので下記のように変更しました。

class Spread_Sheet:
        e_count = 0
        while True:
            try:

                # シートのオブジェクトを作成
                #2つのAPIを記述しないとリフレッシュトークンを3600秒毎に発行し続けなければならない
                scope = ['https://spreadsheets.google.com/feeds','https://www.googleapis.com/auth/drive']
                #認証情報設定
                json_string = gas_json.json_string
                credentials = ServiceAccountCredentials._from_parsed_json_keyfile(json_string, scope)
                #OAuth2の資格情報を使用してGoogle APIにログインします。
                gc = gspread.authorize(credentials)
                #共有設定したスプレッドシートキーを変数[SPREADSHEET_KEY]に格納する。
                SPREADSHEET_KEYS = ff.get_post_spread_key()
                
                # ワークブック main
                self.workbook = gc.open_by_key(SPREADSHEET_KEYS[0])
                # ワークブック 動画のあるシート
                self.workbook_video = gc.open_by_key(SPREADSHEET_KEYS[(ff.get_tool()-1)//10])
                return
            except:
                random.randint(100,150)
                e_count +=1
                if e_count > 5:
                    raise 'Set_ALL_mail_message'


    # シートが存在するのかチェック
    def exists_vsheet_check(self,s_name):
        e_count = 0
        while True:
            try:

                is_Exists = False
                worksheets = self.workbook_video.worksheets() # シート一覧を取得
                worksheets_list = [ss.title for ss in worksheets]
                
                if s_name in worksheets_list:
                    is_Exists = True
                return is_Exists

            except:
                random.randint(100,150)
                e_count +=1
                if e_count > 5:
                    raise 'exists_vsheet_check'

全てのアクセスするとき、エラーが出たら5回は繰り返しています　笑
絶対間違っていると思ってここをなんとかしたいです。

考えたこと

要は、例外処理する場所を一か所にしたいです。
なので、高級関数を使って全部関数が経由する関数を作れば良いのでは無いかと思いましたが、引数がまちまちだったので向いていないと判断しました。

classにそういう機能が無いかな?と思ったのですが見つけることができませんでした。
何か良い方法はないでしょうか?

Answer 1

---

※本来は、たとえ面倒でも、関数ごとに例外処理を行うのが正攻法だと個人的には考えます。
（ちょっと楽にしたいならデコレータを用いましょう）
なぜならば、関数（処理の単位）ごとにキャッチすべき例外や対処内容は異なることが普通だと思うからです。
また、サーバへの負荷を軽減するためのタイムアウト措置をかような方法で回避することは、本来の趣旨から外れるのでは、という考え方もできます
（タイムアウト措置が実装されている意味は、「タイムアウトが頻発するほどの速度で連続処理を行うべきではない」というイデオロギーが根底にあるかもしれないのです）

---

閑話休題。

「例外処理する場所を一か所にしたい」とは、かみ砕くと、gspreadを操作するとき回数制限で発生する例外処理を関数毎に書きたくない、ことだと理解しました。

この目的を達するために、やや強引ですが、gspreadのソースを参考に、すべてのAPIリクエストの元締めであるgspreadライブラリのclientオブジェクトのrequest関数を書き換えてみました。

下記はクラスを使わず平文でシート操作する場合となっています。
質問者さんの元のSpreadSheetクラスへの組み込みとかは御自分で考えてみてください。
（元のコードはgas_jsonとかよくわからなかったので）

また、下記はシングルスレッドで各処理を行っているという前提です。

import gspread

from gspread.exceptions import APIError
import time


# 認証部分は省略

# gspread.Client.request()関数を上書きするカスタム関数
def custom_request(self, method, endpoint, params=None, data=None, json=None,
                   files=None, headers=None,
    ):
    for _ in range(5):
        response = getattr(self.session, method)(
            endpoint, json=json, params=params,
            data=data, files=files, headers=headers,
        )

        if response.ok:
            return response
        if response.status_code == 429:
            # リクエスト制限が発生したときは100+1秒待つ
            print("request limit. wait 100 secs")
            time.sleep(101)
        else:
            # その他のエラーの場合は5秒待ってリトライ
            print(f"Unknown Error:{response.status_code}")
            time.sleep(5)
        continue
    # リクエスト制限以外で5回連続でエラーが起きた場合は本当に例外を発生させる。
    else:
        raise APIError(response)

# 関数を書き換える。
gspread.Client.request = custom_request



# 以下は動作確認用のコード--------------
for i in range(1,200):
    print(i)
    val = int(worksheet.cell(i,1).value)

    output_val = val + 100
    worksheet.update_cell(i+1,1, ouput_val)

出力結果

1
2
(略)
112
113
114
115
request limit. wait 100 secs  <==ここでリクエスト制限がかかり約101秒待つ
116　　　　　　　　　　　　　　　<==100秒以上待ったのでリクエスト制限が解除されて動作再開
117
(略)

Comments

[Yuiti628]

これは本当にすごいです！
ありがとうございます????

gspreadの方をいじるなんて考えもしなかったです。

そんな方法があるなんて。

Answer 2

・自動ツールで、処理のパラメータなどをスプレッドシートから読み取り、結果をスプレッドシートに記入しています。

の内容が書いてないので推測になりますが、新規シートの追加が頻繁に行なわれるのでなければ、exists_vsheet_checkで毎回シートタイトルのリストworksheets_listを生成するのではなく、__init___であらかじめ作成しておき、exists_vsheet_checkではそれを参照するだけにすれば、self.workbookへのアクセス数を抑えることができるのでは。

途中で新規シートが追加されるのであれば、その後でworksheets_listを更新するメソッド(を追加作成しておいて)を呼べばいいのです。

Comments

[Yuiti628]

説明不足で申し訳ありません。
中の結構色々あります。データ取得からセットまで。
また、ツール数も多いので、どうしてもアクセス制限にはなると思っています。

そのため、例外の書き方についての質問でした。
省略しすぎて申し訳ありません。