Pythonのrequestsを使いこなしたい

requests

HTTP リクエストを送信しやすくしたライブラリで、わたしは今回リトライやタイムアウトの機能が使いたくて探していたらみつけた。

ちなみに標準ライブラリにはとくにリトライがないと思っているけど、あったら教えて下さい🙏

Lambda 上で動かす関数を実装しているので、できるだけ使用しているライブラリは少なくしたいから、標準ライブラリで対応しているようだったらそちらを使いたい。

インストールは、Poetryなら

poetry add requests

でおっけー！

yuki

このあたりを参考にしながらリトライを実装する。

yuki

ちょっと雑で汎用性がないけど、たとえば抽象化していくとこんな感じになるかな？基本はDefaultHttpClientを呼んでもらって、リトライは絶対走る状態にして使ってもらう。もしカスタマイズが必要ならCustomHttpClientを呼んでもらう。

Protocolにしてあるので、HttpClientを使用したいクラスに依存させておけばあとは切り替えができる。

ちなみにこの実装、まだ動かしてないので、実行時エラーが出るかも。

from typing import Protocol, Any
from requests.models import Response
from requests.packages.urllib3.util.retry import Retry
from requests.adapters import HTTPAdapter
from requests import Session
import json


class HttpClient(Protocol):
    def get(self, url: str) -> Response:
        raise NotImplementedError

    def post(self, url: str, body: Any) -> Response:
        raise NotImplementedError


class DefaultHttpClient:
    def __init__(self) -> None:
        retry_strategy = Retry(
            total=3, status_forcelist=[429, 500, 502, 503, 504], backoff_factor=2
        )
        adapter = HTTPAdapter(max_retries=retry_strategy)
        http = Session()
        http.mount("http://", adapter)
        http.mount("https://", adapter)
        self.client: Session = http

    def get(self, url: str) -> Response:
        return self.client.get(url=url)

    def post(self, url: str, body: Any) -> Response:
        return self.client.post(url, json=json.dumps(json).encode("utf-8"))


class CustomHttpClient:
    def __init__(self, retry_strategy: Retry) -> None:
        adapter = HTTPAdapter(max_retries=retry_strategy)
        http = Session()
        http.mount("http://", adapter)
        http.mount("https://", adapter)
        self.client: Session = http

    def get(self, url: str) -> Response:
        return self.client.get(url=url)

    def post(self, url: str, body: Any) -> Response:
        return self.client.post(url, json=json.dumps(json).encode("utf-8"))

処理が似通った場所が出てきているので、ここからさらにまとめられるね。

yuki

Protocol の PEP をちょっと確認して直してみた。

from typing import Protocol, Any
from requests.models import Response
from requests.packages.urllib3.util.retry import Retry
from requests.adapters import HTTPAdapter
from requests import Session
import json


class HttpClient(Protocol):

    client: Session

    def get(self, url: str) -> Response:
        return self.client.get(url=url)

    def post(self, url: str, body: Any) -> Response:
        return self.client.post(url, json=body.__dict__)


class DefaultHttpClient(HttpClient):
    def __init__(self) -> None:
        retry_strategy = Retry(
            total=3, status_forcelist=[429, 500, 502, 503, 504], backoff_factor=2
        )
        adapter = HTTPAdapter(max_retries=retry_strategy)
        http = Session()
        http.mount("http://", adapter)
        http.mount("https://", adapter)
        self.client: Session = http

    def get(self, url: str) -> Response:
        return super().get(url)

    def post(self, url: str, body: Any) -> Response:
        return super().post(url, body)


class CustomHttpClient(HttpClient):
    def __init__(self, retry_strategy: Retry) -> None:
        adapter = HTTPAdapter(max_retries=retry_strategy)
        http = Session()
        http.mount("http://", adapter)
        http.mount("https://", adapter)
        self.client: Session = http

    def get(self, url: str) -> Response:
        return super().get(url)

    def post(self, url: str, body: Any) -> Response:
        return super().post(url, body)

yuki

json.dumps に class を投げ込むとシリアライズできない

に、直面した。そんなに甘くなかった。

from dataclasses import dataclass
from typing import List

@dataclass
class SimpleGreeter:
    greet: str
    names: List[str]

json.dumps(SimpleGreeter(greet="hello", names=["yuki"]))

これだと、シリアライズできません系の実行時エラーが出る。

self = <json.encoder.JSONEncoder object at 0x109dcd310>, o = SimpleGreeter(greet='hello', names=['yuki'])

    def default(self, o):
        """Implement this method in a subclass such that it returns
        a serializable object for ``o``, or calls the base implementation
        (to raise a ``TypeError``).
    
        For example, to support arbitrary iterators, you could
        implement default like this::
    
            def default(self, o):
                try:
                    iterable = iter(o)
                except TypeError:
                    pass
                else:
                    return list(iterable)
                # Let the base class default method raise the TypeError
                return JSONEncoder.default(self, o)
    
        """
>       raise TypeError(f'Object of type {o.__class__.__name__} '
                        f'is not JSON serializable')
E       TypeError: Object of type SimpleGreeter is not JSON serializable

~/.pyenv/versions/3.8.5/lib/python3.8/json/encoder.py:179: TypeError

class は標準ではシリアライズ可能な状態になっていない。なので、エンコーダーが認識できずに落ちる。

回避策はいくつかあるけど、フィールドを削るなどの特殊な操作が必要なく、単純にクラスの内容をJSONへエンコードしたい場合には、辞書型に直してしまうと早い。Python では特殊メソッドで __dict__ が存在するので、これを使用する。

# 先程の json.dumps ... のコードを下記に書き換える。
json.dumps(SimpleGreeter(greet="hello", names=["yuki"]).__dict__)

これは動作する。

このスクラップは2020/12/09にクローズされました