🐎

urllib を拡張する

2022/09/10に公開

例えば rdflib で HTTP 越しに SPARQL クエリしたいとき。一般的には公開 SPARQL エンドポイントでクエリする例がよく出ている。このように平文通信でアクセスできるなら問題ないけれども、認証がかかっていると工夫がいる。ドキュメントによると urllib を使っているから、よろしくやるように書かれている。

あらかじめ用意されている auth パラメータで HTTP Basic 認証は対応できる。これ以上、たとえば、未認証の場合に別のサイトにリダイレクトで飛ばされて、そこで認証が終わってから戻されるといったフローの場合は下準備がいる。

どのようにするかあまり例がなかったので、ここで挙げておこうと思う。Authelia で認証ゲートウェイが構成されている場合の例になる。

http://wikidata-query.example.jp/ が SPARQL エンドポイントのホスト
http://authelia.example.jp/ が authelia 認証ページ
authelia first factor での認証を行う

まず urllib にハンドラを追加して、モジュールにインストールする。

import urllib
import urllib.parse
import ssl
import json

class AutheliaHandler(urllib.request.BaseHandler):
    handler_order = 490
    def __init__(self, auth_site, username, password):
        self.auth_site = auth_site
        self.username = username
        self.password = password
    
    def http_error_302(self, req, fp, code, msg, headers):
        newurl = headers.get("location", "")
        if newurl.startswith(self.auth_site):
            newurl = urllib.parse.urljoin(newurl, "/api/firstfactor")
            auth = urllib.request.Request(
                newurl,
                method="POST",
                headers={"Content-type":"application/json"},
                data=json.dumps(dict(
                    username=self.username,
                    password=self.password,
                )).encode("UTF-8")
            )
            if self.parent.open(auth).status == 200:
                return self.parent.open(req)

ctx = ssl.SSLContext()
ctx.load_verify_locations("chain.cer")
opener = urllib.request.build_opener(
    urllib.request.HTTPSHandler(context=ctx),
    urllib.request.HTTPCookieProcessor(),
    AutheliaHandler("https://authelia.example.jp/", "username", "password"))
urllib.request.install_opener(opener)

ここまでで urllib に細工を施したことになる。次にいつも通り rdflib から SPARQL クエリを発行する。結果が得られれば成功。

import rdflib
g = rdflib.Graph()
res = g.query('''
PREFIX wdt: <https://wikidata.example.jp/prop/direct/>

SELECT ?s WHERE {
  SERVICE <https://wikidata-query.example.jp/proxy/wdqs/bigdata/namespace/wdq/sparql> {
    ?s wdt:P1 ?o .
  }
}
LIMIT 3
''')
for r in res:
    print(r.s)

Discussion