mypyで静的型チェックを導入する

仕事で既存のコードへmypyの導入を試みる機会がありましたので、使い方とtipsの備忘録としてまとめます。

mypyとは

mypyはPythonで静的型チェックを行うライブラリです。型はtypingで定義します。

インストール

typingと異なり、標準ライブラリではないので、インストールが必要です。

$ python --version
Python 3.7.4

$ pip install mypy

$ mypy --version
mypy 0.782

型チェックの実行

mypyで型チェックできます。以下ではsrcディレクトリ以下の型チェックが走ります。既存のプロジェクトへのmypy導入時には大量のエラー・警告メッセージが表示されるかと思います。

$ mypy src

Tips: 型チェック対象を限定する

コマンドライン引数でファイルやディレクトリ単位の指定が可能です。

$ mypy src/common src/models src/services

これらの型チェック対象をローカル環境やCI環境で共有する場合、以下のように列挙したテキストファイルを作成しておきます。mypy @{作成したファイルパス}とすることでテキストファイルで指定したファイル・ディレクトリのみがチェックされます。

$ cat mypy_files.txt
src/common
src/models
src/services

$ mypy @mypy_files.txt

pipenvの場合、Pipfileに以下のように定義しておくと楽になるかと思います。

[scripts]
mypy = "mypy @mypy_files.txt"

設定

mypyはカレントディレクトリのmypy.ini (あるいはsetup.cfg) で設定します。詳細なパラメータはドキュメントを参照してください。

  • warn_retun_any = Trueとすることで、返す型を明示していないインタフェースに警告させてます (デフォルトはFalse)
[mypy]
python_version = 3.7
warn_return_any = True
warn_unused_configs = True

Tips: スタブが無いパッケージのエラーは無視させる

よく見るエラーの1つがerror: Skipping analyzing '{パッケージ名, 例えばnumpy}': found module but no type hints or library stubsです。これは型情報を持つstubファイル (.pyi) を含まないパッケージをインポートすると起きます。
サードパーティのライブラリだといちいち追加するのも大変なので、無視するようにパッケージ単位でignore_missing_imports = Trueを追加します。

[mypy-numpy]
ignore_missing_imports = True

[mypy-scipy.*]
ignore_missing_imports = True

コードの修正

mypyで指摘された事項を修正していきます。基本的には、正しく型アノテーションを付けることで解消していきます。

返り値がintでアノテーションされてますが、Noneを返すケースもあります。

import re

def atoi(a: str) -> int:
    if re.match(r"[+-]?[0-9]+", a):
        return int(a)
    else:
        return None

mypyでそれが指摘されます。

% mypy src
src/main.py:7: error: Incompatible return value type (got "None", expected "int")
Found 1 error in 1 file (checked 1 source file)

なのでtypingを使って書き換えていきます。

from typing import 

def atoi(a: str) -> Optional[int]:
    # ...

基本はこの繰り返しになりますが、いくつか対処に困るケースもあります。

Tips: Unionでの型エラー

Union型の変数を、いずれかの型であることを前提とした処理を書くことがしばしばあります。例えば以下のコードでは、辞書のキーが "_list" の場合はlist、そうでなければintとして分けています。

from typing import Dict, Union, List

def sum_dict(d: Dict[str, Union[int, List[int]]]) -> int:
    total = 0
    for k, v in d.items():
        if k.endswith("_list"):
            total += sum(v)
        else:
            total += v
    return total

total = sum_dict({
    "ShopA": 100,
    "ShopB_list": [200, 300]
})

print(total)  # 600

問題なく実行もできるのですが、mypyではエラーとなってしまいます。

% mypy src
src/main.py:7: error: Argument 1 to "sum" has incompatible type "Union[int, List[int]]"; expected "Iterable[int]"
src/main.py:9: error: Unsupported operand types for + ("int" and "List[int]")
src/main.py:9: note: Right operand is of type "Union[int, List[int]]"

解決策が2つあります。

1つ目はtyping.castを使って、チェック時に型を教えてあげる方法です。値の型は変わりません。

from typing import Dict, Union, List, cast

def sum_dict(d: Dict[str, Union[int, List[int]]]) -> int:
    total = 0
    for k, v in d.items():
        if k.endswith("_list"):
            total += sum(cast(List[int], v))
        else:
            total += cast(int, v)
    return total

2つ目はisinstanceで明示的に処理分岐させる方法です。こうするとmypyは分岐を解析し、ifの方はlist、elseの方はintで解釈するようになります。

def sum_dict(d: Dict[str, Union[int, List[int]]]) -> int:
    total = 0
    for k, v in d.items():
        if isinstance(v, list):
            total += sum(v)
        else:
            total += v
    return total

個人的には、可能であればisinstanceで実行時の型チェック、それも難しい場合にtyping.castを使うのが安全かなと思います。

Tips: 特定箇所の型チェックエラーを無視したい

どうしても型チェックエラーを無視したいケースもあります。インタフェースを簡単に変えられないケースなどです。

エラーとなっている箇所に# type: ignoreとコメントすることで除外できます。

import re

def atoi(a: str) -> int:
    if re.match(r"[+-]?[0-9]+", a):
        return int(a)
    else:
        return None  # type: ignore

まとめ

今回はmypyの使い方とTipsを紹介しました。

AWS CodeBuildでDockerビルドしてECRへプッシュする

以前AWS CodeBuildでPythonアプリケーションをビルドしてS3へアップロードする方法を紹介しました。

今回はDockerイメージをビルドして、ECRにプッシュする方法について見ていきたいと思います。

サンプル

サンプルとして以前作成したPythonアプリケーションを使います。これに簡単なDockerfileを追加します。

f:id:ohke:20200926153719p:plain

FROM python:3.8

WORKDIR /root

RUN pip install pipenv==2018.11.26

COPY ./Pipfile Pipfile
COPY ./Pipfile.lock Pipfile.lock

RUN pipenv install --system

COPY . /root

buildspec.yml

CodeBuildのパイプラインはbuildspec.ymlで定義します。各フェーズで以下の処理を行います。

  • pre_buildでテスト
  • buildでイメージビルド
  • post_buildでECRへプッシュ
version: 0.2

phases:
  install:
    runtime-versions:
      python: 3.8
    commands:
      - pip install pipenv==2018.11.26
  pre_build:
    commands:
      - pipenv install --dev
      - pipenv run flake8
      - pipenv run pytest
  build:
    commands:
      - docker build -t codebuild-test .
  post_build:
    commands:
      - aws ecr get-login-password --region ap-northeast-1 | docker login --username AWS --password-stdin 000000000000.dkr.ecr.ap-northeast-1.amazonaws.com
      - docker tag codebuild-test:latest 000000000000.dkr.ecr.ap-northeast-1.amazonaws.com/codebuild-test:latest
      - docker push 000000000000.dkr.ecr.ap-northeast-1.amazonaws.com/codebuild-test:latest

AWS CodeBuildの設定

次にAWSコンソール画面からCodeBuildのビルドプロジェクトを設定をしていきます。前回とほぼ同じ手続きでOKですが、2点ばかり変更が必要です。

  • Dockerビルドするので、特権 (privileged) を有効にする必要があります
  • ECRへにログインおよびプッシュするので、サービスロールにECRのポリシーを付与する必要があります

f:id:ohke:20200926155644p:plain

ビルドを実行するとこんな感じのログが出力され、ECRへプッシュされます。

[Container] 2020/09/26 05:44:49 Waiting for agent ping
[Container] 2020/09/26 05:44:51 Waiting for DOWNLOAD_SOURCE
[Container] 2020/09/26 05:44:54 Phase is DOWNLOAD_SOURCE
[Container] 2020/09/26 05:44:54 CODEBUILD_SRC_DIR=/codebuild/output/src553119189/src/github.com/ohke/codebuild-test
[Container] 2020/09/26 05:44:54 YAML location is /codebuild/output/src553119189/src/github.com/ohke/codebuild-test/buildspec.yml
[Container] 2020/09/26 05:44:54 Processing environment variables
[Container] 2020/09/26 05:44:54 No runtime version selected in buildspec.
[Container] 2020/09/26 05:44:54 Moving to directory /codebuild/output/src553119189/src/github.com/ohke/codebuild-test
[Container] 2020/09/26 05:44:54 Registering with agent
[Container] 2020/09/26 05:44:54 Phases found in YAML: 1
[Container] 2020/09/26 05:44:54  BUILD: 1 commands
[Container] 2020/09/26 05:44:54 Phase complete: DOWNLOAD_SOURCE State: SUCCEEDED
[Container] 2020/09/26 05:44:54 Phase context status code:  Message: 
[Container] 2020/09/26 05:44:54 Entering phase INSTALL
[Container] 2020/09/26 05:44:54 Phase complete: INSTALL State: SUCCEEDED
[Container] 2020/09/26 05:44:54 Phase context status code:  Message: 
[Container] 2020/09/26 05:44:54 Entering phase PRE_BUILD
[Container] 2020/09/26 05:44:54 Phase complete: PRE_BUILD State: SUCCEEDED
[Container] 2020/09/26 05:44:54 Phase context status code:  Message: 
[Container] 2020/09/26 05:44:54 Entering phase BUILD
[Container] 2020/09/26 05:44:54 Running command docker build .
Sending build context to Docker daemon  43.01kB

Step 1/7 : FROM python:3.8
3.8: Pulling from library/python
...
Status: Downloaded newer image for python:3.8
 ---> bbf31371d67d
Step 2/7 : WORKDIR /root
 ---> Running in 8bf52f1e4fe6
Removing intermediate container 8bf52f1e4fe6
 ---> d0abc177c2e9
Step 3/7 : RUN pip install pipenv==2018.11.26
 ---> Running in 995508830aa0
Collecting pipenv==2018.11.26
  Downloading pipenv-2018.11.26-py3-none-any.whl (5.2 MB)
...
Installing collected packages: virtualenv-clone, certifi, distlib, appdirs, six, filelock, virtualenv, pipenv
Successfully installed appdirs-1.4.4 certifi-2020.6.20 distlib-0.3.1 filelock-3.0.12 pipenv-2018.11.26 six-1.15.0 virtualenv-20.0.31 virtualenv-clone-0.5.4
Removing intermediate container 995508830aa0
 ---> 5bb947b68efe
Step 4/7 : COPY ./Pipfile Pipfile
 ---> 539480213459
Step 5/7 : COPY ./Pipfile.lock Pipfile.lock
 ---> 314848853e56
Step 6/7 : RUN pipenv install --system
 ---> Running in 2d266779cfb6
Installing dependencies from Pipfile.lock (731881)…
Removing intermediate container 2d266779cfb6
 ---> 997f1bf10f60
Step 7/7 : COPY . /root
 ---> 7e6aea040923
Successfully built 7e6aea040923

[Container] 2020/09/26 05:45:31 Phase complete: BUILD State: SUCCEEDED
[Container] 2020/09/26 05:45:31 Phase context status code:  Message: 
[Container] 2020/09/26 05:45:31 Entering phase POST_BUILD

f:id:ohke:20200926160925p:plain

イメージのレイヤキャッシュ

上の設定では常にフルビルドが走ります。これでは実行のたびに時間がかかってしまいますので、Dockerのレイヤキャッシュを行います。

CodeBuildのドキュメントではローカルキャッシュの一機能としてレイヤキャッシュもサポートされているとのことだったのですが、設定を有効にしても私の環境ではキャッシュしてくれませんでした。

代替策として、ビルド前にECRからイメージをpullし、--cache-fromオプションでpullしたイメージを指定することができます。リモートレポジトリから都度pullしてくるのでその分の通信時間はかかりますが、永続化されたイメージをキャッシュとして利用できるため、ビルドの間隔が空いてもキャッシュの恩恵を受けることができます。 (ローカルキャッシュは一定時間で揮発するので、間隔が空くとフルビルドが走ってしまう。)

version: 0.2

phases:
  install:
    runtime-versions:
      python: 3.8
    commands:
      - pip install pipenv==2018.11.26
  pre_build:
    commands:
      - pipenv install --dev
      - pipenv run flake8
      - pipenv run pytest
  build:
    commands:
      - $(aws ecr get-login --no-include-email --region ap-northeast-1)
      - docker pull 000000000000.dkr.ecr.ap-northeast-1.amazonaws.com/codebuild-test:latest || true
      - docker build --cache-from 000000000000.dkr.ecr.ap-northeast-1.amazonaws.com/codebuild-test:latest --tag 000000000000.dkr.ecr.ap-northeast-1.amazonaws.com/codebuild-test:latest .
  post_build:
    commands:
      - aws ecr get-login-password --region ap-northeast-1 | docker login --username AWS --password-stdin 000000000000.dkr.ecr.ap-northeast-1.amazonaws.com
      - docker push 000000000000.dkr.ecr.ap-northeast-1.amazonaws.com/codebuild-test:latest

main.pyを修正してCodeBuildを実行すると、ログにStep 6/7までキャッシュが利用されていることが確認できます。

Sending build context to Docker daemon  67.07kB

Step 1/7 : FROM python:3.8
3.8: Pulling from library/python
57df1a1f1ad8: Already exists
71e126169501: Already exists
1af28a55c3f3: Already exists
03f1c9932170: Already exists
65b3db15f518: Already exists
3e3b8947ed83: Already exists
f156949921a1: Already exists
1c1931013093: Already exists
51fff639b6bf: Already exists
Digest: sha256:1a126607adde46a706e76357c910f36b9f5529fb575d4d86a639a4997daceba7
Status: Downloaded newer image for python:3.8
 ---> bbf31371d67d
Step 2/7 : WORKDIR /root
 ---> Using cache
 ---> eb39a705a351
Step 3/7 : RUN pip install pipenv==2018.11.26
 ---> Using cache
 ---> c39e0cb776bd
Step 4/7 : COPY ./Pipfile Pipfile
 ---> Using cache
 ---> 5fef8209ac4d
Step 5/7 : COPY ./Pipfile.lock Pipfile.lock
 ---> Using cache
 ---> dea18f792eea
Step 6/7 : RUN pipenv install --system
 ---> Using cache
 ---> d590683b4da4
Step 7/7 : COPY . /root
 ---> be6b7aed2b45
Successfully built be6b7aed2b45
Successfully tagged 000000000000.dkr.ecr.ap-northeast-1.amazonaws.com/codebuild-test:latest

(参考) ビルド時に変数を渡す

Dockerビルドに限った話ではないのですが、CodeBuildの実行時に変数 (例えばイメージタグなど) を外部から渡したい場合があります。これは環境変数にセットすることでコントロールできます。

コンソールから実行する場合は以下のようにセットします。

  • AWS CLIの場合は start-build の--environment-variables-override オプションで指定できます (参考)

f:id:ohke:20200926165510p:plain

buildspec.ymlは${IMAGE_TAG}で参照します。これで、latestに加えてIMAGE_TAGで指定した値 (ここではstg) のタグでプッシュされます。

  build:
    commands:
      - $(aws ecr get-login --no-include-email --region ap-northeast-1)
      - docker pull 000000000000.dkr.ecr.ap-northeast-1.amazonaws.com/codebuild-test:latest || true
      - docker build --cache-from 279213333729.dkr.ecr.ap-northeast-1.amazonaws.com/codebuild-test:latest --tag 000000000000.dkr.ecr.ap-northeast-1.amazonaws.com/codebuild-test:latest --tag 000000000000.dkr.ecr.ap-northeast-1.amazonaws.com/codebuild-test:${IMAGE_TAG} .
  post_build:
    commands:
      - aws ecr get-login-password --region ap-northeast-1 | docker login --username AWS --password-stdin 000000000000.dkr.ecr.ap-northeast-1.amazonaws.com
      - docker push 000000000000.dkr.ecr.ap-northeast-1.amazonaws.com/codebuild-test:latest
      - docker push 000000000000.dkr.ecr.ap-northeast-1.amazonaws.com/codebuild-test:${IMAGE_TAG}

まとめ

AWS CodeBuildを使ってDockerビルドを実行する方法について紹介しました。

sqlparseを使って簡単にSQLをパースする

大量のSQLのリストやログから特定テーブルへのUPDATEを行うクエリだけを抽出したいといったケースが、しばしばあります。1つのアプリケーションからアクセスされるDBであればコードを追っていくことでどうにかなるのですが、異なる言語やORMで構成された複数のアプリケーションで共用していたり、エンジニアやアナリストがアドホックにクエリを実行していたりすると、なかなか厄介です。

  • 予約語を大文字にしたり小文字にしたり、改行やインデントなど、フォーマットがまちまち
  • サブクエリを含むと "どのテーブルに対する操作か" が、単純な文字列マッチ (字句解析) などではわからない

PythonでSQLを簡単にパースするライブラリとして sqlparse を紹介します。

$ python --version
Python 3.6.9

$ pip list | grep sqlparse
sqlparse                      0.3.1

sqlparseのsplitメソッドでステートメントの分割、formatメソッドで整形などの簡単な処理ができます。

import sqlparse

# ステートメントの分割
statements = sqlparse.split("""
select id, name from customers where deleted = 0;
select created_at where id = 10
""")

print(statements)
# ['select id, name from customers where deleted = 0;', 'select created_at where id = 10']
print(type(statements[0])  # <class 'str'>

# 整形
statement = sqlparse.format(statements[0], reindent=True, keyword_case="upper")

print(statement)
# SELECT id,
#        name
# FROM customers
# WHERE deleted = 0;

さらにparseでトークン解析もできます。parseではStatementインスタンスが返されます。このインスタンスのtokensプロパティがTokenのリストとなってます。Tokenは再帰的な木構造になってます。

  • Token.ttypeにDML or DLL、Where、Keywordなどの種類が入ってます
  • Tokenにもtokensプロパティを持ちます
    • 以下の例ではIdentifierListのが4つのトークン ("id", ",", " ", "name") を持っています
    • parentも持っており、上にたどることもできます
parsed_statements = sqlparse.parse("""
select id, name from customers where deleted = 0;
select created_at where id = 10
""")
pprint(parsed_statements)
# (<Statement ' selec...' at 0x7F39CE745390>,
#  <Statement ' selec...' at 0x7F39CE669ED0>)

parsed_statement = parsed_statements[0]
pprint(parsed_statement.tokens)
# [<Newline ' ' at 0x7F39CE6FAC48>,
#  <DML 'select' at 0x7F39CE6D13A8>,
#  <Whitespace ' ' at 0x7F39CE6D1048>,
#  <IdentifierList 'id, na...' at 0x7F39CE6D8660>,
#  <Whitespace ' ' at 0x7F39CE6D1288>,
#  <Keyword 'from' at 0x7F39CE6D12E8>,
#  <Whitespace ' ' at 0x7F39CE6D1348>,
#  <Identifier 'custom...' at 0x7F39CE6D85E8>,
#  <Whitespace ' ' at 0x7F39CE7398E8>,
#  <Where 'where ...' at 0x7F39CE6D8570>]

print(type(parsed_statement.tokens[0]))
# <class 'sqlparse.sql.Token'>

pprint(parsed_statement.tokens[3].tokens)
# [<Identifier 'id' at 0x7F39CE69BD68>,
#  <Punctuation ',' at 0x7F39CE69DA08>,
#  <Whitespace ' ' at 0x7F39CE699BE8>,
#  <Identifier 'name' at 0x7F39CE69B840>]