mizzsugar’s blog

Pythonで学んだことや読書録を書きます。

ざっくりPyConJP2019の1日目感想

今年もPyConJPに参加しました!

1日目に拝聴したセッションの感想をざっくり書き残そうと思います。

なお、今日の発表の様子はYouTubeにすべて上がっています。

t.co

個人的に、YouTube版はちゃんと聞き取れるしスライドの内容もちゃんと見えるので良かったです!


Djangoで実践ドメイン駆動設計による実装

Twitter

twitter.com

YouTube

youtu.be

めちゃめちゃ大きいテーマに挑戦されていて、尊敬します。

DDDの考え方もしっかり説明してくださってありがたかったです。Pythonで具体的なコードまで載せており、こういう実装方法もあるんだと学びになりました。

「あれ? 結局これ取り入れてで何を解決したいんだっけ?」とならないように

戦略的DDDをしっかり行うという前提でDjangoでの戦術的DDDに挑戦したいなあと身が引き締まりました。


Pythonを使った APIサーバー開発を始める際に 整備したCIとテスト機構

Twitter

twitter.com

スライド

speakerdeck.com

「CIに教わる」という名言が印象的でした。

モックにどれだけたよっているか、どんなふうにたよっているかは

きれいにDIできていて修正時の影響範囲を抑えたコードになっているかを確かめる観点ということで

改めて自分のテストコードを見て確認したいと思いました。

Pythonのモック、なんでもモックできちゃうので

「ただモックした!」から前進するヒントになりそうな発表でした。

YouTube

youtu.be


ListはIteratorですか?

Twitter

twitter.com

スライド

docs.google.com

YouTube

youtu.be


とても楽しそうにお話していたのが印象的でした。

スライドにあるListやSetやIterableなどの各クラスの継承関係の図はありそうで今までないやつだったので感動しました・・・! あれは重宝しそう・・・!

あの図を見ながらなら、広く引数の型を受け入れる使いやすい関数を書けそうな気がします。

初学者にもわかりやすくも、継承関係や豆知識で深いところまで扱っていて面白かったです。


[おまけ]セッション以外で感じたこと

去年の初参加のときはPython始めたばかりで全然知り合いがいなかったのですが、

今年は同じ会社の人がいたり勉強会で一緒だったorTwitterでやりとりしていた方など

いろんな人とお話できて楽しかったです!

知り合いがまったくいなかった去年もトークを聞くだけでも十分楽しかったですが、

いろんな人がいっしょくたに集まるのは感慨深いですね。

アフターパーティで私の友達が私の会社の人と友達になる(友達の友達は友達的なやつ?)というのもこういうイベントならではなのかな〜など月並にも思いました。

【Python】Poetryでパッケージ管理 on Docker Container

DockerでPython開発する時、パッケージ管理はpip install -r requirement.txtで行なっていましたが、

requirement.txtのみではパッケージ同士の依存関係まで管理できないことが問題でした。

そこで、Poetryを使ってみることにしました。

github.com

使用技術

- Python 3.7.3
- Docker 19.03.1
- poetry 0.12.17

ディレクトリ構成です。 今回は、Djangoプロジェクトをappというコンテナに、DBをdbというコンテナで管理することにしました。

poetry.lockとpyproject.tomlをどこに入れるか迷いましたが、Pythonのコードを管理しているapp配下に。


環境関連は全てdocker以下に!ということでdocker/appに入れてみたのですが、新しいパッケージを導入したいと思うたびに

appからdockerに移動しないといけないのが面倒だと気づいてしまいました汗

Dockerコンテナに入って開発する際にworkdirがapp以下になるだろうということで、 app以下にこの2つを置いてpoetryの操作をしやすいようにしました。

ディレクトリ構成

├── app
│   ├── poetry.lock
│   ├── project
│   │   ├── config
│   │   │   ├── __init__.py
│   │   │   ├── manage.py
│   │   │   ├── settings
│   │   │   │   ├── __init__.py
│   │   │   │   └── base.py
│   │   │   ├── urls.py
│   │   │   └── wsgi.py
│   │   ├── manage.py
│   │   ├── sample_app
│   │   │   ├── __init__.py
│   │   │   ├── admin.py
│   │   │   ├── apps.py
│   │   │   ├── migrations
│   │   │   │   ├── __init__.py
│   │   │   ├── models.py
│   │   │   ├── urls.py
│   │   │   └── views.py
│   │   └── tests
│   │       └── __init__.py
│   └── pyproject.toml
├── docker
│   ├── app
│   │   ├── Dockerfile
│   │   └── start-server.sh
│   ├── etc
│   │   └── gunicorn.conf
│   ├── db
│   │   └── Dockerfile
│   └── var
│       └── log
│           └── gunicorn
└── docker-compose.yaml


PythonコンテナのDockerfileです。

poetryのインストールのみにとどめ、 poetry installアプリケーションサーバーの立ち上げは 独自にシェルスクリプトを作成して、そちらに実行させることにしました。

Dockerfile

FROM python:latest
ENV PYTHONUNBUFFERED 1
LABEL maintainer "sample_name <hogehoge@example.com>"
RUN pip install --upgrade pip
RUN pip install poetry


シェルスクリプトの中身をみましょう。

Dockerコンテナを立ち上げる際に下記のコマンドが実行されるようにしました。 serverも立ち上がるようにし、コンテナが立ち上がった後にlocalhost:8080にアクセスするとDjangoで作成した画面が見れ流ようにしました。

本番環境と開発環境で別コマンドが実行されるようにします。

#!/bin/bash
cd /home/app

# /home/app下にpoetry.lockとpyproject.tomlがあるのでインストールします。
poetry install
cd project


# poetry run [実行したいコマンド]でpipenvで作成される仮想環境でコマンドを実行します。
if [ "${DJANGO_ENV}" = 'production' ]; then
    # 本番環境
    poetry run python manage.py migrate --settings config.settings
    poetry run python manage.py collectstatic --noinput
    poetry run gunicorn config.wsgi:application -c /home/docker/etc/gunicorn.conf -b :8080
else
 # 開発環境
    poetry run python manage.py migrate
    poetry run python manage.py runserver 0.0.0.0:8000
fi


docker-compose.yaml

version: "3"
services:
  app:
    build:
      context: ./docker/app
      dockerfile: Dockerfile
    env_file: .env
    environment:
      - POSTGRESQL_HOST=db
      - POSTGRESQL_PORT=5432
    volumes:
      - ./app:/home/app
      - ./docker/app:/home/docker
      - ./docker/etc/gunicorn.conf:/etc/gunicorn.conf
      - ./docker/var/log/gunicorn:/var/log/gunicorn
    working_dir: /home/app
    entrypoint: "/bin/sh"
    command: "/home/docker/start-server.sh"
    ports:
      - "8080:8080"
    tty: true
    depends_on:
      - db
  db:
    build:
      context: ./docker/postgres
      dockerfile: Dockerfile
    env_file: .env
    ports:
      - 5432:5432
    volumes:
      - ./docker/db/pgsql-data:/var/lib/postgresql/data


つづいて、操作方法について。

なんかしらPythonの操作したい

まずコンテナに入ります。

docker-compose exec app bash

poetryの仮想環境下で操作する必要があるので、 まずpoetry shellでpoetryが作成する仮想環境に入ってからコマンドを実行

poetry shell

[実行したいコマンド]

またはこちらで。 poetry runを頭につけると、poetryの仮想環境に入った状態でコマンドが実行される動きになります。

poetry run [実行したいコマンド]

新しくパッケージを導入したい

docker-compose exec app bash
# poetryにてインストール
poetry add [パッケージ名]

パッケージのバージョンをあげたい

docker-compose exec app bash
# 全部のパッケージをアップデートしたい時
poetry update

# 特定のパッケージのみアップデートしたい時
poetry update [パッケージ名]


リポジトリはこちら。

github.com


おまけ。Pipenv編リポジトリ

(やることはあまり変わらなかった)

GitHub - mizzsugar/pipenv_on_docker

【Django】独自ヘッダーをつけてリクエストを送る

地味に苦労したやつです。

利用技術

- python 3.7.3

- Django 2.2.1

ヘッダーに関して、Djangoの公式ドキュメントはなんて言っているでしょうか。

HttpRequest.META
利用できるすべての HTTP ヘッダーが格納されたディクショナリです。
-- Django公式ドキュメントより

https://docs.djangoproject.com/ja/2.2/ref/request-response/#django.http.HttpRequest.META


なるほど、request.METAから取り出せば良いのか。

curl -X POST -H 'Content-Type:application/json' -H "CUSTOM_HEADER:customheader" http://127.0.0.1/api/


def sample_view(request):
    print(request.META.get('CUSTOM_HEADER')
    ...

printの中身がNoneになる・・・ なぜや〜


もうちょっと公式ドキュメントを読むと・・・

With the exception of CONTENT_LENGTH and CONTENT_TYPE, as given above,
any HTTP headers in the request are converted to META keys by converting all characters to uppercase,
replacing any hyphens with underscores and adding an HTTP_ prefix to the name. 
So, for example, a header called X-Bender would be mapped to the META key HTTP_X_BENDER.

-- Django公式ドキュメントより

https://docs.djangoproject.com/ja/2.2/ref/request-response/#django.http.HttpRequest.META


CONTENT_LENGTHとCONTENT_TYPEは例外として、上記に示されているように、
リクエストのHTTPヘッダーはMETAキーに変換されます。全て大文字になり、ハイフンがアンダースコアになり、ヘッダー名の頭に「HTTP_」がつきます。
例えば、「X-Bender」というヘッダーは「HTTP_X_BENDER」というMETAキーに格納されます。
-- mizzsugarによる意訳

request.METAに変換する際にヘッダー名が変換されるということっぽい。

これならいけるか?

request.META.get('HTTP_CUSTOM_HEADER')

ダメだった・・・

「ハイフンがアンダースコアになり」。ということは、アンダースコアはダメ?

これでどうだ↓

curl -X POST -H 'Content-Type:application/json' -H "CUSTOM-HEADER:customheader" http://127.0.0.1/api/


request.META.get('HTTP_CUSTOM_HEADER')


取得できました!

どうやら、こういう仕様のようです。

* 名前に区切りを使いたいならハイフンを使わないといけない
* request.METAから取り出す場合は全て大文字にして、名前の頭に「HTTP_」をつけ、ハイフンの箇所をアンダースコアにする

ハイフンじゃないと反応しないというところが個人的にハマりポイントでした(^^;

【Django】複数のファイルをまとめてテストするとTransactionErrorやIntegurityErrorになってしまう事件について

仕事でテスト周りについて色々あったので備忘録として

例えば、こういうViewとModelがあったとして・・・

(まあ、こんなこと普通しないとは思いつつ簡単な例を出したく)

models.py

from typing import Dict

from django.db import models
from django.db import IntegrityError


class Customer(models.Model):
    GENDER_CHOICES = (
        (0, 'Male'),
        (1, 'Female'),
        (2, 'Other')
    )
    name = models.TextField()
    email = models.EmailField(unique=True)
    birthday = models.DateField()
    gender = models.PositiveSmallIntegerField(choices=GENDER_CHOICES)


    @classmethod
    def create(cls, data: Dict) -> None:
        """フォームのデータから顧客を新規登録するメソッド
        
        dataの中身
        {
            'name': 'name',
            'email': 'email@example.com',
            'birthday': datetime.date(1990, 1, 1),
            'gender': 0
        }
        """
        try:
            cls.objects.create(
                name=data.get('name'),
                email=data.get('email'),
                birthday=data.get('birthday'),
                gender=data.get('gender')
            )
        except IntegrityError:
            raise

views.py

from django.http import HttpResponse
from django.views import View
from django.shortcuts import render
from django import forms
from django.db import IntegrityError

from .models import Customer

# ブログを簡素にしたいのでViewに書いたけどforms.pyに書いてもよし
class CustomerForm(forms.Form):
    GENDER_CHOICES = (
        (0, 'Male'),
        (1, 'Female'),
        (2, 'Other')
    )
    name = forms.CharField(max_length=255)
    email = forms.EmailField()
    gender = forms.ChoiceField(choices=GENDER_CHOICES)


class CreateCustomerView(View):
    def get(self, request):
        """顧客名一覧を表示
        """
        costomers = [
            customer.name
            for customer in Customer.objects.iterator()
        ]
        return render(
            request,
            'form.html',
            {
                'costomers': costomers,
                'form': CustomerForm()
            }
        )

    def post(self, request):
        """顧客を新規登録

        request.postの形式
        {
            "name": "Taro Tanaka",
            "birthday": "1980-01-01",
            "gender": 0
        }
        """
        form = CustomerForm(request.POST)
        if form.is_valid():
            try:
                Customer.create(data=form.cleaned_data)
            except IntegrityError:
                form.add('email', 'すでに登録しているメールアドレスは登録できません')
        return render(
            requset,
            'form.html',
            {
                'costomers': costomers,
                'form': form
            }
        )

こうテストを書くと

test_models.py

import datetime

from django.db import IntegrityError
from django.test import TestCase

from .models import Customer


class TestCustomerModel(TestCase):
    @classmethod
    def setUpTestData(cls):
        Customer.objects.create(
            name='Taro Yamada',
            email='sample@example.com',
            birthday=datetime.date(1980, 1, 1),
            gender=0
        )

    def test_email_duplicate(self):
        data = {
            name: 'Jiro Yamada',
            email: 'sample@example.com',  # すでに登録しているメールアドレス
            birthday: datetime.date(1980, 2, 2),
            gender: 0
        }
        with self.assertRaises(IntegrityError):
            Customer.create(data=data)

    def test_create_valid_customer(self):
        data = {
            name: 'Hanako Takada',
            email: 'sample_2@example.com',
            birthday: datetime.date(1980, 3, 3),
            gender: 1
        }
        Customer.create(data=data)

test_views.py

from django.test import TestCase, Client

from .models import Customer
from .views import CreateCustomerView


class TestCustomerView(TestCase):
    @classmethod
    def setUpTestData(cls):
        Customer.objects.create(
            name='Taro Yamada',
            email='sample@example.com',
            birthday=datetime.date(1980, 1, 1),
            gender=0
        )

    def test_email_duplicate(self):
        data = {
            name: 'Jiro Yamada',
            email: 'sample@example.com',  # すでに登録しているメールアドレス
            birthday: datetime.date(1980, 2, 2),
            gender: 0
        }
        response = self.client.post(
            '',
            {
                "name": "Taro Yamada",
                "email": "sample@example.com",
                "gender": "0",
                "birthday": "1980-01-01",
                "position": "100",
            }
        )
        self.assertContains(
            'すでに登録しているメールアドレスは登録できません'
            response
        )

TransactionErrorが〜

IntegurityErrorが〜


・・・なぜ起こるか。

TestCaseを継承するクラスにsetUpClassとtearDownClassがないことに注目。

Djangoでは、setUpClassがそのテストクラスでテストを実行するための準備をするために、テストクラスが初期化される際に一度だけ呼ばれます。

tearDownClassがテストクラス内のテストがすべて終わりテストクラスが解放される際に一度だけ呼ばれる仕組みになっています。tearDownClassはsetUpClassが行った準備をすべてクリーンにするイメージかと思います。

DjangoのTestCaseを継承している場合、tearDownClassによって、トランザクションロールバックされます。テスト用データベース内のデータがすべて一掃されます。

上記の例だと、

各テストクラスにsetUpClassとtearDownClassがないため、データベースのトランザクションが終わってない状態であるがゆえにTransactionError

最初に実行したテストクラスでのデータが残っているためユニークであるはずのデータが重複しようとしてしまい意図しないIntegurityError・・・

となってしまいました。

じゃあこうすればいいの!?  

class TestCustomerModel(TestCase):
    @classmethod
    def setUpClass(cls):
        pass

# 中略
...
    
    @classmethod
    def tearDownClass(cls):
        pass

いいえ、違います。

ただ最初にsetUpClassとtearDownClassを追加しただけだとトランザクション開始と後処理の働きをしないので super()をつけることによってDjangoのTestCaseのsetUpClassとtearDownClassの働きを継承してあげましょう。

class TestCustomerModel(TestCase):
    @classmethod
    def setUpClass(cls):
        super().setUpClass()

# 中略
...
    
    @classmethod
    def tearDownClass(cls):
        super().tearDownClass()

他にもテストクラス内のテストメソッドは順番関係なく実行されてもOKなように・・・などありますが

他のファイルが影響するということは避けられるようになりました!

(Viewのテストは実際にトランザクション走らせずモック使っても・・・というのはまた別の記事で!)

<参考>

qiita.com

【Django】複合ユニーク制約を実装する

Djangoで複合主キーっぽいことをやりたくて。

使用技術

* Django 2.2.1
* Python 3.7.3


Djangoでは単一主キーのみをサポートしているため複合主キーはできないのですが 複合ユニーク制約ならできるようです。

Django2.2にて、MetaオプションにUniqueConstraintというものが登場しました。


models.py

from django.db import models


class Reservation(models.Model):
    room = models.CharField(verbose_name="部屋名", max_length=20)
    date = models.DateField(verbose_name="予約日")

    class Meta:
        constraints = [
            # 同じ日に部屋の予約を重複させない
            models.UniqueConstraint(fields=['room', 'date'], name='unique_booking'),
        ]

        

※こちらよりコードを引っ張りました。

Django 2.2 LTS 主な変更点まとめ – CreditEngine Tech – Medium


class Metaにて、constraintsに制約を設定するオブジェクトの配列を代入します。

上記では、UniqueConstraintの引数「fields」に設定した'room', 'date'の2つのカラムが複合ユニークとなります。

https://docs.djangoproject.com/ja/2.2/ref/models/constraints/#django.db.models.UniqueConstraint


なお、UniqueConstraintに関してはバリデーションの代わりとはならず、既存組み合わせのレコードを登録しようとするとIntegurityErrorが発生します

>>> import datetime
>>> from .models import Reservation
>>> # 正常にcreateが実行される
Reservation.objects.create(room='room_a', date=datetime.date(2019, 6, 1))

>>> # 既存のroomとdateの組み合わせをcreateしようとするとIntegurityErrorとなる
Reservation.objects.create(room='room_a', date=datetime.date(2019, 6, 1))
# django.db.utils.IntegrityErrorが発生する


なので、バリデーションをしたい場合は

例えば下記のように実装します。

まずは指定されたroomとdateの組み合わせがすでにDBに登録されているか確認するメソッドを作成します。

models.py

import datetime

from django.db import models


class Reservation(models.Model):
    room = models.CharField(verbose_name="部屋名", max_length=20)
    date = models.DateField(verbose_name="予約日")

    class Meta:
        constraints = [
            # 同じ日に部屋の予約を重複させない
            models.UniqueConstraint(fields=['room', 'date'], name='unique_booking'),
        ]
    
    # 追加
    @classmethod
    def check_duplicate(cls, room: str, date: datetime.date) -> bool:
        """同じ日に同じ部屋がすでにDBに登録されているどうかを判定します

        登録されていたらTrue, されていなかったらFalseを返します。
        """
        return cls.objects.filter(room=room, date=date).exists()


続いて、フォームで下記のように実装します。

今回はdjango.formsのFormとModelFormの両方で作成してみました。

どちらを使ってもroomとdateの重複をフォームのバリデーションの時点で確認できることが分かりました。

forms.py

import datetime

from django import forms

from .models import Reservation


class ReservationForm(forms.Form):
    room = forms.CharField(max_length=20)
    date = forms.DateField()

    def clean(self):
        cleaned_data = super(ReservationForm, self).clean()
        
        room = cleaned_data.get("room")
        date = cleaned_data.get("date")

        if Reservation.check_duplicate(room=room, date=date):
            raise forms.ValidationError(f'{date}に{room}はすでに予約されているため登録できません')

        return cleaned_data


class ReservationModelForm(forms.ModelForm):
    class Meta:
        model = Reservation
        fields = ['room', 'date']


python manage.py shellで実行すると下記のようになりました。

>>> # python manage.py shellにて
>>> # DBには room='room_a', date=datetime.date(2019, 6, 1)のレコードがすでに存在する状態

>>> from app.forms.reservation_form import ReservationForm, ReservationModelForm


>>> # まずはforms.Formで実装したフォーム

>>> form_a = ReservationForm({'room':'room_a', 'date':datetime.date(2019,6,1)})
>>> form_a.is_valid()
False

>>> form_a.errors
{'__all__': ['2019-06-01にroom_aはすでに予約されているため登録できません']}
>>> # forms.ValidationErrorで設定したエラーメッセージが返される

>>> # 続いてforms.ModelFormで実装したフォーム

>>> form_b = ReservationModelForm({'room': 'room_a', 'date': datetime.date(2019, 6, 1)})
>>> form_b.is_valid()
False

>>> form_b.errors
{'__all__': ['この 部屋名 と 予約日 を持った Reservation が既に存在します。']}
>>> # DjangoのModelFormでデフォルトで実装されているエラーメッセージが返される


また、Metaオプションで複合ユニーク制約する方法として unique_togetherというものもあります。

ただし、こちらに関しては将来廃止される可能性もあると公式ドキュメントにも書いているので Django2.2以降をご利用の場合はUniqueConstraintで実装するようにしましょう。

https://docs.djangoproject.com/en/2.2/ref/models/options/#unique-together


UniqueConstraintのついでに複合インデックスもやってくれてないかなと思って調べたところ、 複合ユニーク制約のみで複合インデックスは別に実装する必要があるようです。

Metaオプションにindexesを追加します。 indexesはmodel.Indexを要素とする配列となります。

Indexの引数fieldsに複合インデックスとしたいフィールドの名前からなる配列を代入してください。

models.py

from django.db import models


class Reservation(models.Model):
    room = models.CharField(verbose_name="部屋名", max_length=20)
    date = models.DateField(verbose_name="予約日")

    class Meta:
        constraints = [
            # 同じ日に部屋の予約を重複させない
            models.UniqueConstraint(fields=['room', 'date'], name='unique_booking'),
        ]

        # 複合インデックスの実装
        indexes = [
            models.Index(fields=['room', 'date'])
        ]


複合インデックスを設定する方法でMetaクラスにindex_togetherを設定する方法がありますが、 こちらは将来廃止される可能性もあると公式ドキュメントにも書いているので Django2.2以降をご利用の場合はindexesで実装するようにしましょう。

https://docs.djangoproject.com/en/2.2/ref/models/options/#index-together

【Python】GoogleAPIを利用したメソッドのテストでモックを使う

外部APIを利用したメソッドの単体テスト書きたいなあ・・・ 

単体テストで何回も外部APIに接続するのはお行儀悪いって聞くけどどう書けば良いのだろう?と思い。

簡単な機能のテストを書きました。

ちょっと変更しましたが、Google Calendar APIのquick startでやっていることをやります。

developers.google.com


一つのメソッドが長いので下記のような役割のメソッドに分割しました。

  1. pickleに保存した認証情報をロードする

  2. 認証情報が存在しないor有効期限切れならば認証情報を再生成する

  3. イベント情報を取得する役割のオブジェクトを生成する

  4. イベント情報を取得する


今回テストするプロダクトコード↓

event.py

from dataclasses import dataclass
from typing import Iterable, Optional, Dict
import datetime
import pickle
import os.path
import pathlib

import googleapiclient.discovery
import google_auth_oauthlib.flow
from google.auth.transport.requests import Request
import google.oauth2.credentials
import iso8601


# 環境変数にした方がよいですが、ブログではわかりやすいように表示しています
SCOPES = ['https://www.googleapis.com/auth/calendar.readonly']


class ServiceBuilder:
    @classmethod
    def renew_token(cls, credentials: google.oauth2.credentials.Credentials) -> google.oauth2.credentials.Credentials:
        # google.oauth2.credentials.Credentialsは外部APIのクラス
        if credentials.expired and credentials.refresh_token:
            # 外部APIに接続
            credentials.refresh(Request())
        else:
            # from_client_secrets_file returns google_auth_oauthlib.flow.Flow object
            flow = google_auth_oauthlib.flow.InstalledAppFlow.from_client_secrets_file(
                'credentials.json', SCOPES)
            credentials = flow.run_local_server()
        return credentials

    @classmethod
    def build(cls, credentials: google.oauth2.credentials.Credentials) -> googleapiclient.discovery.Resource:
        """トークンを元にServiceを生成します
        """

        # If there are no (valid) credentials available, let the user log in.
        # TODO: 42-43はrenew_tokenの先頭の方がよい
        if not credentials.valid:
            credentials = cls.renew_token(credentials)
        
        # 外部APIに接続
        return googleapiclient.discovery.build('calendar', 'v3', credentials=credentials)


@dataclass
class Event:
    summary: str
    start: datetime.datetime
    end: datetime.datetime


class EventAdapter:
    @classmethod
    def from_google_api(cls, google_event: Dict) -> Event:
        """GoogleAPIから取得したDict型のイベント情報をEventオブジェクトに変換します

        Dict型のままだと利用しない情報も含まれている、かつネストが深くて使いづらいため

        iso8601について
        https://pypi.org/project/iso8601/
        """
        return Event(
            summary=google_event.get('summary'),
            start=iso8601.parse_date(google_event.get('start').get('dateTime')),
            end=iso8601.parse_date(google_event.get('start').get('dateTime'))
        )


class EventManager:
    @classmethod
    def fetch_events(cls, number: int, service: googleapiclient.discovery.Resource) -> Iterable[Event]:
        """Google Calendar APIからイベントを取得します。

        number: int 取得するイベントの最大数
        service: イベントを取得するResourceオブジェクト
        """
        # service = ServiceBuilder.build()
        now = datetime.datetime.utcnow().isoformat() + 'Z' # 'Z' indicates UTC time
        print('Getting the upcoming 10 events')
        events_result = service.events().list(calendarId='primary', timeMin=now,
                                            maxResults=10, singleEvents=True,
                                            orderBy='startTime').execute()
        events = events_result.get('items', [])

        if not events:
            return []
        return [EventAdapter.from_google_api(event) for event in events]

main.py

import pathlib
from typing import Optional
import pickle

import google.oauth2.credentials

import event


def load_token(file: pathlib.Path) -> Optional[google.oauth2.credentials.Credentials]:
    """トークンをロードします。

    pickleの中にtokenがなければNoneを返します。
    """
    try:
        # pathlib.Path('token.pickle').read_bytes()をモックしたテスト
        with file.open('rb') as f:
            return pickle.load(f)
    except FileNotFoundError:
        return None


def main():
    token_file = pathlib.Path('token.pickle')
    credentials = event.ServiceBuilder.renew_token(credentials=load_token(token_file))

    with token_file.open('wb') as f:
        pickle.dump(credentials, f)
    
    service = event.ServiceBuilder.build() # missing one argument
    events = event.EventManager.fetch_events(number=2, service=service)
    print(events)

if __name__ == '__main__':
    main()


テストしたいこと

  1. 認証情報が有効切れor認証情報がpickleにない場合は新たにトークンが生成されるかどうか

  2. fetch_eventsでイベントが取得されるかどうか。

  3. fetch_eventsでGoogle Calendar APIから取得されたイベントが0件の場合は空の配列が返されるかどうか


from_google_apiとload_tokenに関しては

ちゃんとインスタンスができているか? 例外を発生するか?

といったようにPythonの仕様の確認になってしまうのでテストしないことにしました


1. 認証情報が有効切れor認証情報がpickleにない場合は新たにトークンが生成されるかどうか

まず、どこが外部APIに接続しているか調べます。

class ServiceBuilder:
    @classmethod
    def renew_token(cls, credentials: google.oauth2.credentials.Credentials) -> google.oauth2.credentials.Credentials:
        # google.oauth2.credentials.Credentialsは外部APIのクラス
        if credentials.expired and credentials.refresh_token:
            credentials.refresh(Request())  # 外部APIに接続
        else:
            # from_client_secrets_file returns google_auth_oauthlib.flow.Flow object
            flow = google_auth_oauthlib.flow.InstalledAppFlow.from_client_secrets_file(
                'credentials.json', SCOPES)   # 外部APIに接続
            credentials = flow.run_local_server()   # 外部APIに接続
        return credentials


「外部APIに接続」の部分をモックすれば人様のAPIに迷惑をかけない!

renew_token 自体をテストしようとしました。

でも、renew_token自体をテストしても良いけれども、 renew_tokenをまるっとモックしてServiceBuilder.buildで確認しちゃった方が楽だなあと。

credentailsの状態によってbuildの振る舞いが変わるので credentialsのモックを作り、credentialsモックの状態がそれぞれ異なる場合のテストを作ります。

test.py

@unittest.mock.patch('event.ServiceBuilder.renew_token')
@unittest.mock.patch('googleapiclient.discovery.build')
def test_valid_token(mock_build, mock_renew_token):
    credentials = unittest.mock.Mock(valid=True)

    event.ServiceBuilder.build(credentials=credentials)
    assert not mock_renew_token.called

@unittest.mock.patch('event.ServiceBuilder.renew_token')
@unittest.mock.patch('googleapiclient.discovery.build')
def test_expired_token_has_refresh_token(mock_build, mock_renew_token):
    credentials = unittest.mock.Mock(valid=False, expired=True, refresh_token='I am refresh token')

    event.ServiceBuilder.build(credentials=credentials)
    assert  mock_renew_token.called

credentailsの状態が妥当でない場合に認証情報が再生成されるメソッドが呼ばれるかどうかを確認できました!


2. fetch_eventsでイベントが取得されるかどうか。

3. fetch_eventsでGoogle Calendar APIから取得されたイベントが0件の場合は空の配列が返されるかどうか

イベントの取得のメソッドテストでは、イベントをちゃんと取得できるかのみを確認したいです。

serviceにGoogle Calendar APIからイベントを取得させていますが、serviceに利用されているcredentialsが有効期間切れでrefresh_tokenが実行されてしまう・・・

なんてことがあったら面倒です。


なので、fetch_eventsの外でserviceを生成させ、serviceをfetch_eventsの引数にします。

fetch_eventsのテストでは、serviceをモックし、serviceが正常に動くことを前提とすることにしました。

Google Calendar APIのドキュメントを確認すると、 イベント取得メソッドは下記のようになっています。

service.events().list(calendarId='primary', timeMin=now,
                                        maxResults=10, singleEvents=True,
                                        orderBy='startTime').execute()

https://developers.google.com/calendar/quickstart/python


serviceって何クラスのオブジェクトだと調べました。

from googleapiclient.discovery import build

service = build('calendar', 'v3', credentials=creds)


serviceはbuildによって生成されたオブジェクトらしい・・・

googleapiclient.discoveryのbuildはResourceオブジェクトを返すそうです。

https://github.com/googleapis/google-api-python-client/blob/master/googleapiclient/discovery.py#L170


Resourceを探したところ、

googleapiclient.discoveryのResourceだとわかりました。

https://github.com/googleapis/google-api-python-client/blob/master/googleapiclient/discovery.py#L995


ということでserviceはResourceオブジェクトをモックしよう。

service.events().list(calendarId='primary', timeMin=now,
                                        maxResults=10, singleEvents=True,
                                        orderBy='startTime').execute()


上記をもってGoogle Calendar APIからイベント情報のDictが返ってきます。

ということで、serviceをモックしてこんな感じでイベント情報を返すとします。

実際のAPIではもっとたくさんの情報が返されますが

テストに必要な情報のみにしました。

https://developers.google.com/calendar/v3/reference/events/list

mock_resource.events.return_value.list.return_value.execute.return_value = {
        "items": [
                    {
                        "summary": "test",
                        "start": {
                            "dateTime": "2019-06-03T02:00:00+09:00"
                        },
                        "end": {
                            "dateTime": "2019-06-03T02:45:00+09:00"
                        },
                    },
                        {
                        "summary": "test",
                        "start": {
                            "dateTime": "2019-06-03T02:00:00+09:00"
                        },
                        "end": {
                            "dateTime": "2019-06-03T02:45:00+09:00"
                        },
                    },
        ]
    }

そして2と3のテストを

@unittest.mock.patch('googleapiclient.discovery.Resource')
def test_fetch_events(mock_resource):
    mock_resource.events.return_value.list.return_value.execute.return_value = {
        "items": [
                    {
                        "summary": "test",
                        "start": {
                            "dateTime": "2019-06-03T02:00:00+09:00"
                        },
                        "end": {
                            "dateTime": "2019-06-03T02:45:00+09:00"
                        },
                    },
                        {
                        "summary": "test",
                        "start": {
                            "dateTime": "2019-06-03T02:00:00+09:00"
                        },
                        "end": {
                            "dateTime": "2019-06-03T02:45:00+09:00"
                        },
                    },
        ]
    }
    actual = event.EventManager.fetch_events(number=2, service=mock_resource)
    assert 2 == len(actual)

    for item in actual:
        assert isinstance(item, event.Event)


@unittest.mock.patch('googleapiclient.discovery.Resource')
def test_fetch_no_events(mock_resource):
    mock_resource.events.return_value.list.return_value.execute.return_value = {}

    actual = event.EventManager.fetch_events(number=2, service=mock_resource)
    assert [] == actual


ポイントは

  • 外部APIのオブジェクトは何かを調べてそのクラスのオブジェクトをモックすること

  • return_valueをどう実行するか

だと思いました。

外部APIのソースをある程度読むことになり、少し大変でしたが 勉強になりました。


今回のソースコードはこちら

github.com


また、NaritoさんがDjangoを利用してGoogle Calendar APIと連携したアプリケーションを作成されていました。

https://torina.top/detail/466/


近いうちにDjangoGoogle Calendar APIのテストの記事も書こうかなと思います。

(内容はあまり変わらないと思いますが)

【Django】フォームのカスタムバリデーションをテストする

Djangoのフォームで独自のバリデーションを実装した際のテスト方法を紹介します。 間違いがありましたら、ご指摘お願いします(>人<)

今回使ったバージョン

* Python 3.7.1
* Django 2.1.7


今回実装するフォームはこちら

forms.py
import re

from django import forms
from django.core.exceptions import ValidationError


def is_hiragana(value):
    pattern = r'^[ぁ-ん]+$'
    if not re.fullmatch(pattern, value):
        raise ValidationError('ひらがなで書いてください')


def validate_chatwork_id(value):
    pattern = r'^[0-9]+$'
    if not re.fullmatch(pattern, value) or len(value) != 7:
        raise ValidationError('半角数字7文字で書いてください')


class EmployeeForm(forms.Form):
    name = forms.CharField(
        max_length=225,
        label='名前',
        widget=forms.TextInput(attrs={'class': 'form-control'})
    )
    name_kana = forms.CharField(
        max_length=225,
        label='名前(かな)',
        validators=[is_hiragana],
        required=False,
        widget=forms.TextInput(attrs={'class': 'form-control'})
    )
    chatwork_id = forms.CharField(
        label='チャットワークID',
        validators=[validate_chatwork_id],
        widget=forms.TextInput(attrs={'class': 'form-control'})
    )
テストしたいこと
  • 名前(かな)にひらがな以外が含まれていたらエラーを表示したい

  • チャットワークIDが半角数字7文字以外だったらエラーを表示したい


form.is_valid()でTrueかFalseを返すテストは良くみるんだけど、

正しくエラーメッセージが出力されるかの方法がわからん・・・


とりあえず書いてみた。

from django.core.exceptions import ValidationError
from django.test import TestCase
from .forms import EmployeeForm


class TestEmployeeForm(TestCase):
    @classmethod
    def setUpClass(cls):
        pass

    def test_invalid_form_not_hiragana(self):
        params = dict(
            name='サンプル太郎',
            name_kana='サンプルタロウ',
            chatwork_id='0123456'
        )
        form = EmployeeForm(params)
        form.is_valid()
        self.assertEqual(
            'ひらがなで書いてください',
            form.errors
        )

    @classmethod
    def tearDownClass(cls):
        pass


結果、怒られる。

AssertionError: 'ひらがなで書いてください' != {'name_kana': ['ひらがなで書いてください']}


Dict[List]か・・・なら・・・、と試しにやってみる。

    def test_invalid_form_not_hiragana(self):
        params = dict(
            first_name='サンプル太郎',
            first_name_kana='サンプルタロウ',
            chatwork_id='0123456'
        )
        form = EmployeeForm(params)
        form.is_valid()
        self.assertTrue(
            'ひらがなで書いてください' in form.errors['name_kana']
        )


できた!!!


同じ要領でchatwork_idもやってみると

    def test_invalid_form_chatwork_id(self):
        with self.subTest('半角数字以外を含むとエラーになる'):
            params = dict(
                first_name='サンプル太郎',
                first_name_kana='さんぷるたろう',
                chatwork_id='012aaaaa'
            )
            form = EmployeeForm(params)
            form.is_valid()
            self.assertTrue(
                '半角数字7文字で書いてください' in form.errors['chatwork_id']
            )
            
        with self.subTest('7文字以上だとエラーになる'):
            params = dict(
                first_name='サンプル太郎',
                first_name_kana='さんぷるたろう',
                chatwork_id='012345678'
            )
            form = EmployeeForm(params)
            form.is_valid()
            self.assertTrue(
                '半角数字7文字で書いてください' in form.errors['chatwork_id']
            )

        with self.subTest('7文字以内だとエラーになる'):
            params = dict(
                first_name='サンプル太郎',
                first_name_kana='さんぷるたろう',
                chatwork_id='012345'
            )
            form = EmployeeForm(params)
            form.is_valid()
            self.assertTrue(
                '半角数字7文字で書いてください' in form.errors['chatwork_id']
            )


form.errorsが何者かみてみると・・・

print(type(form.errors))  # django.forms.utils.ErrorDictが出力される
print(type(form.errors['name_kana']))  # django.forms.utils.ErrorListが出力される

print(form.errors)  # <ul class="errorlist"><li>last_name<ul class="errorlist"><li>この項目は必須です。</li></ul></li><li>name_kana<ul class="errorlist"><li>ひらがなで書いてください</li></ul></li></ul>
print(form.errors['name_kana'])  # <ul class="errorlist"><li>ひらがなで書いてください</li></ul>

form.errorsはErrorDictというDictを継承したオブジェクトで

form.errorsはErrorListというListを継承したオブジェクトのようです。


ErrorDictのkeyはそのフォームの各フィールドになっているそうです。 ErrorListは各フィールドのエラーの表示方法を決めてるっぽいですね。

print()でulタグが出力されたのは、__str__の処理の中がas_ulだからかと謎が解けました。

https://github.com/django/django/blob/master/django/forms/utils.py


少しテストから話が逸れますが validatorsを使わずclean_field使用してもテストは同じ結果で通りました。

class EmployeeForm(forms.Form):
    name = forms.CharField(
        max_length=225,
        label='名前',
        widget=forms.TextInput(attrs={'class': 'form-control'})
    )
    name_kana = forms.CharField(
        max_length=225,
        label='名前(かな)',
        required=False,
        widget=forms.TextInput(attrs={'class': 'form-control'})
    )
    chatwork_id = forms.CharField(
        label='チャットワークID',
        widget=forms.TextInput(attrs={'class': 'form-control'})
    )

    def clean_name_kana(self):
        pattern = r'^[ぁ-ん]+$'
        name_kana = self.cleaned_data['name_kana']
        if not re.fullmatch(pattern, name_kana):
            raise ValidationError('ひらがなで書いてください')
        return name_kana

    def clean_chatwork_id(self):
        pattern = r'^[0-9]+$'
        chatwork_id = self.cleaned_data['chatwork_id']
        if not re.fullmatch(pattern, chatwork_id) or len(chatwork_id) != 7:
            raise ValidationError('半角数字7文字で書いてください')
        return chatwork_id

ちなみに、validatorとclean_fieldの違いについてはこちらの記事をご参照ください。 validatorはそれぞれのフィールドに入力された値をチェックするけれどもフォーム自体の内容は確認しないっぽいですね。

cleanが通らない場合があると考えると、clean_fieldを使った方が無難そう

hateda.hatenadiary.jp

今回使用したコードまとめ

forms.py

import re

from django import forms
from django.core.exceptions import ValidationError


def is_hiragana(value):
    pattern = r'^[ぁ-ん]+$'
    if not re.fullmatch(pattern, value):
        raise ValidationError('ひらがなで書いてください')


def validate_chatwork_id(value):
    pattern = r'^[0-9]+$'
    if not re.fullmatch(pattern, value) or len(value) != 7:
        raise ValidationError('半角数字7文字で書いてください')


class EmployeeForm(forms.Form):
    name = forms.CharField(
        max_length=225,
        label='名前',
        widget=forms.TextInput(attrs={'class': 'form-control'})
    )
    name_kana = forms.CharField(
        max_length=225,
        label='名前(かな)',
        validators=[is_hiragana],
        required=False,
        widget=forms.TextInput(attrs={'class': 'form-control'})
    )
    chatwork_id = forms.CharField(
        label='チャットワークID',
        validators=[validate_chatwork_id],
        widget=forms.TextInput(attrs={'class': 'form-control'})
    )

"""こっちでも良い

class EmployeeForm(forms.Form):
    name = forms.CharField(
        max_length=225,
        label='名前',
        widget=forms.TextInput(attrs={'class': 'form-control'})
    )
    name_kana = forms.CharField(
        max_length=225,
        label='名前(かな)',
        required=False,
        widget=forms.TextInput(attrs={'class': 'form-control'})
    )
    chatwork_id = forms.CharField(
        label='チャットワークID',
        widget=forms.TextInput(attrs={'class': 'form-control'})
    )
    
    def clean_name_kana(self):
        pattern = r'^[ぁ-ん]+$'
        name_kana = self.cleaned_data['name_kana']
        if not re.fullmatch(pattern, name_kana):
            raise ValidationError('ひらがなで書いてください')
        return name_kana
    
    def clean_chatwork_id(self):
        pattern = r'^[0-9]+$'
        chatwork_id = self.cleaned_data['chatwork_id']
        if not re.fullmatch(pattern, chatwork_id) or len(chatwork_id) != 7:
            raise ValidationError('半角数字7文字で書いてください')
        return chatwork_id

 """


test_forms.py(ちょっとテスト増やして整えた)

from django.core.exceptions import ValidationError
from django.test import TestCase
from .forms import EmployeeForm


class TestEmployeeForm(TestCase):
    @classmethod
    def setUpClass(cls):
        pass

    def test_valid_form(self):
        params = dict(
            name='サンプル太郎',
            name_kana='さんぷるたろう',
            chatwork_id='0123456'
        )
        form = EmployeeForm(params)
        self.assertTrue(form.is_valid())

    def test_invalid_form_not_hiragana(self):
        params = dict(
            first_name='サンプル太郎',
            first_name_kana='サンプルタロウ',
            chatwork_id='0123456'
        )
        form = EmployeeForm(params)
        form.is_valid()

        with self.subTest('フォームのバリーデーションで妥当ではないと返される'):
            self.assertFalse(form.is_valid())

        with self.subTest('ひらがなで書けというエラーになる'):
            self.assertTrue(
                'ひらがなで書いてください' in form.errors['name_kana']
            )

        self.assertTrue(
            'ひらがなで書いてください' in form.errors['name_kana']
        )

    def test_invalid_form_chatwork_id_has_charactor(self):
        params = dict(
            name='サンプル太郎',
            name_kana='さんぷるたろう',
            chatwork_id='012aaaaa'
        )
        form = EmployeeForm(params)
        form.is_valid()
        with self.subTest('半角数字以外を含むとフォームのバリーデーションで妥当ではないと返される'):
            self.assertFalse(form.is_valid())

        with self.subTest('半角数字以外を含むと半角数字7文字で書けというエラーになる'):
            self.assertTrue(
                '半角数字7文字で書いてください' in form.errors['chatwork_id']
            )

    def test_invalid_form_chatwork_id_is_over_7(self):
        params = dict(
            name='サンプル太郎',
            name_kana='さんぷるたろう',
            chatwork_id='012345678'
        )
        form = EmpForm(params)
        form.is_valid()

        with self.subTest('7文字以上だとフォームのバリーデーションで妥当ではないと返される'):
            self.assertFalse(form.is_valid())

        with self.subTest('7文字以上だと半角数字7文字で書けというエラーになる'):
            self.assertTrue(
                '半角数字7文字で書いてください' in form.errors['chatwork_id']
            )

    def test_invalid_form_chatwork_id_is_less_than_7(self):
        params = dict(
            name='サンプル太郎',
            name_kana='さんぷるたろう',
            chatwork_id='012345'
        )
        form = EmpForm(params)
        form.is_valid()
        with self.subTest('7文字以内だとフォームのバリーデーションで妥当ではないと返される'):
            self.assertFalse(form.is_valid())

        with self.subTest('7文字以内だと半角数字7文字で書けというエラーになる'):
            self.assertTrue(
                '半角数字7文字で書いてください' in form.errors['chatwork_id']
            )

    @classmethod
    def tearDownClass(cls):
        pass