RendorでFlask × SeleniumのwebアプリをDockerでデプロイする

Seleniumによるブラウザの自動化をすることでスクレイピングが可能となります。今回は、スクレイピングした結果を使うようなwebサイトをFlaskで作成し、Rendorで公開するまでの流れを紹介します。

Rendorとは

Rendorとは、様々なwebアプリをデプロイするためのPaaSであり、Herokuのようなサービスを展開しています。 Rendor上では、Githubと連携することで特定のブランチにプッシュされたときのみアプリをデプロイするなど、自動デプロイが可能となります。

公式サイトが出している料金比較にもあるように、有料化したHerokuと比べても比較的安価に運用ができるため、さくっとサービスを公開したいときに適しているのではないかと思います。

ただし、Rendorには注意点もあり、docker composeをサポートしてはいません。その代わり、render.yamlという独自のYAMLファイルを書く必要があります。今回は、docker composeを使って開発したFlaskアプリを render.yaml に置き換えてデプロイするまでを書いていきます。

docs.render.com

Flask × Seleniumのアプリのローカル起動まで

今回は以下の記事をベースとした Flask x Seleniumのwebアプリを使います。

datascience-beginer.com

ローカルで実行する際には、Python用のコンテナとSelenium用のコンテナをdocker composeで同時にビルドし、起動するような構成となります。

ディレクトリ構造

以下のディレクトリ構造のプロジェクトで進めていきます。

.
├── Dockerfile
├── app
│   ├── __init__.py
│   ├── app.py
│   ├── driver.py
│   └── templates
│       ├── index.html
│       └── result.html
├── docker-compose.yml
└── requirements.txt

docker-compose.yml

docker composeは以下のようになっています。

version: '3'

services:
  selenium-worker:               
    image: selenium/standalone-chrome:latest 
    shm_size: 2gb         
    ports:
      - 4444:4444         
      - 7900:7900         

  python:
    build: .
    volumes:
      - ./app:/app
    tty: true
    depends_on:
      - selenium-worker
    ports:
      - 5000:5000
    environment:
      - SELENIUM_WORKER_HOST=selenium-worker

Dockerfile

Flask用のDockerfileは以下のようになっており、5000番ポートでリクエストを受ける形となります。

FROM python:latest
RUN apt-get update && \
    apt-get install -y \
    build-essential \
    cmake \
    git \
    sudo \
    wget \
    vim

RUN pip install --upgrade pip

COPY ./requirements.txt /requirements.txt

RUN pip install -r /requirements.txt

WORKDIR /app

COPY ./app .

ENV FLASK_APP=/app/app.py

EXPOSE 5000

CMD ["gunicorn", "--bind", "0.0.0.0:5000",  "app:app"]

app/app.py

app.pyはFlaskアプリの実装しており、大まかな実装は以下となります。

from flask import Flask, render_template, request

from driver import init_driver, finish_driver

app = Flask(__name__)


@app.route("/")
def index():
    return render_template("index.html")


@app.route("/scrape")
def scrape():

    ...

    driver = init_driver()

    # スクレイピング処理

    finish_driver(driver)

    return render_template("result.html", ...)


if __name__ == '__main__':
    app.run()

app/driver.py

driver.pyはChromeのリモートweb driverを初期化、終了するのに使われます。これらはapp.pyで呼び出されています。

import os

from selenium import webdriver


def init_driver():
    options = webdriver.ChromeOptions()

    ...

    driver = webdriver.Remote(
        command_executor='http://{}:4444/wd/hub'.format(os.environ['SELENIUM_WORKER_HOST']),
        options=options
    )
    driver.set_window_size(500, 500)
    driver.implicitly_wait(10)

    return driver


def finish_driver(driver):
    driver.quit()

ローカル起動

この構成のもと、docker-compose build から docker-compose up をすると、localhost:5000 からアプリにアクセスでき、適当なAPIを叩くことでスクレイピングした結果を返すことができます。ここまでが今回の記事で扱うwebアプリの実装となります。

Renderへのデプロイ

ここからは本題である、Renderへのデプロイに移っていきます。 最初にも書きましたが、Renderでは docker-compose.yml をサポートしていません。 その代わり、独自の render.yaml というものがあり、これを使ってデプロイすることになります。

docs.render.com

Dockerfile.selenium と render.yaml の追加

render.yaml を使った形に変更するにあたって、先ほどのプロジェクトに新たに Dockerfile.selenium と render.yaml を追加します。

render.yaml は先ほどから出てきているように、docker-compose.yml の代わりのファイルであり、Dockerfile.selenium はSelenium standaloneのイメージを利用するためのDockerfileとなります。 docker-compose.yml と違って、 render.yaml ではイメージを指定したりすることができません。そのため、サービスごとに Dockerfile を作って、それぞれに設定を追加していく必要があります。最終的なディレクトリ構造は以下となります。

.
├── Dockerfile
├── Dockerfile.selenium
├── app
│   ├── __init__.py
│   ├── app.py
│   ├── driver.py
│   └── templates
│       ├── index.html
│       └── result.html
├── docker-compose.yml
├── render.yaml
└── requirements.txt

Dockerfile.selenium

Dockerfile.seleniumの中身は以下となります。docker-compose.ymlの中身を書き換えただけなので、非常にシンプルになります。

FROM selenium/standalone-chrome:latest

EXPOSE 4444
EXPOSE 7900

render.yaml

次に、render.yamlは以下となります。

services:
  - type: pserv
    name: selenium-worker
    runtime: docker
    dockerfilePath: ./Dockerfile.selenium
    rootDir: .
    envVars:
      - key: PORT
        value: 4444

  - type: web
    name: flask
    runtime: docker
    dockerfilePath: ./Dockerfile
    rootDir: .
    envVars:
      - key: FLASK_APP
        value: /app/app.py
      - key: SELENIUM_WORKER_HOST
        value: ***

render.yamlでは、各サービスごとにtypeを設定することができ、selenium-workerでは pserv、flaskでは web となっています。 公式サイトによると、それぞれ以下のようなサービスとなっています。

  • web：いわゆるwebサービス。外部からのアクセスができる。
  • pserv (private service)：外部からのアクセスができないサービス。Renderにデプロイされたwebサービスからはアクセスが可能だが、ブラウザなどの外部からはアクセスできない。

今回のwebアプリではselenium-workerはflaskからのみアクセスされるため、pserv としています。

docs.render.com

docs.render.com

ファイル下部のSELENIUM_WORKER_HOSTという環境変数はこの後設定していくため、一旦適当な文字にしておいてください。

Render上での操作

次に、Renderのダッシュボードでの操作に移ります。

Step 1 Blueprintを選択

Step 2 レポジトリを選択

GitHubと連携すると、デプロイしたいレポジトリが選択できます。Connectを押して連携します。

Step 3 プロジェクト名、ブランチの選択

適宜好きな名前を入力しましょう。ここで、下の方に flask-mn4b と selenium-worker-mn4b というサービス名が表示されていることに注目してください。これらが各サービスの名前となっていて、例えばwebサービスはhttps://flask-mn4b.onrender.comというURLでアクセスすることができるようになります。

Step 4 render.yaml の再編集

ここまで来れば気づくかもしれませんが、先ほど省略した SELENIUM_WORKER_HOST という環境変数の値には selenium-worker-mn4b を入れることになります。 ローカルのときには、selenium-workerへは http://selenium-worker:4444/... にアクセスすればよかったのですが、Render上ではURLが変わり、http://selenium-worker-mn4b:4444/... にアクセスするということになります。

結果、render.yamlは以下のようになります。

各サービスの名前はランダムにつけられるようで、デプロイするまで分かりません。 そのため、一回失敗すること前提でRender上でデプロイし、生成された名前をコピペして環境変数に入れ、再度変更をプッシュすることでしかデプロイできないのかもしれません。

services:
  - type: pserv
    name: selenium-worker
    runtime: docker
    dockerfilePath: ./Dockerfile.selenium
    rootDir: .
    envVars:
      - key: PORT
        value: 4444

  - type: web
    name: flask
    runtime: docker
    dockerfilePath: ./Dockerfile
    rootDir: .
    envVars:
      - key: FLASK_APP
        value: /app/app.py
      - key: SELENIUM_WORKER_HOST
        value: selenium-worker-mn4b

Step 5 デプロイ結果

修正した render.yaml をプッシュした結果、デプロイが成功することが確認できました。

まとめ

今回はFlaskとSeleniumを使ったwebアプリをDockerを使ってRenderにデプロイする方法を紹介しました。 Renderは初めて使ってみたのですが、自動デプロイは高速で簡単なので、インフラ周りを触ることなくデプロイができ非常に便利だと思いました。

一方、docker composeには対応しておらず、render.yaml という独自のyamlファイルを書く必要があるため、不便さもありました。 docker composeを使えるようにしてくれという声は昔からたくさんあるようなのですが、いまだに対応していない様子を見ると、今後もdocker composeがサポートされることは期待できないです。

feedback.render.com

また、docker-compose.yml ではselenium-workerにshm_size: 2gb を設定することでメモリ不足になることを防いでいたのですが、render.yaml ではshm_sizeを設定することはできないため、デプロイはできてもメモリ不足に陥ってしまうことが多いと思います。 今後 shm_size をRenderがサポートするのを期待したいです。

[Go] 並行処理を用いたモンテカルロ法

Go golang

Go言語による並行処理の第4章4.6パイプラインをもとにモンテカルロ法で円周率を計算する。 www.amazon.co.jp

乱数による円周率の計算

以下の積分にモンテカルロ法を適用し、円周率をモンテカルロ法で求める。 $$ \int_{0}^{1} \frac{4}{ 1 + x^{2} } dx = \pi $$

サンプル数 $ M $ を$ 10,10^{3},10^{5},10^{7} $として、

$$ \bar{f_i} = \frac{1}{M} \sum_{t = 1, ..., M} \frac{4}{1 + {X_t}^{2}} $$

を求め、推定値

$$ \bar{ \bar{f} } = \frac{\sum_i \bar{f}_i}{N_M} $$

を計算する。ただし、$ N_{M}$は独立なモンテカルロ計算の回数である。

並行処理

パイプライン: データを受け取って、何らかの処理をし、どこかに渡すという一連の処理。

本に書いてあるrepeatFn関数やtake関数を用いて実装する。

実装

func repeatFn(done <-chan interface{}, fn func() float64) <-chan float64 {
    valueStream := make(chan float64)
    go func() {
        defer close(valueStream)
        for {
            select {
            case <-done:
                return
            case valueStream <- fn():
            }
        }
    }()
    return valueStream
}

func take(done <-chan interface{}, valueStream <-chan float64, num int) <-chan float64 {
    takeStream := make(chan float64)
    go func() {
        defer close(takeStream)
        for i := 0; i < num; i++ {
            select {
            case <-done:
                return
            case takeStream <- <-valueStream:
            }
        }
    }()
    return takeStream
}

func calc(done <-chan interface{}, valueStream <-chan float64, fn func(float64) float64) <-chan float64 {
    calcStream := make(chan float64)
    go func() {
        defer close(calcStream)
        for x := range valueStream {
            select {
            case <-done:
                return
            case calcStream <- fn(x):
            }
        }
    }()
    return calcStream
}

func calculatePi(n int) float64 {
    var piApprox float64
    for i := 0; i < n; i++ {
        x := rand.Float64()
        piApprox += (4 / (1 + x*x))
    }
    piApprox /= float64(n)
    return piApprox
}

func main() {
    done := make(chan interface{})
    defer close(done)

    var piApprox float64
    var n float64 = math.Pow(10, 6)
    rand := func() float64 { return rand.Float64() }
    f := func(x float64) float64 { return 4 / ((1 + x*x) * n) }

    start := time.Now()
    loop := func() <-chan float64 {
        valueStream := make(chan float64)
        go func() {
            defer close(valueStream)
            for i := 0; i < 10; i++ {
                var mean float64
                for num := range calc(done, take(done, repeatFn(done, rand), int(n)), f) {
                    mean += num
                }
                valueStream <- mean
            }
        }()
        return valueStream
    }

    for num := range loop() {
        piApprox += num
    }

    fmt.Printf("calculated pi: %v, time: %v s\n", piApprox/10, time.Since(start).Seconds())

    start = time.Now()

    piApprox = 0
    for i := 0; i < 10; i++ {
        piApprox += calculatePi(int(n))
    }
    fmt.Printf("calculated pi: %v, time: %v s\n", piApprox/10, time.Since(start).Seconds())

}

calc関数では乱数$ x$に対して$ \frac{4}{1 + x^{2}}$を計算し、それをcalcStreamに追加している。

また、calculatePiは並行処理をせずに単純にモンテカルロ計算をするための関数である。

結果

  • 並行処理
サンプル数 計算された円周率 実行時間(s)
$10$ $ 3.15711$ $ 4.36634 \times 10^{-4}$
$10^{3}$ $ 3.14847$ $0.018$
$10^{5}$ $ 3.14215$ $1.771$
$10^{7}$ $ 3.14151$ $174.374$
  • 単純に処理
サンプル数 計算された円周率 実行時間(s)
$10$ $ 3.23199$ $2.757 \times 10^{-6}$
$10^{3}$ $ 3.14217$ $ 1.75575 \times 10^{-4}$
$10^{5}$ $ 3.14065$ $0.019$
$10^{7}$ $ 3.141571$ $1.772$

考察

明らかに単純にcalculatePi関数を用いて計算した方が速いです。何重にもするのではなく、もっとパイプラインをまとめたほうが速くなりそうです。

また、パイプラインの使い方を本を元に実装する上でちょうど良い例になったのではないかと思います。

[Go] Go言語による並行処理 4章メモ1

Go golang

Go言語による並行処理の第4章をまとめていく。 www.amazon.co.jp

4.1 拘束

並行プロセスを安全にする方法として、

  • データをイミュータブルにする(作成後に変更できないようにする)
  • データを拘束によって保護する

などが挙げられる。 拘束について、

  • アドホック拘束: 規約によって拘束を達成する
  • レキシカル拘束: レキシカルスコープを使って適切なデータと並行処理のプリミティブだけを複数の並行処理プロセスで使えるように公開する

の2つがある。

レキシカル拘束

3章でみた、チャネルの読み書きの必要な権限だけを公開するというのもその1つ

次の例では、printDataはdataスライスには直接アクセスできず、引数として渡さねばならない。 また、printDataの中ではdataスライスの一部しか見ることができない。

printData := func(wg *sync.WaitGroup, data []byte) {
    defer wg.Done()

    var buff bytes.Buffer
    for _, b := range data {
        fmt.Fprintf(&buff, "%c", c)
    }
    fmt.Println(buff.String())
}

var wg sync.WaitGroup
wg.Add(2)
data := []byte("golang")
go printData(&wg, data[:3])
go printData(&wg, data[3:])

wg.Wait()

for-selectループ

for-selectループを用いるパターンを挙げる。

  • チャネルから繰り返しの変数を送出する
for _, s := range []string{"a", "b", "c"} {
    select {
    case <- done:
        return
    case stringStream <- s:
    }
}
  • 停止シグナルを待つ無限ループ
for {
    select {
    case <-done:
        return
    default:
    }
    // 割り込みできない処理
}

または、

for {
    select {
    case <-done:
        return
    default:
        // 割り込みできない処理
    }
}

4.3 ゴルーチンリークを避ける

ゴルーチンが終了するのは、

  1. 処理を完了する場合
  2. 回復不可能なエラーで処理が続けられない場合
  3. 停止するよう命令された場合

あり、3のようにゴルーチンを終了するには、doneという読み込み専用チャネルを使用する。 ゴルーチンがゴルーチンを生成したのなら、その生成したゴルーチンを停止できるようにするべきである。

doWork := func(done <-chan interface{}, strings <-chan string) <- chan interface{} {
    terminated := make(chan interface{})
    go func() {
        defer fmt.Println("doWork exited")
        defer close(terminated)
        for {
            select {
            case s := <- strings:
                // 処理
                fmt.Println(s)
            case <- done:
                return
            }
        }
    }()
    return terminated
}

done := make(chan interface{})
terminated := doWork(done, nil)

go func() {
    // 1秒後に操作をキャンセル
    time.Sleep(1 * time.Second)
    fmt.Println("Canceling doWork goroutine...")
    close(done)
}

<-terminated
fmt.Println("done")

4.4 orチャネル

1つ以上のチャネルを1つのチャネルにまとめ、どれか1つが閉じられたら全部のチャネルを閉じるようにする。 再帰をつかったorチャネルで実現できる(実装略)

4.5 エラーハンドリング

取得される結果とエラーを対にする。 http.Get(url)をするプログラムであれば、

type Result struct {
    Error error
    Response *http.Response
}

を作り、2つをまとめる。

4.6 パイプライン

  • パイプライン: データを受け取って、何らかの処理を行い、どこかに渡すという操作をまとめたもの
  • ステージ: パイプラインで行う操作1つ1つのこと
    • 受け取る型と返す型が同じ
    • 具体化されていない
    • バッチ処理とストリーム処理
      • バッチ処理: データを一気に処理
      • 要素を1つ受け取って、1つ返す

[Go] Go言語による並行処理 3章メモ2

Go golang

Go言語による並行処理の第3章をまとめていく。 www.amazon.co.jp

3.3 チャネル

宣言、初期化

値をchan型の変数に渡し、プログラムの別の場所でその値をチャネルから読み取る。 以下で宣言、初期化ができる。

var dataStream chan interface{}
dataStream = make(chan interface{})

読み込み、書き込み

送信の場合は->をチャネルの右側に、受診の場合は->を左側に置く。

stringStream := make(chan string)
go func() {
    stringStream <- "hello" // 送信
}

fmt.Println(<-stringStream) // 受診
}

上の例で送信を行うゴルーチンが終了する前にメインゴルーチンが終了しないのは、チャネルがブロックをするからである。

  • チャネルのキャパシティがいっぱいのとき: チャネルに空きができるまで書き込みを待機
  • チャネルが空: 少なくとも1つの要素が入るまで読み込みを待機

チャネルを閉じる

s, ok := <- stringStreamとしたとき、2つ目の戻り値は読み込みができたかどうか、もしくは閉じたチャネルから生成されたデフォルト値のどちらかを示す。チャネルを閉じるのは以下で行う。

valueStream := make(chan interface{})
close(valueStream)

これにより、チャネルをループで処理することができる。

intStream := make(chan int)
go func() {
    defer close(intStream)
    for i := 1; i <= 5; i++ {
        intStream <- i
    }
}()

for integer := range intStream {
    fmt.Println("%v ", integer)
}

また、3.2節のCond型で複数のゴルーチンに同時にシグナルを送信する方法もあったが、それもチャネルで代用できる。(コード略)

バッファ付きチャネル

初期化の際にキャパシティ(nとする)が与えられたチャネルを生成することで、一度も読み込みが行われなくてもn回書き込みが可能。 バッファ付きチャネルはFIFOキューと同じ動作をする。

var dataStream chan interface{}
dataStream = make(chan interface{}, 4) // キャパシティが4

チャネルの所有権

所有権: チャネルを初期化し、書き込み、閉じるゴルーチン 単方向チャネルを宣言するとチャネルを所有するゴルーチンとチャネルを利用するだけのゴルーチンを区別できる。 プログラム内ではチャネルの所有権のスコープを小さくする。

3.4 select文

読み込みの場合、チャネルに書き込みがあったか、閉じられたかを、書き込みの場合キャパシティいっぱいになっていないかを確認する。どのチャネルも当てはまらない場合はブロックする。 case文全体に対して擬似乱数による一様選択をしており、複数のチャネルがあった場合、どのcaseが実行されるかはランダムとなる。

var c1, c2 <- chan interface{}
var c3 chan<- interface{}

select {
case <- c1:
    ...
case <- c2:
    ...
case c3 <- struct{}{}:
    ...
}

すべてのチャネルがブロックしているときに何かするにはdefault節を用いる。 通常、default節はfor-selectループの中で用いられる。

[Go] Go言語による並行処理 3章メモ1

Go golang

Go言語による並行処理の第3章をまとめていく。 www.amazon.co.jp

3.1 ゴルーチン

関数呼び出しの前にgoキーワードを置くことでゴルーチンを起動できる。

func main() {
    go f()
    ...
}

func f {
    ...
}

3.2 syncパッケージ

3.2.1 WaitGroup

Addでカウンターを1増やし、Doneで1減らす。Waitを呼び出すとカウンターがゼロになるまでブロックする。

var wg sync.WaitGroup

wg.Add(1)
go func() {
    defer wg.Done()
    ...
}

wg.Wait()

3.2.2 MutexとRWMutex

プログラム内のクリティカルセクションを保護する。 以下の例の場合、変数aを保護、その解除を行っている。

var a int
var lock sync.Mutex

... 
go func() {
    lock.Lock()
    defer lock.Unlock()
    a++
}
...

RWMutexは読み込みと書き込みを区別することができる。Lockは上のMutexでのLockと同様、読み書き両方をロックする。RLockは書き込みのみをロックし、読み込みは可能となる。

var lock sync.RWMutex

lock.RLock()
lock.RUnlock()

lock.Lock()
lock.Unlock()

3.3 Cond

イベントの待機や発生を知らせる。

c := sync.NewCond(&sync.Mutex{})

func f() {
    c.L.Lock()
    // クリティカルセクションを操作する処理
    c.L.Unlock()
    c.Signal()
}

func main() {
    c.L.Lock()
    for conditionTrue() == false {
        c.Wait()
    }
    // クリティカルセクションを操作する処理
    go f()
    c.L.Unlock()
}

sync - Go 言語において、Waitは

自動的に c.L のロックを解除し，呼び出し側のゴルーチンの実行を中断します。 後で実行を再開した後， Wait は c.L をロックしてから戻ります。

これを上の例で見ると以下のようになる。

  1. conditionTrue()がfalseになったら、ループに入り、c.Wait()を呼び出すことで呼び出し側(main関数)の処理を一旦停止し、c.Lのロックを解除する。
  2. 関数fでクリティカルセクションを操作する処理が行われ、c.Signal()を実行することで、c.Wait()で停止されていたmain関数の処理が再開される。その際、main関数でc.Lが再びロックされる。
  3. conditionTrue()がfalseのままであったら、1に戻り、trueであったらループから抜け出し、その後の処理を行う。

c.Broadcast()ではc.Wait()となっているすべてのゴルーチンにシグナルを伝える。

3.2.4 Once

sync.Once.Doは一度しか呼び出されない。

Pool

使うものを決まった分だけ作る方法。 Getメソッドではプールに必要なインスタンスがあるか確認し、あれば呼び出し元にそれを返す。なければ、Newを呼び出した結果を返す。作業が終わるとPutメソッドを用いて使っていたインスタンスをプールに返す。

[Python] Import cyclesを解消する

python

はじめに

テスト駆動開発のjavaで書かれた部分をpythonで実装しているとき、

ImportError: cannot import name 'Money' from partially initialized module 'moneys.money' (most likely due to a circular import)

というエラーに遭遇しました。 これは循環参照と呼ばれるものであり、

# money.py
from __future__ import annotations
from .expression import Expression

class Money(Expression):
    ...
    def plus(self, added: Money) -> Expression:
        from .total import Total
        return Total(self, added)
   ...

# expression.py
from __future__ import annotations
from abc import ABCMeta, abstractmethod
from .money import Money


class Expression(metaclass=ABCMeta):
    @abstractmethod
    def reduce(self, to: str) -> Money:
        pass

というような形で、money.pyでExpressionをインポートしているのに、expression.pyではMoneyをインポートするという循環参照と呼ばれる状態に陥っているため生じているエラーです。

このエラーはfrom typing import TYPE_CHECKINGを用いて解消できます。

解消法

# expression.py
from __future__ import annotations
from abc import ABCMeta, abstractmethod

# 追加
if TYPE_CHECKING:
    from .money import Money

class Expression(metaclass=ABCMeta):
    @abstractmethod
    def reduce(self, to: str) -> Money:
        pass

とするだけでエラーは解消されます。

より詳しく

どうしてこの2行のコードで解消されるのかをみていきます。

string literal types

pythonの関数で型を指定するとき、その型をstring型で書くことができます。

例えば、下のコードではdef f(a: A) -> None: ...と引数aの型をAとしてしまうと、pythonはまだclass Aを読み込んでいないことからエラーを出します。 しかし、引数の型をstring型の'A'としてやるとエラーを解消できます。

def f(a: 'A') -> None: ...

class A:
    pass

annotations

同様のことがfrom __future__ import annotationsにより可能となります。

以下のコードでは関数fの引数はstring型ではありませんが、annotationsをインポートしていることでエラーなしで実行できます。これは自動で型アノテーションをstring literal化するようなものだとドキュメントに書かれています。

from __future__ import annotations
def f(a: A) -> None: ...

class A:
    pass

typing.TYPE_CHECKING

typingモジュールはTYPE_CHECKINGという変数を定義しており、この変数はランタイム中にはTYPE_CHECKING=Falseであり、型チェック中にはTYPE_CHECKING=Trueとなります。

このことから、if TYPE_CHECKING:内に循環参照の原因となるモジュールをインポートするとランタイム中には実行されないため、エラーを起こさず実行することができます。

したがって、解消法で示したように循環参照のもととなっているモジュールのインポートをif TYPE_CHECKING:内に書くことでエラーが解消できることがわかります。

最後に

参考にしたドキュメントは以下になります。 mypy.readthedocs.io

こちらはTDDの本のコードをpythonで実装したときのレポジトリです。 github.com