jqfpyを使ってswagger spec中の$refを取り出す

github.com

かんたんなjqの処理はjqfpyでどうにかなるということがわかったのでもう少し複雑な処理をしてみることにした。今度はswagger中の$refを抽出してみる。 このswagger中の$refを取り出すことは意外と大変な気がしている。

そもそもjqだけでやるのは辛いし。せっかくpythonなので便利なライブラリをimportしてそれで終わらせる。

題材にするのはこのswaggerファイル

$ curl https://raw.githubusercontent.com/everett-toews/a-restful-adventure/gh-pages/design/hypermedia-based/swagger.json > swagger.json
$ cat swagger.json | jqfpy --squash -r 'import dictknife.walkers as w; itr = w.DictWalker(["$ref"]).iterate(get()); [d["$ref"] for _, d in itr]'
#/definitions/Characters
#/definitions/Character
#/definitions/Error
#/definitions/Character
#/definitions/Error
#/definitions/Character
#/definitions/Error
#/definitions/Error
#/definitions/CharacterLocation
#/definitions/Error
#/definitions/Error
#/definitions/Dungeons
#/definitions/Dungeon
#/definitions/Error
#/definitions/Room
#/definitions/Error
#/definitions/Error
#/definitions/Link
#/definitions/Character
#/definitions/Link
#/definitions/Link
#/definitions/Link
#/definitions/Link
#/definitions/Dungeon
#/definitions/Link
#/definitions/Link

sortしたりuniqueにしたりはpythonでやっても良いけれど。パイプでつなげてしまっても良い。

$ cat swagger.json | jqfpy --squash -r 'import dictknife.walkers as w; itr = w.DictWalker(["$ref"]).iterate(get()); [d["$ref"] for _, d in itr]' | sort -u
#/definitions/Character
#/definitions/CharacterLocation
#/definitions/Characters
#/definitions/Dungeon
#/definitions/Dungeons
#/definitions/Error
#/definitions/Link
#/definitions/Room

内部的な話

今回のコードは以下の様な感じになっている。

def _transform(get):
    import dictknife.walkers as w
    itr = w.DictWalker(["$ref"]).iterate(get())
    return [d["$ref"] for _, d in itr]

jqfpy自体はライブラリを呼び出しているだけ。こういうことができるあたりがpythonにして良いという点の１つ。

bufferedがデフォルトに

昨日の記事ではunbufferedがデフォルトということにしていたけれど。 unbufferedの場合には、長めの入力を途中で切ることがあり、それがちょうどJSONとしてvalidだったときに失敗してしまうということが起きたりしてしまった。 そんなわけでちょっとunbufferedをデフォルトにすると困る。なのでbufferedをでふぉるとにすることにした。これはjqと同じ。

以下みたいな形で --unbuffered をつける必要があるようになった。

$ tail -f <log file> | jqfpy --unbuffered '<code>' | grep <pattern>

おまけ

ついでに、どこで$refが使われているかも抽出してみましょう。ちょっと複雑になりますがこれもまだワンライナーでできる範囲なきがします。

$ cat swagger.json | jqfpy 'from collections import defaultdict; import dictknife.walkers as w; D = defaultdict(list); itr = w.DictWalker(["$ref"]).iterate(get()); [D[d["$ref"]].append("/".join([p.replace("/", "~1") for p in path[:-1]])) for path, d in itr]; D'
{
  "#/definitions/Characters": [
    "paths/~1characters/get/responses/200/schema"
  ],
  "#/definitions/Character": [
    "paths/~1characters/post/responses/201/schema",
    "paths/~1characters~1{character_id}/get/responses/200/schema",
    "paths/~1characters~1{character_id}/put/parameters/[]/schema",
    "definitions/Characters/properties/characters/items"
  ],
  "#/definitions/Error": [
    "paths/~1characters/post/responses/default/schema",
    "paths/~1characters~1{character_id}/get/responses/default/schema",
    "paths/~1characters~1{character_id}/put/responses/default/schema",
    "paths/~1characters~1{character_id}/delete/responses/default/schema",
    "paths/~1characters~1{character_id}~1location/get/responses/default/schema",
    "paths/~1characters~1{character_id}~1location/put/responses/default/schema",
    "paths/~1dungeons~1{dungeon_id}/get/responses/default/schema",
    "paths/~1dungeons~1{dungeon_id}~1rooms~1{room_id}/get/responses/400/schema",
    "paths/~1dungeons~1{dungeon_id}~1rooms~1{room_id}/get/responses/default/schema"
  ],
  "#/definitions/CharacterLocation": [
    "paths/~1characters~1{character_id}~1location/get/responses/200/schema"
  ],
  "#/definitions/Dungeons": [
    "paths/~1dungeons/get/responses/200/schema"
  ],
  "#/definitions/Dungeon": [
    "paths/~1dungeons~1{dungeon_id}/get/responses/200/schema",
    "definitions/Dungeons/properties/dungeons/items"
  ],
  "#/definitions/Room": [
    "paths/~1dungeons~1{dungeon_id}~1rooms~1{room_id}/get/responses/200/schema"
  ],
  "#/definitions/Link": [
    "definitions/Character/properties/links/items",
    "definitions/Characters/properties/links/items",
    "definitions/CharacterLocation",
    "definitions/Room/properties/links/items",
    "definitions/Dungeon/properties/links/items",
    "definitions/Dungeons/properties/links/items",
    "definitions/Error/properties/link"
  ]
}

実際に実行されるのは以下の様なコード。

def _transform(get):
    from collections import defaultdict
    import dictknife.walkers as w
    D = defaultdict(list)
    itr = w.DictWalker(["$ref"]).iterate(get())
    [D[d["$ref"]].append("/".join([p.replace("/", "~1") for p in path[:-1]])) for path, d in itr]
    return D