2020-03-14

PyMongoを使って、PythonからMongoDBにアクセスしてみる

Python MongoDB

これは、なにをしたくて書いたもの？

MongoDBの練習がてらに、PythonからMongoDBにアクセスするプログラムを書いてみようかなと。

MongoDBのPythonドライバー

MongoDBのPythonドライバーは、MongoDBのサイトを見るとPyMongoとMotorがあるようです。

MongoDB Python Drivers — MongoDB Ecosystem

PyMongo — MongoDB Ecosystem

Motor (Async Driver) — MongoDB Ecosystem

PyMongoが推奨のドライバーで、MotorはTornadeやasyncioと互換性のある非同期ドライバーのようです。

MongoDB本体のバージョンとの互換性を確認してみると、Motorは途中で止まっているようにも思います。

PyMongo / Compatibility

Motor（Async Driver） / Compatibility

この意味でも、PyMongoを選択したよさそうですね。

PyMongoの、ドキュメントおよびAPIリファレンスはこちらです。

PyMongo 3.9.0 Documentation — PyMongo 3.9.0 documentation

API Documentation — PyMongo 3.9.0 documentation

では、使っていってみましょう。

環境

今回の環境は、こちらです。

$ python3 -V
Python 3.6.9


$ pip3 -V
pip 9.0.1 from /path/to/venv/lib/python3.6/site-packages (python 3.6)

また、MongoDBのバージョンは4.2.3とし、MongoDBが動作しているサーバーのIPアドレスは172.17.0.2とします。

インストール

ドキュメントに沿って、PyMongoをインストールします。

Installing / Upgrading — PyMongo 3.9.0 documentation

Python 2.7、3.4以上に対応しているようですね。

PyMongo supports CPython 2.7, 3.4+, PyPy, and PyPy3.5+

pipでインストール。

$ pip3 install pymongo

今回は、3.10.1がインストールされました。

$ pip3 freeze | grep pymongo
pymongo==3.10.1

が、この時点のドキュメントの「current」は、3.9.0なんですよね。これは、どういうことでしょう。まあ、今回は気にせずいきますか…。
このエントリが参照しているドキュメントへのリンクは、3.9.0で固定してあります。

あと、確認はテストコードで行おうと思うので、pytestもインストールしておきます。

$ pip3 install pytest
$ pip3 freeze | grep pytest
pytest==5.4.1

ディレクトリと__init__.pyを作成して、準備完了。

$ mkdir tests && touch tests/__init__.py

テストコードの雛形

テストコードの雛形は、こちら。
tests/test_pymongo.py

import re

from pymongo import MongoClient
import pymongo

# ここに、テストを書く！

基本的には、PyMongoのチュートリアルを見つつ、必要に応じてAPIリファレンスを参照して進めていくのかな、と思います。

Tutorial — PyMongo 3.9.0 documentation

API Documentation — PyMongo 3.9.0 documentation

PyMongoからMongoDBに接続する

PyMongoからMongoDBに接続する方法は、こちら。

Making a Connection with MongoClient

接続先となるサーバーの記載方法は、2通りあるようですね。また、MongoClientはコンテキストマネージャープロトコルを実装して
いるので、with構文も使えます。

こんな感じですね。

def test_connect_mongo():
    client = MongoClient("172.17.0.2", 27017)

    assert client is not None

    client.close()


def test_connect_mongo_with():
    with MongoClient("mongodb://172.17.0.2:27017") as client:
        assert client is not None

使わなくなった接続は、closeしておきましょう。

ちなみに、MongoClientはコネクションプールの役割も持っています。

How does connection pooling work in PyMongo?

スレッドセーフであり、プロセスセーフではありません、と。

Is PyMongo thread-safe?

Is PyMongo fork-safe?

続いて、データベースとコレクションの取得。

Getting a Database

Getting a Collection

今回は書籍をお題に、データベースとコレクションを作成。

def test_get_database_and_collection():
        with MongoClient("172.17.0.2", 27017) as client:
            test_db = client.test_db
            book_collection = test_db.book_collection

            assert book_collection is not None

ドキュメント1件の登録、1件の取得

では、ドキュメントを登録してみます。

まずは1件登録して、1件取得。

Inserting a Document

Getting a Single Document With find_one()

こんな感じで、コレクションに対してinsert_oneでドキュメントを1件登録、find_oneで指定のドキュメントを1件取得できます。

def test_document_insert_and_find_one():
        with MongoClient("172.17.0.2", 27017) as client:
            test_db = client.test_db
            book_collection = test_db.book_collection

            book = {"isbn": "978-4873117386", "title": "入門 Python 3", "price": 4070}

            book_collection.insert_one(book)

            assert book_collection.find_one({"isbn": "978-4873117386"})["title"] == book["title"]
            assert book_collection.count_documents({}) == 1

            book_collection.delete_many({})
            assert book_collection.count_documents({}) == 0

テストコードの都合上、カウントと全件取得も行っていますが。

Counting

delete_many

複数件のドキュメントの登録、取得

続いて、複数ドキュメントの登録と、複数ドキュメントの検索を行ってみましょう。

Bulk Inserts

Querying for More Than One Document

こんな感じで。

def test_document_insert_and_find_many():
        with MongoClient("172.17.0.2", 27017) as client:
            test_db = client.test_db
            book_collection = test_db.book_collection

            books = [
                {"isbn": "978-4873117386", "title": "入門 Python 3", "price": 4070},
                {"isbn": "978-4048930611", "title": "エキスパートPythonプログラミング改訂2版", "price": 3960},
                {"isbn": "978-4297111113", "title": "Python実践入門", "price": 3278}
            ]

            book_collection.insert_many(books)
            assert book_collection.count_documents({}) == 3

            found_all_books = [book for book in book_collection.find({}, sort = [("price", pymongo.DESCENDING)])]
            assert found_all_books[0]["title"] == "入門 Python 3"
            assert len(found_all_books) == 3

            found_books = [book for book in book_collection.find({"title": re.compile("Python"), "price": {"$gt": 3500}}, sort = [("price", pymongo.ASCENDING)])]
            assert found_books[0]["title"] == "エキスパートPythonプログラミング改訂2版"
            assert len(found_books) == 2

            book_collection.delete_many({})

            assert book_collection.count_documents({}) == 0

ドキュメントを複数件登録するには、コレクションに対してinsert_manyを呼び出せばOKです。引数は、登録したいドキュメントの
リストです。

検索はfindで行うのですが、戻り値はCursorになるので今回は内包表記でリストに変換しています。

            found_all_books = [book for book in book_collection.find({}, sort = [("price", pymongo.DESCENDING)])]

あと、価格の降順ソートにしておきました。

また、もうひとつ検索のパターンとしては、AND条件で正規表現や比較演算子を使うものも。

            found_books = [book for book in book_collection.find({"title": re.compile("Python"), "price": {"$gt": 3500}}, sort = [("price", pymongo.ASCENDING)])]

正規表現で検索を行う場合は、Pythonの正規表現オブジェクトを渡せばOKです。こちらのソートは、価格の昇順にしておきました。

とりあえず、こんなところでしょうか。

まとめ

MongoDBのPythonドライバー、PyMongoからMongoDBにアクセスして、初歩的な操作をざっと試してみました。

ディクショナリーをベースにして使えるので、簡単に使えてよいのではないでしょうか。

ちょっとした時に利用していくんだろうなーと思います。

2020-03-14

pytestを使って、Pythonのテストコードを書く

Python

これは、なにをしたくて書いたもの？

Pythonでテストコードを書く時には、pytestを使うのが良いという話を聞きまして。1度、自分でも押さえておこうかな、と。

https://docs.pytest.org/en/latest/

以前、unittestで記載した内容のpytest版です。

unittestライブラリで、Pythonのテストコードを書いて実行する - CLOVER🍀

pytestとは

いわゆる、テスティングフレームワークです。

以下が特徴だそうです。

アサーションに失敗した時の詳細情報表示
テストモジュールおよび関数のオートディスカバリー
小さなテストや、パラメタライズ化された長期間のテストのためのテストリソースを管理する、モジュール化されたfixture
unittestおよびnoseのテストケースを実行可能
Python 3.5以上、PyPy3上で動作
リッチなプラグインアーキテクチャによる、315以上の外部プラグインとコミュニティ

下位互換性や、Python 2.7および3.4に関するサポートについては、こちら。

Backwards Compatibility Policy — pytest documentation

Python 2.7 and 3.4 support — pytest documentation

APIドキュメントについては、こちらです。

API Reference — pytest documentation

まあ、とりあえず使っていってみましょう。

環境

今回の環境は、こちらです。

$ python3 -V
Python 3.6.9


$ pip3 -V
pip 9.0.1 from /path/to/venv/lib/python3.6/site-packages (python 3.6)

インストール

まずは、インストールしましょう。

Installation and Getting Started — pytest documentation

pip3でインストール。

$ pip3 install pytest

今回使用するpytestのバージョンは、5.3.5です。

$ pip3 freeze | grep pytest
pytest==5.3.5

ちなみに、このページにサポートするPythonのバージョンとプラットフォームがハッキリと書かれていますね。

Pythons: Python 3.5, 3.6, 3.7, PyPy3

Platforms: Linux and Windows

この環境上に、テスト対象およびテストコードを書いていきましょう。

お題とプロジェクト構成

お題とプロジェクト構成も、基本的にはunittestの時に書いたものに合わせます。

unittestライブラリで、Pythonのテストコードを書いて実行する - CLOVER🍀

まずはディレクトリレイアウト。

Choosing a test layout / import rules

テストコードをテスト対象（アプリケーションコード）側に置くスタイルと、そうでないスタイルがあるようですが、今回は
アプリケーションコードとテストコードをハッキリ分けて置きたいと思います。

Tests outside application code

アプリケーションコードを置くディレクトリを「sample」、テストコードを置くディレクトリを「tests」とします。

$ mkdir sample tests

アプリケーションコード側を、1度「src」ディレクトリに置くという話もあるようですが、これはまた別の機会に見てみましょう。

Packaging a python library | ionel's codelog

アプリケーションコードおよび、テストを作成して実行してみる

では、まずアプリケーションコードを作成。
sample/calc.py

class Calc:
    def add(self, x, y):
        return x + y

    def minus(self, x, y):
        return x - y

    def multiply(self, x, y):
        return x * y

    def divide(self, x, y):
        return x / y

続いて、テストコードを書いてみます。
tests/test_calc.py

from sample.calc import Calc

def test_add():
    calc = Calc()
    assert calc.add(1, 3) == 4

def test_minus():
    calc = Calc()
    assert calc.minus(5, 3) == 2

def test_multiply():
    calc = Calc()
    assert calc.multiply(2, 3) == 6

def test_divide():
    calc = Calc()
    assert calc.divide(10, 2) == 5

pytestがどのようにテストコードを検出するかは、こちらに記載があります。

Conventions for Python test discovery

「test*.py」または「*test.py」ファイルを探索し、関数名に「test」が接頭辞として付与された関数およびクラス名に「Test」が
接頭辞として付与されたテスト関数を対象とするようです。

要するに、「test」（クラスの場合は「Test」）を付けてください、と。

また、テストコードを認識してもらうには「init.py」が必要なようなので

Choosing a test layout / import rules

You can use Python3 namespace packages (PEP420) for your application but pytest will still perform test package name discovery based on the presence of init.py files.

作成しておきます。

$ touch tests/__init__.py

では、pytestを実行してみます。

まずはpytestコマンドを引数なしで実行。

$ pytest
========================================================================== test session starts ===========================================================================
platform linux -- Python 3.6.9, pytest-5.3.5, py-1.8.1, pluggy-0.13.1
rootdir: /path/to/pytest
collected 4 items                                                                                                                                                        

tests/test_calc.py ....                                                                                                                                            [100%]

=========================================================================== 4 passed in 0.01s ============================================================================

テストコードを探してきて、実行してくれました。

テストコードを指定。

$ pytest tests/test_calc.py 
========================================================================== test session starts ===========================================================================
platform linux -- Python 3.6.9, pytest-5.3.5, py-1.8.1, pluggy-0.13.1
rootdir: /path/to/pytest
collected 4 items                                                                                                                                                        

tests/test_calc.py ....                                                                                                                                            [100%]

=========================================================================== 4 passed in 0.01s ============================================================================

「-v」オプションを付けて実行すると、もっと詳細に情報を出力してくれます。

$ pytest -v
========================================================================== test session starts ===========================================================================
platform linux -- Python 3.6.9, pytest-5.3.5, py-1.8.1, pluggy-0.13.1 -- /path/to/venv/bin/python3
cachedir: .pytest_cache
rootdir: /path/to/pytest
collected 4 items                                                                                                                                                        

tests/test_calc.py::test_add PASSED                                                                                                                                [ 25%]
tests/test_calc.py::test_minus PASSED                                                                                                                              [ 50%]
tests/test_calc.py::test_multiply PASSED                                                                                                                           [ 75%]
tests/test_calc.py::test_divide PASSED                                                                                                                             [100%]

=========================================================================== 4 passed in 0.01s ============================================================================

ちょっと、テストを1度失敗させてみましょう。

テストコードをこのように変更して

def test_add():
    calc = Calc()
    assert calc.add(1, 2) == 4
#    assert calc.add(1, 3) == 4

実行。

$ pytest
=========================================================================== test session starts ===========================================================================
platform linux -- Python 3.6.9, pytest-5.3.5, py-1.8.1, pluggy-0.13.1
rootdir: /path/to/pytest
collected 4 items                                                                                                                                                         

tests/test_calc.py F...                                                                                                                                             [100%]

================================================================================ FAILURES =================================================================================
________________________________________________________________________________ test_add _________________________________________________________________________________

    def test_add():
        calc = Calc()
>       assert calc.add(1, 2) == 4
E       assert 3 == 4
E        +  where 3 = <bound method Calc.add of <sample.calc.Calc object at 0x7febb711b128>>(1, 2)
E        +    where <bound method Calc.add of <sample.calc.Calc object at 0x7febb711b128>> = <sample.calc.Calc object at 0x7febb711b128>.add

tests/test_calc.py:5: AssertionError
======================================================================= 1 failed, 3 passed in 0.03s =======================================================================

どういう状態で失敗したかがわかるんですね。

>       assert calc.add(1, 2) == 4
E       assert 3 == 4
E        +  where 3 = <bound method Calc.add of <sample.calc.Calc object at 0x7febb711b128>>(1, 2)
E        +    where <bound method Calc.add of <sample.calc.Calc object at 0x7febb711b128>> = <sample.calc.Calc object at 0x7febb711b128>.add

The writing and reporting of assertions in tests — pytest documentation

例外をアサーションしたい場合は、こちら。

Assertions about expected exceptions

確認できたので、テストコードは失敗しないように戻しておきます。

前処理・後処理を入れる

よくある、テストの実行前後に前処理・後処理を入れてみたいと思います。

Fixtureを使うみたいです。

Fixture finalization / executing teardown code

先ほど作ったテストコードをベースにして

$ cp tests/test_calc.py tests/test_calc_seup_teardown.py

こんな感じに作成。
tests/test_calc_seup_teardown.py

from sample.calc import Calc

import pytest

@pytest.fixture
def setup_and_teardown():
    print("Setup.")
    yield
    print("Teardown.")

def test_add(setup_and_teardown):
    calc = Calc()
    print("call add")
    assert calc.add(1, 3) == 4

def test_minus(setup_and_teardown):
    calc = Calc()
    print("call minus")
    assert calc.minus(5, 3) == 2

def test_multiply():
    calc = Calc()
    print("call multiply")
    assert calc.multiply(2, 3) == 6

def test_divide():
    calc = Calc()
    print("call divide")
    assert calc.divide(10, 2) == 5

@pytest.fixtureでデコレートした関数を定義して

@pytest.fixture
def setup_and_teardown():
    print("Setup.")
    yield
    print("Teardown.")

前後処理を挟み込みたいテスト関数の引数に設定します。

def test_add(setup_and_teardown):
    calc = Calc()
    print("call add")
    assert calc.add(1, 3) == 4

def test_minus(setup_and_teardown):
    calc = Calc()
    print("call minus")
    assert calc.minus(5, 3) == 2

確認。

$ pytest tests/test_calc_seup_teardown.py 
=========================================================================== test session starts ===========================================================================
platform linux -- Python 3.6.9, pytest-5.3.5, py-1.8.1, pluggy-0.13.1
rootdir: /path/to/pytest
collected 4 items                                                                                                                                                         

tests/test_calc_seup_teardown.py ....                                                                                                                               [100%]

============================================================================ 4 passed in 0.01s ============================================================================

どうなったか、まったくわかりません。

これは、今回printでデバッグ的に確認しようとしているのですが、stdin／stdoutがpytestでキャプチャされているからです。

Capturing of the stdout/stderr output — pytest documentation

今回の確認目的では、キャプチャをオフにすればprintの様子を観ることができます。

$ pytest --capture=no tests/test_calc_seup_teardown.py

## もしくは
$ pytest -s tests/test_calc_seup_teardown.py
=========================================================================== test session starts ===========================================================================
platform linux -- Python 3.6.9, pytest-5.3.5, py-1.8.1, pluggy-0.13.1
rootdir: /path/to/pytest
collected 4 items                                                                                                                                                         

tests/test_calc_seup_teardown.py Setup.
call add
.Teardown.
Setup.
call minus
.Teardown.
call multiply
.call divide
.

============================================================================ 4 passed in 0.01s ============================================================================

テスト関数の引数にFixtureを設定した、add、minusに関してはSetupとTeardownのメッセージが見えていますね。

ちなみに、キャプチャをオフにしなくてもテストに失敗した場合は

def test_add(setup_and_teardown):
    calc = Calc()
    print("call add")
    assert calc.add(1, 4) == 4

その関数でキャプチャされた情報などが出力されます。

$ pytest tests/test_calc_seup_teardown.py 
=========================================================================== test session starts ===========================================================================
platform linux -- Python 3.6.9, pytest-5.3.5, py-1.8.1, pluggy-0.13.1
rootdir: /path/to/pytest
collected 4 items                                                                                                                                                         

tests/test_calc_seup_teardown.py F...                                                                                                                               [100%]

================================================================================ FAILURES =================================================================================
________________________________________________________________________________ test_add _________________________________________________________________________________

setup_and_teardown = None

    def test_add(setup_and_teardown):
        calc = Calc()
        print("call add")
>       assert calc.add(1, 4) == 4
E       assert 5 == 4
E        +  where 5 = <bound method Calc.add of <sample.calc.Calc object at 0x7fddc8d5fcc0>>(1, 4)
E        +    where <bound method Calc.add of <sample.calc.Calc object at 0x7fddc8d5fcc0>> = <sample.calc.Calc object at 0x7fddc8d5fcc0>.add

tests/test_calc_seup_teardown.py:14: AssertionError
-------------------------------------------------------------------------- Captured stdout setup --------------------------------------------------------------------------
Setup.
-------------------------------------------------------------------------- Captured stdout call ---------------------------------------------------------------------------
call add
------------------------------------------------------------------------ Captured stdout teardown -------------------------------------------------------------------------
Teardown.
======================================================================= 1 failed, 3 passed in 0.03s =======================================================================

また、Fixtureを使うことでテスト関数の引数を設定したり、テスト間でデータを共有できたりします。

Fixtures as Function arguments

Sharing test data

このあたりも、覚えておきましょう。

まとめ

Pythonのテスティングフレームワーク、pytestを試してみました。

とりあえず、初歩的なことは少しは押さえられたのではないでしょうか。これから、使っていってみましょう。