csvのheaderを気にしてOrderedDictで読む方法

問題

例えば以下のようなcsvがあるときに、読み込んだ後にcsvのheaderの順序を保持してほしいという場合がある。

"age"    "name"
"20"    "foo"

カジュアルに以下の様なコードで読んだ場合に順序は不定。

import csv
import sys


r = csv.DictReader(sys.stdin, delimiter="\t")
print(list(r))

順序は不定。

[{'name': 'foo', 'age': '20'}]
[{'name': 'foo', 'age': '20'}]
[{'age': '20', 'name': 'foo'}]

対応方法

csv.DictReaderを継承するしか無いの?

例えば以下の様な感じにする。dict_factoryみたいな形で引数になっているとありがたいのだけれど。

import csv
import sys
from collections import OrderedDict


class OrderedDictReader(csv.DictReader):
    def __next__(self):
        if self.line_num == 0:
            # Used only for its side effect.
            self.fieldnames
        row = next(self.reader)
        self.line_num = self.reader.line_num

        # unlike the basic reader, we prefer not to return blanks,
        # because we will typically wind up with a dict full of None
        # values
        while row == []:
            row = next(self.reader)
        d = OrderedDict(zip(self.fieldnames, row))
        lf = len(self.fieldnames)
        lr = len(row)
        if lf < lr:
            d[self.restkey] = row[lf:]
        elif lf > lr:
            for key in self.fieldnames[lr:]:
                d[key] = self.restval
        return d


r = OrderedDictReader(sys.stdin, delimiter="\t")
print(list(r))

今度は大丈夫。

[OrderedDict([('age', '20'), ('name', 'foo')])]
[OrderedDict([('age', '20'), ('name', 'foo')])]
[OrderedDict([('age', '20'), ('name', 'foo')])]

logging入門

長すぎにならない程度に使い方をまとめてみる。

loggingの使い方

ライブラリの利用者

既に存在するアプリを実行するファイルの場合

if __name__ == "__main__":
    import logging
    logging.basicConfig(level=logging.DEBUG)  # or INFO or WARNING or ERROR
    run()

個人的には時間もみたいので以下の様にしている。

 logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")

設定可能な属性について詳しくはここ

ライブラリの作成者

ライブラリ内の話。基本的にはモジュール毎に __name__ でロガーを作る。アプリを作成する際にはご自由に。

import logging
logger = logging.getLogger(__name__)


# おもに問題を診断するときにのみ関心があるような、詳細な情報。
logger.debug("hmm")
# 想定された通りのことが起こったことの確認。
logger.info("hmm")
# 想定外のことが起こった、または問題が近く起こりそうである (例えば、'disk space low') ことの表示。
logger.warning("hmm")
# より重大な問題により、ソフトウェアがある機能を実行できないこと。
logger.error("hmm")
# プログラム自体が実行を続けられないことを表す、重大なエラー。
logger.critical("hmm")

最悪、debug,info,errorだけ使えば良い。

例外発生時のtracebackを出力したい場合

stack traceもログに出力したい場合には exc_info=True をつける

try:
    foo()
except:
    logger.warning("hmm", exc_info=True)

うるさいloggerを黙らせる

ロガーの名前が foo.bar.boo の場合

logging.getLogger("foo.bar.boo").setLevel(logging.CRITICAL)

(sentryで適切にaggregationしたい場合)

loggerに渡す文字列をaggregation用のidとして利用できるようにする。具体的にはformat文字列などを利用して文字列を生成しない。

# ok
logger.info("name: %s", name)
# ng
logger.info("name: {}".format(name))

後者は文字列フォーマットの適用結果がsentryに送られてしまうため。適切に発生したエラーをaggregateできない。

ses also

marshmallow-polyfieldを使ってoneOf的な構造のdataを扱う

はじめに

例えば、以下のよう1つのfieldに複数の形状の値が入ることがある。そして、その形状を決めるためにtypeなどfieldを含まれているJSONがあるとする。 以下の様な感じ(下の例では、personとgroupという2つの形状がobに入る可能性がある)。

{
  "ob": {
    "age": 20,
    "name": "foo"
  },
  "type": "person"
}

もしくはこう。

{
  "ob": {
    "name": "A",
    "members": [
      {
        "age": 20,
        "name": "foo"
      }
    ]
  },
  "type": "group"
}

それぞれ、typeで判別できるけれど。これを良い感じにmarshmallowでserialize,deserializeしたいという話し。

準備

事前に以下が必要。marshmallow-polyfieldを使う。

$ pip install marshmallow-polyfield

方法

以下の様な感じ。

from marshmallow import Schema, fields
from marshmallow_polyfield import PolyField


class Person(Schema):
    name = fields.String(required=True)
    age = fields.Integer(required=True)


class Group(Schema):
    name = fields.String(required=True)
    members = fields.List(fields.Nested(Person()), required=True)


def selector_for_deserialize(d, parent):
    if parent.get("type") == "group":
        return Group()
    else:
        return Person()


def selector_for_serialize(ob, parent):
    if "members" in ob:
        parent["type"] = "group"
        return Group()
    else:
        parent["type"] = "person"
        return Person()


class S(Schema):
    type = fields.String(required=True)
    ob = PolyField(
        serialization_schema_selector=selector_for_serialize,
        deserialization_schema_selector=selector_for_deserialize,
        required=True
    )


print(S().load({"ob": {"name": "foo", "age": 20}, "type": "person"}))
print(S().load({"ob": {"name": "A", "members": [{"name": "foo", "age": 20}]}, "type": "group"}))
print(S().dump({"ob": {"name": "foo", "age": 20}}))
print(S().dump({"ob": {"name": "A", "members": [{"name": "foo", "age": 20}]}}))

# UnmarshalResult(data={'ob': {'name': 'foo', 'age': 20}, 'type': 'person'}, errors={})
# UnmarshalResult(data={'ob': {'name': 'A', 'members': [{'name': 'foo', 'age': 20}]}, 'type': 'group'}, errors={})
# MarshalResult(data={'ob': {'name': 'foo', 'age': 20}, 'type': 'person'}, errors={})
# MarshalResult(data={'ob': {'name': 'A', 'members': [{'name': 'foo', 'age': 20}]}, 'type': 'group'}, errors={})

marshmallowで相互排他的なfieldを定義する方法

はじめに

こういうJSONを許したい。

{
  "left": {
    "name": "foo",
    "value": 100,
  }
}

あるいはこう。

{
  "right": {
    "name": "foo",
    "value": 100.0,
  }
}

left,rightというfieldの内どちらか1つだけ値が入るデータをvalidとしたい。例えば以下はダメ。

{}

これもだめ。

{
  "left": {
    "name": "foo",
    "value": 100,
  },
  "right": {
    "name": "foo",
    "value": 100.0,
  }
}

2つのleft,rightというfieldの構造がテキトウ過ぎるけれど。2つ(Nつ)あるfieldの内1つだけに値が入るという状態にしたい。

方法

field自体にこのような機能をつけるのは無理で。validate_schemasというschemaレベルのvalidationを使う。

import marshmallow as ma


class Item(ma.Schema):
    name = ma.fields.String(required=True)
    value = ma.fields.Integer(required=True)


class Item2(ma.Schema):
    name = ma.fields.String(required=True)
    value = ma.fields.Number(required=True)


class S(ma.Schema):
    left = ma.fields.Nested(Item)
    right = ma.fields.Nested(Item2)

    @ma.validates_schema
    def mutual(self, data):
        items = [item for item in [data.get("left"), data.get("right")] if item]
        if len(items) != 1:
            raise ma.ValidationError("items0 or items1")


print(S().load({}))
print(S().load({"left": {"name": "foo", "value": 10}}))
print(S().load({"right": {"name": "foo", "value": 10}}))
print(S().load({"left": {"name": "foo", "value": 10}, "right": {"name": "foo", "value": 10}}))

# UnmarshalResult(data={}, errors={'_schema': ['items0 or items1']})
# UnmarshalResult(data={'left': {'value': 10, 'name': 'foo'}}, errors={})
# UnmarshalResult(data={'right': {'value': 10.0, 'name': 'foo'}}, errors={})
# UnmarshalResult(data={'right': {'value': 10.0, 'name': 'foo'}, 'left': {'value': 10, 'name': 'foo'}}, errors={'_schema': ['items0 or items1']})

pythonのunittest用のmarkerライブラリを作りました

github.com

pythonのunittest用のmarkerライブラリを作りました

markerライブラリ?

特定のテストケースにマーカーを付けるためのライブラリです。 例えば、実行に時間がかかることが多いdbを利用したテストにマーカーをつけておいて、そのテストを避けてテストをするなどが挙げられます。

似たような機能としてpytestにはmarkerの機能が存在しています。

リンク先には、特定のテストにslowというマーカーをつけ、--runslowというオプションを付けなければslowのマーカーが付いたテストは実行しないという機能をどうやって実装するのかについての説明があります。

おおよそやりたいことはこれと同じようなことです。

testmarker?

pytestのmarkerの機能はただただマーカーをつけるだけの汎用的な機能なのですが。作ったライブラリのtestmarkerはもっと機能を絞っています。具体的には上の例であげたような実行される(skipされる)テストケースを指定することに用途を絞っています。

特徴をあげるとすると以下の様になります。

  • 標準ライブラリのみに依存
  • マーカーの利用はテストの実行/skipの指定に限定

install方法

installは通常通り以下です。

pip install testmarker

使いかた

使いかたを以下の2つの観点に分けて説明します。

  • テストの作成
  • テストの実行

テストの作成

テストの作成時には以下のように、testmarker.markを利用してテストケースやテストメソッドにマーカーを指定していきます。

test_it.py

import unittest
from testmarker import mark


@mark.a
class Test0(unittest.TestCase):
    def test_it(self):
        pass


class Test1(unittest.TestCase):
    @mark.a
    def test_it(self):
        pass

    @mark.b
    def test_it2(self):
        pass


class Test2(unittest.TestCase):
    def test_it(self):
        pass

上のコードでは以下2つのマーカーを指定しています。

  • a
  • b

テストの実行

テストの実行については説明するべきことが幾つかあります。マーカーの利用方法がいくつかあります。

  • 環境変数によるマーカーの指定
  • python -m testmarker discover での利用
  • python setup.py test からの利用

環境変数によるマーカーの指定

マーカーを指定するとそのマーカーの名前に対応した環境変数を通じてテストのskipを指定できます。例えば上の例ではaというマーカーを利用していたので以下の様に、NO_A=1という環境変数を指定して呼ぶことで、aのテストをスキップさせることができます(何がスキップされたかわかりやすいようにverboseオプションを付けています)。

$ NO_A=1 python -m unittest discover tests --verbose
test_it (test_it.Test0) ... skipped 'a'
test_it (test_it.Test1) ... skipped 'a'
test_it2 (test_it.Test1) ... ok
test_it (test_it.Test2) ... ok

----------------------------------------------------------------------
Ran 4 tests in 0.000s

OK (skipped=2)

同様に、NO_B=1などとしてあげるとbのマーカーが指定されていたテストをスキップできます。

$ NO_A=1 NO_B=1 python -m unittest discover tests --verbose
test_it (test_it.Test0) ... skipped 'a'
test_it (test_it.Test1) ... skipped 'a'
test_it2 (test_it.Test1) ... skipped 'b'
test_it (test_it.Test2) ... ok

----------------------------------------------------------------------
Ran 4 tests in 0.000s

OK (skipped=3)
デフォルトスキップのテストを有効にする

逆もまたできます。今までのマーカーはデフォルトの実行ではテスト対象に含まれていましたが、環境変数の指定によりスキップを行っていました。 逆に、デフォルトではスキップするテストケースを定義しておき、環境変数の指定により実行を許可するという形にもできます。

このときには、mark()時にskipオプションを付けてください

test_it2.py

import unittest
from testmarker import mark


@mark("x", skip=True)
class Tests(unittest.TestCase):
    def test_it(self):
        pass

デフォルトではスキップされます。

$ python -m unittest discover tests2 --verbose
test_it (test_it.Tests) ... skipped 'x'

----------------------------------------------------------------------
Ran 1 test in 0.000s

OK (skipped=1)

xというマーカー名に対応したX=1というオプションを付けてあげるとスキップせずテストを実行してくれます。 (Xというのはマーカー名をstr.upper()した文字列です)

$ X=1 python -m unittest discover tests2 --verbose
test_it (test_it.Tests) ... ok

----------------------------------------------------------------------
Ran 1 test in 0.000s

OK

まとめると以下の様になります

envname description
NO_<MARKER NAME> 対応するマーカーが指定されたテスト対象をスキップ
<MARKER NAME> 対応するマーカーが指定されたテストを実行対象に含める

python -m testmarker discover による実行

順番は前後してしまいますが。pythonの標準ライブラリのunittestを-m付きでコマンドラインから実行するとテストの実行を行うことができます。この機能と同様のインターフェイスpython -m testmarker discoverで実行できるようにしてみました。使えるオプションとして以下2つのオプションが増えます。

  • --ignore
  • --only
--ignore によるmarkerの指定

--ignore は実行から除外するmarkerの指定です。先ほどの環境変数を介した例でのNO_<MARKER NAME>と同様です。 例えば、a,bのマーカーの除外は以下の様にすることでも行えます。

$ python -m testmarker discover tests --verbose --ignore=a,b
test_it (test_it.Test0) ... skipped 'a'
test_it (test_it.Test1) ... skipped 'a'
test_it2 (test_it.Test1) ... skipped 'b'
test_it (test_it.Test2) ... ok

----------------------------------------------------------------------
Ran 4 tests in 0.000s

OK (skipped=3)

## 以下の様にも書ける
$ python -m testmarker discover tests --verbose --ignore=a --ignore=b
--only によるmarkerの指定

--only は逆にこのオプションによって指定されたマーカーのみをテストの実行対象にするオプションです。--only--ignoreを同時に指定することはできません。

そして --onlyの実行で特殊なのはマーカーが設定されていないtest対象の扱いです。onlyというのはそれだけという意味なので、markerが指定されていなかったテストもまたスキップされます。

$ python -m testmarker discover tests --verbose --only=a
test_it (test_it.Test0) ... ok
test_it (test_it.Test1) ... ok
test_it2 (test_it.Test1) ... skipped 'b'
skipped 'Test2 is skipped by --only option'

----------------------------------------------------------------------
Ran 3 tests in 0.000s

OK (skipped=2)

元のコードでは、何もマーカーが指定されていなかったTest2が実行されていません。元のコードは以下の様なものです。

import unittest
from testmarker import mark


@mark.a
class Test0(unittest.TestCase):
    def test_it(self):
        pass


class Test1(unittest.TestCase):
    @mark.a
    def test_it(self):
        pass

    @mark.b
    def test_it2(self):
        pass


class Test2(unittest.TestCase):
    def test_it(self):
        pass

aでmarkされているTest0Test1.test_itだけが実行されています。

python setup.py test からの利用

これはおまけ的な機能でそれほど多くの人が使うとは思えませんが。pythonのsetup.py経由での実行でも先程の--only--ignoreが使えるようにできます。setup関数にcmdclassを渡してあげてください。

from setuptools import setup, find_packages
from testmarker.setupcmd import test

setup(
    name='foo',
    version='0.0',
    description='-',
    packages=find_packages(exclude=["foo.tests"]),
    test_suite="foo.tests",
    cmdclass={"test": test}
)

このようにすると。以下のような記述が可能になります。

$ python setup.py test --only=a

動作する実行例はexamplesにあります。

marshmallowでschemaに制限を加えてみる

marshmallowでschemaに制限を加えてみる。この制限をprotocolと呼ぶとしてこのprotocolの実装をどうするかと言うとこんな感じ。 marshmallowの流儀に従うならSchemaOptsを拡張するのが自然そう。

marshmallow_protocol.py

class CustomOpts(SchemaOpts):
    protocol = None


class CustomMeta(SchemaMeta):
    def __init__(self, name, bases, attrs):
        super().__init__(name, bases, attrs)
        k = "_protocol_verified"
        protocol = self.Meta.protocol
        if protocol is not None and k not in self.__dict__:
            protocol(self)
            setattr(self, k, protocol)


class ProtocolError(Exception):
    pass


class Schema(BaseSchema, metaclass=CustomMeta):
    class Meta:
        protocol = None

こんな感じ。marshmallowは何らかの設定(Option)を取る時にMetaというクラスに書くことが多いのだけれど。 (このような実装になっているパッケージは結構多く存在している。一番身近な例でいうとdjangoかもしれない)

このmetaにprotocolという属性を追加する。

protocol

protocolは、schemaを引数として呼べるcallable。やばかったらProtocolErrorの例外が発生する。 ついでに、何度も呼ばれることが無いようcacheする意味合いとどのprotocolで検証されたかを把握するために、_protocol_verifiedというフィールドに利用されたprotocolを埋め込む。

呼ばれるタイミングは、クラス定義時。

利用例

metadataに制限を加えてみようというprotocol

例えば、marshmallowのfields.Fieldはmetadataを付加できる。このmetadataに制限を加えてみようというprotocol。

もう少し問題認識を共有しておくと、まず、marshmallow.fields.Fieldの__init__()は以下の様になっている。

class Field(FieldABC):
    def __init__(self, default=missing_, attribute=None, load_from=None, dump_to=None,
                 error=None, validate=None, required=False, allow_none=None, load_only=False,
                 dump_only=False, missing=missing_, error_messages=None, **metadata):

        # do something()

        self.metadata = metadata

metadataは自由につけられるので例えば以下の様なSchemaを定義してしまえる。

from marshmallow import Schema, fields, validate


class S(Schema):
    v = fields.Integer(validator=validate.Range(max=10))

ちなみに、これはvalidateをvalidatorとtypoしている。このschemaはmetadataとしてvalidate.Rangeのオブジェクトが格納されたvalidな定義。 もちろん、設定したRangeによるvalidateはload時には動かないわけだけれど。

S().load({"v": 1000000000000000})  # エラーは発生しない

このとき、metadataに制限を加えてあげてみることにする。例えば、swaggerなどのようにmetadata的なものをx_ではじまるものだけに絞るというprotocolを定義する。(swaggerの場合はx-ではじまるもの)

from marshmallow.schema import BaseSchema
from marshmallow_protocol import CustomMeta, ProtocolError


def x_vendor_prefix_only_metadata(cls):
    for f in cls._declared_fields.values():
        for k in f.metadata.keys():
            if not k.startswith("x_"):
                raise ProtocolError(k)


class Schema(BaseSchema, metaclass=CustomMeta):
    class Meta:
        protocol = x_vendor_prefix_only_metadata

定義したprotocolがdefaultのSchemaを作り、これを利用してみる。

from marshmallow import fields, validate


class S(Schema):
    # marshmallow_protocol.ProtocolError: validator
    v = fields.Integer(validator=validate.Range(max=10))

validatorはx_で始まっていないのでエラーになる(エラーメッセージが少し雑すぎるかもしれない)。

もちろん、typoしていなければエラーにはならないし。protocolを無効にしてあげればエラーにならない。

from marshmallow import fields, validate



class S2(Schema):
    v = fields.Integer(validate=validate.Range(max=10), x_ja="値")


class Ignored(Schema):
    class Meta:
        protocol = None

    v = fields.Integer(validator=validate.Range(max=10))

おしまい。

kamidanaにバッチモードをつけてみた

github.com

kamidanaにバッチモードをつけた。これは実験的な機能でまだmasterにはマージされていない。 やっぱり一回のコマンドで終わらせたほうが早いので欲しくなってしまった。

ただまだ見通しが良い感じにできていない(そういう意味でもまだ実験的)。

実行時間の比較

雑に単純なテンプレートを3回実行する処理の時間を測る。

個別にレンダリングした場合

個別にレンダリングした場合の時間は以下。

$ time make 00
echo '{"name": "foo"}' | kamidana -i json hello.j2 --dst 00out/foo.hello
echo '{"name": "bar"}' | kamidana -i json hello.j2 --dst 00out/bar.hello
kamidana hello.j2 --data me.json --dst 00out/me.hello

real    0m0.727s
user    0m0.696s
sys     0m0.030s

この時のテンプレートは以下のようなもの。

hello {{name}}

まとめてレンダリングした場合

まとめてバッチモードでレンダリングした場合は以下の様な感じ。kamidanaではなくkamidana-batchコマンドを使う。 jinja2の読み込みなどに時間が掛かるのでやっぱり一回で終わらせたほうが早い。

$ time make 01
kamidana-batch 01batch.json --outdir=01out
INFO:kamidana.driver:out: 01out/foo.hello
INFO:kamidana.driver:out: 01out/bar.hello
INFO:kamidana.driver:out: 01out/me.hello

real    0m0.250s
user    0m0.233s
sys     0m0.017s

この時渡したファイルは以下のようなもの。

01batch.json

[
  {"template": "./hello.j2", "data": {"name": "foo"}, "outfile": "foo.hello"},
  {"template": "./hello.j2", "data": [{"name": "bar"}], "outfile": "bar.hello"},
  {"template": "./hello.j2", "data": "me.json", "outfile": "me.hello"}
]

以下のフィールドは必須

  • template
  • outfile

makefile

作業を行った時のmakefile

default:
  $(MAKE) 00
  $(MAKE) 01
  diff -ur 00out 01out

# onefile
00:
  echo '{"name": "foo"}' | kamidana -i json hello.j2 --dst 00out/foo.hello
  echo '{"name": "bar"}' | kamidana -i json hello.j2 --dst 00out/bar.hello
  kamidana hello.j2 --data me.json --dst 00out/me.hello

# batch
01:
  kamidana-batch 01batch.json --outdir=01out

clean:
  rm -rf 00out 01out