Pythonまとめ(2.標準ライブラリその2)

ロギング

はじめに

Pythonではロギング機能をもつloggingライブラリが標準で提供されている。loggingライブラリを使うにあたり基本となる用語とログ処理の流れを示す。

用語説明サンプル
ロガーログ記録の設定をとりまとめ実行するクラスlogger = logging.getLogger( "FileReader" )
ログレコードログが発生するごとに作成されるログイベント
ハンドラログレコードの処理を行うクラスlogger.addHandler( handler )
フォーマッタログレコードを指定の書式に整形するクラス
フィルタ処理するログレコードを選別するクラスまたは関数logger.addFilter(lambda x: x.getMessage().startswith("!"))
ログレベルログの重大度を表すもの。実態は整数値`logging.INFO
logging関連の基本用語

ログを記録するには以下の手順に従う。なお説明のために順序を問わない手順やオプションの手順も含まれている。

  1. ロガーを取得する
  2. ロガーを設定する
    • ロガーのログレベルを設定する
    • ロガーのフィルタを設定する
  3. ハンドラを生成・設定する
    • フォーマッタを作成し、ハンドラにアタッチする
    • ハンドラのフィルタを作成し、ハンドラにアタッチする
    • ハンドラのログレベルを設定する
    • ロガーにハンドラをアタッチする
  4. ログを記録する

ロガー

Pythonのloggingには名前空間という考えがある。名前はA.B.Cのようにドット区切りの階層構造が利用できる。ログを取得するロガーを取得するときは名前を指定して取得する。名前が同じであれば、別のPythonファイルであっても同じロガーが返される。(もちろん同じPythonインタプリタ上である必要はある)

あなたが画像処理を行うプログラムを書いており、その中でI/O部分の処理に関するログを記録したい。その場合には以下のように「Imaging.IO」という名前のImagingを親とするIO用の子ロガーを取得できる。Imaging.IOという名前のロガーを取得する方法は2通りある。

>>> logger1 = logging.getLogger( "Imaging.IO" )
>>> logger2 = logging.getLogger( "Imaging" ).getChild( "IO" )
>>> logger1 is logger2
True

この階層構造の特徴を使用して、複数ファイルにまたがるプログラムでは、全体で共通な「基本となるロガー名」をmainに相当するプログラムが初期化し、各ファイルではそのロガーを使用すればよい。

import imagingio # imagingioをインポート

def main():
    logger = logging.getLogger( "Imaging" )
    # "Imaging"というロガーの初期設定をここで実施

if __name__ == "__main__":
    main()
logger = logging.getLogger( "Imaging" ) # このロガーはimaging-main.pyで初期設定したロガーと同一
logger.getChild( "IO" ) # 子ロガーを取得し、another.py独自のログ設定を加えることもできる

ロガーにはログを記録するメソッドがログレベルごとに用意されている。警告に相当するログを記録する例を示す。

>>> logger1.warning("Warning!")
Warning!

ハンドラ

ロガーに記録されたログをどのように処理するかはハンドラが決める。この処理には、画面(標準出力)に表示する、ファイルに記録する、シスログに送信するなどが考えられるが、一般的なハンドラがライブラリに用意されている。

ハンドラはロガーに登録して使用する。1つのロガーにたいして複数のハンドラを登録することができるため、メッセージを画面に出力した上でファイルにも記録するという処理も行える。ここでは標準出力に表示するハンドラをロガーに登録するサンプルを示す。他のハンドラは後述する。

logger1 = logging.getLogger("log")
stdouthandler = logging.StreamHandler( stream=sys.stdout ) # ストリームに出力するハンドラStreamHandlerで出力先ストリームは標準出力を指定
logger1.addHandler(stdouthandler) # ハンドラを登録

フォーマッタ

記録されたログがどのように出力されるかはフォーマッタで指定する。例えば「時間 ロガー名 メッセージ」のような書式であったり「時間 ログレベル メッセージ」のような書式が指定できる。フォーマッタはハンドラに設定する。ここではフォーマッタを作成し、ハンドラに設定する部分のみを示し、具体的な設定内容については後述する。

formatter = logging.Formatter(fmt="{asctime}:{name}:{module}.{funcName}.{lineno}:{message}", style='{', datefmt="%Y-%m-%d %H:%M:%S")
stdouthandler.setFormatter(formatter)
# 2023-07-23 02:28:26:log.io.file:WARNING:test3-3 のようなフォーマットになる

フィルタ

ロガーに記録されたログのうち、条件を満たしたログだけを処理するようにフィルタで設定できる。フィルタはロガーとハンドラの両方に設定できる。ここではハンドラにフィルタを設定するサンプルのみを示す。

stdouthandler.addFilter(lambda x:x.getMessage().startswith("test")) # testで始まるログメッセージのみを記録するフィルタ

ハンドラーとロガーの両方にフィルタがある場合は、まずロガーのフィルタが評価される。ロガーのフィルタを通過したログは順に各ハンドラに渡される。各ハンドラでフィルタをさらに適用し、フィルタごとに処理をするかどうか決定する。複数のフィルタが設定されている場合は、すべてがTrueとなるログだけが処理される

ルートロガー

ちょっとしたプログラムならlogging自体をロガーとして使うことができる。loggingはルートロガーである。デフォルトではWARN以上のログが標準エラー出力に表示される。

>>> logging.warning("Warning!")
WARNING:root:Warning!

ロガーを使ってみる

何も設定しなかった場合、WARN以上のログが標準エラー出力にそのまま表示される。

import logging
import sys

logger1 = logging.getLogger("log")

logger1.warning("test1-1")
test1-1

ハンドラの設定

StreamHandlerを設定してみる。StreamHandlerは何も設定しなかった場合、WARN以上のログがストリームにそのまま出力される。

import logging
import sys

logger1 = logging.getLogger("log")
shandler = logging.StreamHandler( stream=sys.stdout )
logger1.addHandler(shandler)

logger1.warning("test2-1")
test2-1

結果だけでは最初と違いがないように見えるが、標準エラー出力と標準出力という点で変わっている。

フォーマッタの設定

ハンドラにフォーマッタを設定することで、ログのフォーマットが変わる。

import logging
import sys

logger1 = logging.getLogger("log")
shandler = logging.StreamHandler( stream=sys.stdout )
formatter = logging.Formatter(fmt="{asctime}:Handler1:{message}", style='{', datefmt="%Y-%m-%d %H:%M:%S")
shandler.setFormatter(formatter)
logger1.addHandler(shandler)

logger1.warning("test3-1")
2023-07-23 02:42:20:Handler1:test3-1

フィルタの設定

ハンドラまたはロガーにフィルタを設定することで、処理するログを選択できる。この例ではハンドラに設定する。

import logging
import sys

logger1 = logging.getLogger("log")
shandler = logging.StreamHandler( stream=sys.stdout )
formatter = logging.Formatter(fmt="{asctime}:Handler1:{message}", style='{', datefmt="%Y-%m-%d %H:%M:%S")
shandler.setFormatter(formatter)
shandler.addFilter(lambda x:x.getMessage().startswith("test")) # testで始まるログだけを記録
logger1.addHandler(shandler)

logger1.warning("test4-1")
logger1.warning("TEST4-2")
2023-07-23 02:45:17:Handler1:test4-1

小文字のtestで始まらない「TEST4-2」は画面に表示されない。

ログレベルの設定

初期設定ではWARN以上だけが表示される。setLevelで取得するログレベルを選択できる。ログレベルの設定はロガーとハンドラの両方に存在する。

import logging
import sys

logger1 = logging.getLogger("log")
shandler = logging.StreamHandler( stream=sys.stdout )
formatter = logging.Formatter(fmt="{asctime}:Handler1:{message}", style='{', datefmt="%Y-%m-%d %H:%M:%S")
shandler.setFormatter(formatter)
shandler.addFilter(lambda x:x.getMessage().startswith("test")) 
logger1.addHandler(shandler)

logger1.info("test5-1")

shandler.setLevel(logging.INFO)
logger1.info("test5-2")

logger1.setLevel(logging.INFO)
logger1.info("test5-3")
2023-07-23 02:47:49:Handler1:test5-3

この例ではログレベルの設定を変えながらINFOのレベルのログを3回記録しているが、処理されるのはtest5-3だけである。test5-1の時点ではロガーとハンドラの両方がWARN以上のログレベル設定であるため処理されず、test5-2ではハンドラはINFO以上であるがロガーがWARN以上であるため処理されない。

プロパゲート

親子関係にあるlogger1とlogger2それぞれにハンドラを設定する。子ロガーであるlogger2でログを発行すると、そのlogger1とlogger2の両方でログが記録される。この子ロガーのログが親ロガーに伝わる(伝搬する)動作をプロパゲートという。

import logging
import sys

# 親ロガー「log」の設定
logger1 = logging.getLogger("log") 
logger1.setLevel(logging.INFO) # ログレベルINFO

# ロガーlogに適用するハンドラの設定
shandler1 = logging.StreamHandler( stream=sys.stdout )
formatter1 = logging.Formatter(fmt="Handler1:{message}", style='{', datefmt="%Y-%m-%d %H:%M:%S")
shandler1.setFormatter(formatter1)
shandler1.addFilter(lambda x:x.getMessage().startswith("test")) # フィルタあり
shandler1.setLevel(logging.INFO) # ログレベルINFO
logger1.addHandler(shandler1)

# 子ロガー「log.io」の設定
logger2 = logging.getLogger("log.io") # logging.getLogger("log").getChild("io")でも同じ
logger2.setLevel(logging.DEBUG) # ログレベルDEBUG

# ロガーlog.ioに適用するハンドラの設定 1つ目
shandler2 = logging.StreamHandler( stream=sys.stdout )
formatter2 = logging.Formatter(fmt="Handler2:{message}", style='{', datefmt="%Y-%m-%d %H:%M:%S")
shandler2.setFormatter(formatter2)
shandler2.setLevel(logging.INFO) # ログレベルINFO
logger2.addHandler(shandler2)

# ロガーlog.ioに適用するハンドラの設定 2つ目
shandler3 = logging.StreamHandler( stream=sys.stdout )
formatter3 = logging.Formatter(fmt="Handler3:{message}", style='{', datefmt="%Y-%m-%d %H:%M:%S")
shandler3.setFormatter(formatter3)
shandler3.setLevel(logging.DEBUG) # ログレベルDEBUG
logger2.addHandler(shandler3)

print('TEST1-1')
logger1.info("test1 1-info")
print('TEST1-2')
logger2.info("test1 2-info")
print('TEST1-3')
logger2.info("TEST1 3-info")
print('\n')
print('TEST2-1')
logger1.debug("test2 1-debug")
print('TEST2-2')
logger2.debug("test2 2-debug")
テストケース結果説明
TEST1-1Handler1:test1 1-infologger1でINFOログを記録。
logger1で記録されたメッセージは子ロガーであるlogger2には流れない。
TEST1-2Handler2:test1 2-info
Handler3:test1 2-info
Handler1:test1 2-info
logger2でINFOログを記録。
logger2のすべてのハンドラで処理されたあとで、親ロガーであるlogger1のハンドラへ渡されている。
TEST1-3Handler2:TEST1 3-info
Handler3:TEST1 3-info
logger2で「TEST」から始まるINFOログを記録。
logger2のすべてのハンドラで処理されたあとで、親ロガーであるlogger1へ渡され、logger1のすべてのハンドラで処理される。
ただしshandler1のフィルタ設定により記録されていない。
TEST2-1(出力なし)logger1でDEBUGログを記録。
logger1のログレベル設定がINFOであるため記録されていない。
TEST2-2Handler3:test2 2-debuglogger2でDEBUGログを記録。
logger2にはハンドラが2つあるが、shandler2はログレベル設定がINFOであるため記録されていない。
その後logger1にも渡されているが、logger1のログレベル設定がINFOであるためshandler1は呼び出されない。
実行結果

このプロパゲートを利用してログを記録すると良い。

  • 自分で大元となるロガー名を決める
    • ルートロガー(logging)は使わない
  • mainで大元のロガーを初期化する
  • 他のファイルは子ロガーを取得してログを取る。カスタマイズしたいときは子ロガーをカスタマイズする。
    • 子ロガーでカスタマイズすることで親ロガーを汚さない

なお、プロパゲートは設定で無効にすることもできる。

補足

ログレベルと発行メソッドの補足

ログにはレベルがあり、各ログレベルに対応する記録メソッドが標準で提供されている。

レベル数値発行メソッド
CRITICAL50Logger.critical(msg, *args, **kwargs)
ERROR40Logger.error(msg, *args, **kwargs)
Logger.exeption(msg) ※except句などで使用
WARNING30Logger.warning(msg, *args, **kwargs)
INFO20Logger.info(msg, *args, **kwargs)
DEBUG10Logger.debug(msg, *args, **kwargs)
NOTSET0Logger.log(0, msg, *args, **kwargs)
※logメソッドはログレベルを指定して記録できる

表に記載のない数字(たとえば15)も使用できる。

フィルタの補足

フィルタは結果がTrue(ノンゼロ)の場合のみ記録される。たとえば特定の文字から始まる場合のみ記録するフィルターは以下のように作成できる。

>>> logger.addFilter(lambda x: x.getMessage().startswith("!!!!"))
>>> logger.warning("!!!") # 記録されない
>>> logger.warning("!!!!")
!!!!

ラムダ式の引数xにはLogRecordオブジェクトが渡される。LogRecordオブジェクトはログが記録するたびにLoggerが生成するオブジェクトであり、メッセージに関わるイベントの情報すべてを含んでいる。

なお、Python3.2より前はFilterクラスを作成したり、filterメソッドを持つクラスを使う必要があったが、現在はfilterとして渡されたオブジェクトが、Filterクラスでなく、filterメソッドを持つクラスでもない場合は、callableとして処理されるようになり、上のようにラムダ式が使用できるようになった。旧スタイルでの実装例を示す。

>>> import logging
>>> 
>>> class StartsWithFilter(logging.Filter): # logging.Filterを継承
...     def __init__(self, keyword):
...         super().__init__()
...         self.keyword = keyword
...     
...     def filter(self, record): # filterメソッドを実装
...         return record.getMessage().startswith(self.keyword)
...
>>> logger = logging.getLogger("TEST")
>>> logger.addFilter(StartsWithFilter("!!"))
>>> logger.warning("!") # 表示されない
>>> logger.warning("!!")
!!
>>> logger.warning("!!!")
!!!
>>>
>>> logger.addFilter(StartsWithFilter("!!!")) # フィルターを追加
>>> logger.warning("!") # 表示されない
>>> logger.warning("!!") # 表示されない
>>> logger.warning("!!!")
!!!

ログの書式の補足

フォーマットの指定スタイルにはprintfスタイル(%)、str.formatスタイル({)、string.Templateスタイル($)の3つがあり、後方互換性のためデフォルトは%になっている。そのためフォーマット文字列の書式を使うにはstyle引数に'{'を指定する。また{asctime}の時刻の書式をdatefmt引数で指定できる。

logger.setFormatter( logging.Formatter(fmt="{asctime}:{message}", style='{', datefmt="%Y-%m-%d %H:%M:%S") )

ログの書式で使用できる主要なものを示す。

属性名意味
asctimeログレコードの作成日時(人間が読める形式)
filenameログを呼び出したファイル名
funcNameログを呼び出した関数名
levelnameログレベルの名前(DEBUG, INFOなど)
linenoログを呼び出したソースコードの行番号
messageログメッセージ本文
moduleログを呼び出したモジュール名
nameロガーの名前

定義されていない属性も指定できるが、その情報をログ記録時に渡す必要がある。渡されない場合に例外が発生するため実装に注意が必要となってしまう。理由がないなら使用しないほうが良いだろう。

>>> logger.setFormatter( logging.Formatter( fmt=r"{asctime} [{user}] %{message}",style='{') ) # 対応していないuserがフォーマットに含まれている
>>> logger.warning("warn", extra={"user":"Inoue"}) # extraでuser情報を渡す
2023-07-18 02:02:50,763 [Inoue] warn
>>> logger.warning("warn") # user情報を渡さずにログ発行
--- Logging error ---
(省略)
KeyError: 'user'

During handling of the above exception, another exception occurred:

ValueError: Formatting field not found in record: 'user'
(省略)

ハンドラ

以下のハンドラがライブラリに用意されている。基本的にはこのハンドラを利用する。ハンドラによってインポートするモジュールが違う点に注意。

ハンドラ名動作import
StreamHandlerストリームに出力するlogging
FileHandlerファイルに出力するlogging
RotatingFileHandlerファイルに出力する。ファイルはサイズやバックアップ数に応じてローテーションする。logging.handlers
TimedRotatingFileHandlerファイルに出力する。ファイルは時間間隔やバックアップ数に応じてローテーションする。logging.handlers
SocketHandlerTCPで送信するlogging.handlers
DatagramHandlerUDPで送信するlogging.handlers
SMTPHandlerメールで送信するlogging.handlers
SysLogHandlersyslogに送信するlogging.handlers
NTEventLogHandlerWindowsのイベントログに送信するlogging.handlers
HTTPHandlerHTTP POSTリクエストとして送信するlogging.handlers
NullHandler何もしない(無視する)logging

個人用テンプレート

個人的によく使うlogging設定

import logging
import logging.handlers
import os

logger = logging.getLogger("log")
handler = logging.handlers.RotatingFileHandler(filename="logfile.log", maxBytes=1000000, backupCount=3, encoding='utf-8')
handler.setFormatter( logging.Formatter(fmt="{asctime}.{msecs:03.0f} {name}:{levelname} {message}", style='{', datefmt="%Y-%m-%d %H:%M:%S") )
handler.setLevel(logging.INFO)
logger.setLevel(logging.DEBUG)
logger.addHandler(handler)
logger = logger.getChild( os.path.basename(__file__) )

ロギング設定ファイル

ある程度のプログラム規模になるとログの挙動は設定ファイルなどで定義できることが望ましい。通常はWARNだけを記録しておくが、想定外の異常が発生したらDEBUG記録を有効化することで、異常の原因分析がプログラム自体をいじることなく柔軟にできる。logging.configモジュールでこれを実現できる。

logging.configには辞書形式で設定をインポートする機能がある。なのでそれに対応したYAML形式等で設定をファイルに記録し、それをインポートすれば良い。設定ファイルを読み込むサンプルを示す。

version: 1
formatters:
  normal:
    format: "{asctime}.{msecs:03.0f} {name}:{levelname} {message}"
    style: "{"
    datefmt: "%Y-%m-%d %H:%M:%S"
handlers:
  console:
    class : logging.StreamHandler
    formatter: normal
    level   : INFO
    stream  : ext://sys.stdout
  file:
    class : logging.handlers.RotatingFileHandler
    formatter: normal
    level: INFO
    filename: logfile.log
    maxBytes: 1024
    backupCount: 3
loggers:
  log:
    handlers: [console, file]
    level: INFO
import logging
import logging.handlers
import logging.config
import yaml

with open('logconfig.yaml') as file:
    config = yaml.safe_load(file.read())

logging.config.dictConfig(config) # 辞書を元にロギングを設定

logger = logging.getLogger("log")
logger.warning("警告")

ロギング設定ファイルの書き方

上のサンプルを見ればだいたい分かるが、主にはformatters、handlers、filters、loggersのセクションがあり、各セクションの中で必要な定義を行う。基本的には以下の通りとなる。

TYPE: # 設定するものの種類(主にformattersかhandlersかfiltersかloggers)
  NAME1: # 設定するものの名前(変数名と思えば良い)
    key1 : value1 # NAME1の設定1
    key2 : value2 # NAME1の設定2
  NAME2: # 設定するものの名前(変数名と思えば良い)
    key1 : value1 # NAME2の設定1
    key2 : value2 # NAME2の設定2

以下であればnormalという名前のフォーマッタを定義している。normalというフォーマッタのフォーマットは「{asctime}.{msecs:03.0f} {name}:{levelname} {message}」、フォーマット文字の書式は「{」という具合だ。

formatters:
  normal:
    format: "{asctime}.{msecs:03.0f} {name}:{levelname} {message}"
    style: "{"
    datefmt: "%Y-%m-%d %H:%M:%S"

こうすると、ハンドラを設定するときに上で定義した「normal」フォーマッタを参照できる。

handlers:
  console:                        # ハンドラ名
    class : logging.StreamHandler # ハンドラのクラス名
    formatter: normal             # ハンドラが使用するフォーマッタ
    level   : INFO                # ハンドラのログレベル
    stream  : ext://sys.stdout    # ハンドラ依存の個別設定は「__init__メソッドのキーワード引数:設定値」の書式で書く

そして最後にロガーを設定する。ここでも上で作成したハンドラ名を指定する。ハンドラ名、フォーマッタ名は設定ファイル内で統一されていればなんでも良いが、ロガー名はロガーの名前空間を表すものであため、実際に使うものにする必要がある。

loggers:
  log.io.a:                       # ロガー名( logging.getLogger()で指定する名前 )
    handlers: [console, file]     # ハンドラ名のリスト
    level: INFO                   # ハンドラのログレベル

なお、将来の互換性のためにversionが必須となっているが、2023年現在で指定できるバージョンは1だけだ。

すべてのスキーマはhttps://docs.python.org/ja/3/library/logging.config.html#logging-config-dictschemaにある。

fileConfig()

設定をファイルから直接読み込むfileConfigメソッドもある。ただし、dictConfigのほうが新しいAPIであり推奨されている。

ipaddress

ipaddressはIPアドレスに関する操作やヘルパー関数が含まれる標準ライブラリである。ここではIPv4に重点を置いて説明する。

主要クラス

ipaddressライブラリにはIPアドレス、ネットワーク、ネットワークインタフェースに対応するクラスがある。

種類特徴対応する値の例対応しない値の例
IPアドレス
ipaddress.IPv4Address
ipaddress.IPv6Address
・ネットワークアドレス長を持たない192.168.1.1
192.168.0.0
・ネットワークアドレス長を含む値
192.168.1.0/24
192.168.1.1/32
ネットワークアドレス
ipaddress.IPv4Network
ipaddress.IPv6Network
・ネットワークアドレス長を持つ
・ホストビットが0である
192.168.0.0
192.168.1.0/24
・ホストビットが0でない
192.168.1.1/24
ネットワークインタフェース
ipaddress.IPv4Interface
ipaddress.IPv6Interface
・ネットワークアドレス長を持つ192.168.1.10
192.168.1.0/24
192.168.1.1/24
特になし

IPv4Address

IPアドレスを表すクラスとしてipaddress.IPv4Addressとipaddress.IPv6Addressがある。それぞれのクラスを直接作成しても良いが、ipaddress.ip_addressというIPv4/IPv6を区別しなくても良いメソッドもある。

import ipaddress

ip1 = ipaddress.IPv4Address("192.168.1.1")
print(repr(ip1))

ip2 = ipaddress.ip_address("192.168.2.1")
print(repr(ip2))
IPv4Address('192.168.1.1')
IPv4Address('192.168.2.1')

IPアドレスはネットワークに関する情報を持たない。そのためCIDR表記のアドレスを指定するとエラーになる。

ipaddress.IPv4Address("192.168.1.0/24")
Exception has occurred: AddressValueError
Unexpected '/' in '192.168.1.0/24'

IPv4アドレスは以下の属性を持つ。

reverse_pointer逆引きのレコード名。1.1.168.192.in-addr.arpa など
is_globalIPアドレスがグローバルIPアドレスの場合True
is_link_localIPアドレスがリンクローカルアドレスの場合True
is_loopbackIPアドレスがループバックアドレスの場合True
is_multicastIPアドレスがマルチキャストアドレスの場合True
is_privateIPアドレスがプライベートIPアドレスの場合True
is_reservedIPアドレスがIETFで予約されているアドレスの場合True
is_unspecifiedIPアドレスがRFCで未定義の場合True

IPv4Network

ネットワークを表すクラスとしてipaddress.IPv4Networkとipaddress.IPv6Networkがある。それぞれのクラスを直接作成しても良いが、ipaddress.ip_networkというIPv4/IPv6を区別しなくても良いメソッドもある。ネットワークアドレス長はCIDR表記かマスクで指定する。指定されなかった場合は/32とみなされる。

# 以下はすべておなじになる
ipaddress.ip_network("192.168.2.0/24")
ipaddress.ip_network("192.168.2.0/255.255.255.0")
ipaddress.ip_network("192.168.2.0/0.0.0.255")
ipaddress.ip_network(("192.168.2.0", 24))
ipaddress.ip_network(("192.168.2.0", "255.255.255.0"))
ipaddress.ip_network(("192.168.2.0", "0.0.0.255"))

# intでもOK
ipaddress.ip_network((3232236032, 24))

ネットワークを表すクラスのため、ホストアドレス部は0でなければならない。0でない場合ValueErrorが発生する。

ip1 = ipaddress.ip_network("192.168.2.1/24")
Exception has occurred: ValueError
192.168.2.1/24 has host bits set

ネットワークアドレスに関する属性が利用できる。

with_hostmaskホストマスク、ワイルドカードマスク付きの表記192.168.1.1/0.0.0.255
with_netmaskネットマスク付きの表記192.168.1.1/255.255.255.0
with_prefixlenCIDR表記192.168.1.1/24
network_addressネットワークアドレス192.168.1.0
broadcast_addressブロードキャストアドレス192.168.1.255
netmaskサブネットマスク255.255.255.0
prefixlenプレフィックス長24
reverse_pointer逆引きするときのレコード名0/24.1.168.192.in-addr.arpa

ネットワーク演算

サブネットやスーパーネットなどを取得するメソッドが提供されている。

メソッド名引数戻り値説明
subnetsnew_prefix:新しいサブネット長の指定
prefixlen_diff:相対的な新しいサブネット長の指定
指定のサブネット長に分割したサブネットのgenerator現在のネットワークを指定のサブネット長に分割する。
指定がない場合prefixlen_diff=1の動作になる。
結果は複数のサブネットになるため、generatorで得られる。
supernetnew_prefix:新しいサブネット長の指定
prefixlen_diff:相対的な新しいサブネット長の指定
指定のサブネット長のスーパーネット指定のサブネット長のスーパーネットを取得する。
指定がない場合prefixlen_diff=1の動作になる。
address_exclude除外するサブネット指定のサブネットを除外したサブネットのgenerator現在のネットワークから指定のサブネットを除外する

subnets

ip1 = ipaddress.ip_network("192.168.2.0/24")

print( list(ip1.subnets()) )
print( list(ip1.subnets(prefixlen_diff=2) ) )
print( list(ip1.subnets(new_prefix=27)) )
[IPv4Network('192.168.2.0/25'), IPv4Network('192.168.2.128/25')]
[IPv4Network('192.168.2.0/26'), IPv4Network('192.168.2.64/26'), IPv4Network('192.168.2.128/26'), IPv4Network('192.168.2.192/26')]
[IPv4Network('192.168.2.0/27'), IPv4Network('192.168.2.32/27'), IPv4Network('192.168.2.64/27'), IPv4Network('192.168.2.96/27'), IPv4Network('192.168.2.128/27'), IPv4Network('192.168.2.160/27'), IPv4Network('192.168.2.192/27'), IPv4Network('192.168.2.224/27')]

supernet

ip1 = ipaddress.ip_network("192.168.2.0/24")

print( ip1.supernet() )
print( ip1.supernet(prefixlen_diff=2) )
print( ip1.supernet(new_prefix=21) )
192.168.2.0/23
192.168.0.0/22
192.168.0.0/21

address_exclude

net1 = ipaddress.IPv4Network('192.168.1.0/24')
net2 = ipaddress.IPv4Network('192.168.0.0/16')

exclude=net2.address_exclude(net1)
print( sorted(list(exclude)) )
[IPv4Network('192.168.0.0/24'), IPv4Network('192.168.2.0/23'), IPv4Network('192.168.4.0/22'), IPv4Network('192.168.8.0/21'), IPv4Network('192.168.16.0/20'), IPv4Network('192.168.32.0/19'), IPv4Network('192.168.64.0/18'), IPv4Network('192.168.128.0/17')]

比較演算

ネットワークがあるネットワークのサブネット/スーパーネットか判定するメソッドが用意されている。これで包含判定が行える。このメソッドは等しくてもTrueとなる。

メソッド名引数戻り値
subnet_ofother:比較対象のネットワークネットワークが、比較対象のネットワークのサブネットまたは同じネットワークならTrue
supernet_ofother:比較対象のネットワークネットワークが、比較対象のネットワークのスーパーネットまたは同じネットワークならTrue
overlapsother:比較対象のネットワークネットワークが、比較対象のネットワークと重なっている部分があるならTrue
net1 = ipaddress.IPv4Network('192.168.1.0/24')
net2 = ipaddress.IPv4Network('192.168.2.0/24')
net3 = ipaddress.IPv4Network('192.168.0.0/16')
net4 = ipaddress.IPv4Network('192.0.0.0/8')

print( f"{net1.subnet_of(net1) = }" )
print( f"{net1.subnet_of(net2) = }" )
print( f"{net1.subnet_of(net3) = }" )

print( f"{net3.supernet_of(net1) = }" )
print( f"{net3.supernet_of(net2) = }" )
print( f"{net3.supernet_of(net3) = }" )
print( f"{net3.supernet_of(net4) = }" )
net1.subnet_of(net1) = True     # net1はnet1のサブネットである(net1はnet1に含まれる)
net1.subnet_of(net2) = False    # net1はnet2のサブネットである(net1はnet2に含まれない)
net1.subnet_of(net3) = True     # net1はnet3のサブネットである(net1はnet1に含まれる)

net3.supernet_of(net1) = True   # net3はnet1のスーパネットである
net3.supernet_of(net2) = True   # net3はnet1のスーパネットである
net3.supernet_of(net3) = True   # net3はnet1のスーパネットである
net3.supernet_of(net4) = False  # net3はnet1のスーパネットでない

なお、ネットワーク同士が重なっているか判定するoverlapsメソッドも用意されている。

net1 = ipaddress.IPv4Network('192.168.1.0/24')
net2 = ipaddress.IPv4Network('192.168.2.0/24')
net3 = ipaddress.IPv4Network('192.168.0.0/16')

print( f"{net1.overlaps(net2) = }" )
print( f"{net1.overlaps(net3) = }" )
net1.overlaps(net2) = False
net1.overlaps(net3) = True

IPv4Interface

インタフェースを表すクラスとしてipaddress.IPv4Interfaceとipaddress.IPv6Interfaceがある。それぞれのクラスを直接作成しても良いが、ipaddress.ip_interfaceというIPv4/IPv6を区別しなくても良いメソッドもある。ネットワークアドレス長はCIDR表記かマスクで指定する。指定されなかった場合は/32とみなされる。

# 以下はすべておなじになる
ipaddress.ip_interface("192.168.2.1/24")
ipaddress.ip_interface("192.168.2.1/255.255.255.0")
ipaddress.ip_interface("192.168.2.1/0.0.0.255")
ipaddress.ip_interface(("192.168.2.1", 24))
ipaddress.ip_interface(("192.168.2.1", "255.255.255.0"))
ipaddress.ip_interface(("192.168.2.1", "0.0.0.255"))

# intでもOK
ipaddress.ip_interface((3232236033, 24))

IPv4InterfaceはIPv4AddressとIPv4Networkの両方の特性を併せ持つクラスである。ip属性とnetwork属性で両方のオブジェクトを取得できる。

ipネットワーク情報を持たないAddressの情報<class 'ipaddress.IPv4Address'>
networkインタフェースが属するNetworkの情報<class 'ipaddress.IPv4Network'>
with_prefixlenCIDR表記192.168.1.1/24
with_netmaskネットマスク付きの表記192.168.1.1/255.255.255.0
with_hostmaskホストマスク、ワイルドカードマスク付きの表記192.168.1.1/0.0.0.255
reverse_pointer逆引きするときのレコード名10/24.1.168.192.in-addr.arpa
ipi = ipaddress.IPv4Interface('192.168.1.10/24')

print( type(ipi.ip) )
print( type(ipi.network) )
<class 'ipaddress.IPv4Address'>
<class 'ipaddress.IPv4Network'>

その他の演算・メソッド

IPアドレスとネットワークの包含判定

IPアドレスがネットワークに含まれているか確認するにはin演算子を使う。

ipa = ipaddress.IPv4Address('192.168.1.10')
net1 = ipaddress.IPv4Network('192.168.1.0/24')
net2 = ipaddress.IPv4Network('192.168.2.0/24')

print( f"{ipa in net1=}" )
print( f"{ipa in net2=}" )
ipa in net1=True
ipa in net2=False

順序判定(大小判定)

同じクラスであれば不等号で大小比較が行える。そのためsortも特に意識することなく行える。

ip1, ip2, ip3, ip4  = list(map( ipaddress.IPv4Address, ('192.168.1.1','192.168.2.1','192.168.2.5','192.169.1.0') ))
ni1, ni2, ni3, ni4  = list(map( ipaddress.IPv4Interface, ('192.168.1.1/24','192.168.2.1/24','192.168.2.5/24','192.169.1.0/24') ))
net1,net2,net3,net4 = list(map( ipaddress.IPv4Network, ('192.168.1.0/24', '192.168.2.0/24', '192.168.0.0/16', '192.0.0.0/8') ))

print( f"{ip1 < ip1 = }" )
print( f"{ip1 < ip2 = }" )
print( f"{ip1 < ip3 = }" )
print( f"{ip1 < ip4 = }" )

print( f"{ni1 < ni1 = }" )
print( f"{ni1 < ni2 = }" )
print( f"{ni1 < ni3 = }" )
print( f"{ni1 < ni4 = }" )

print( f"{net1 < net1 = }" )
print( f"{net1 < net2 = }" )
print( f"{net1 < net3 = }" )
print( f"{net1 < net4 = }" )
ip1 < ip1 = False
ip1 < ip2 = True
ip1 < ip3 = True
ip1 < ip4 = True

ni1 < ni1 = False
ni1 < ni2 = True
ni1 < ni3 = True
ni1 < ni4 = True

net1 < net1 = False
net1 < net2 = True
net1 < net3 = False
net1 < net4 = False

IPアドレスとネットワークは種類が違うため通常は順序判定の対象となるものではないが、それでもソートしたい場合などはipaddress.get_mixed_type_keyメソッドをソートのキーとして利用できる。

アドレス集約

ipaddress.collapse_addressメソッドで連続するネットワークを集約することができる。戻り値はgeneratorである。

net1 = ipaddress.IPv4Network('192.168.0.0/24')
net2 = ipaddress.IPv4Network('192.168.1.0/24')
net3 = ipaddress.IPv4Network('192.168.4.0/24')
net4 = ipaddress.IPv4Network('192.168.0.0/16')
list1 = [net1,net2]
list2 = [net1,net2,net3]
list3 = [net1,net4]

print( "list( ipaddress.collapse_addresses(list1) )" )
print( "list( ipaddress.collapse_addresses(list2) )" )
print( "list( ipaddress.collapse_addresses(list3) )" )
[IPv4Network('192.168.0.0/23')]
[IPv4Network('192.168.0.0/23'), IPv4Network('192.168.4.0/24')]
[IPv4Network('192.168.0.0/16')]

ネットワークの算出

先頭と末尾のアドレスから、そのアドレス範囲をカバーできるネットワークアドレスのリストを求める。戻り値はgeneratorである。

ipa1 = ipaddress.IPv4Address("192.168.10.0")
ipa2 = ipaddress.IPv4Address("192.168.10.16")
ipa3 = ipaddress.IPv4Address("192.168.10.63")
ipa4 = ipaddress.IPv4Address("192.168.10.64")

print( list( ipaddress.summarize_address_range(ipa1,ipa2) ) )
print( list( ipaddress.summarize_address_range(ipa1,ipa3) ) )
print( list( ipaddress.summarize_address_range(ipa1,ipa4) ) )
[IPv4Network('192.168.10.0/28'), IPv4Network('192.168.10.16/32')]
[IPv4Network('192.168.10.0/26')]
[IPv4Network('192.168.10.0/26'), IPv4Network('192.168.10.64/32')]

DNS(dnspython)

標準機能における名前解決

単純にIPアドレスとFQDNを相互に名前解決できるだけで良いなら、socketライブラリのメソッドを使用できる。

import socket

ipaddr = socket.gethostbyname("example.com") 
print( ipaddr )

fqdn = socket.getfqdn("1.1.1.1")
print( fqdn )
93.184.216.xx
one.one.one.one

それ以上の機能が必要ならdnspythonライブラリを使用する。たとえばMXレコードを取得したり、名前解決するDNSサーバを指定したりだ。

dnspythonの主要クラス

ゾーン、リソースレコード関連

リソースレコードに対応するクラスの概念図を以下に示す。

dns.rrset.RRsetDNSのリソースレコードセット(RRset)に対応するクラス
ドメイン名、タイプ、クラスが同じRdataの集合
dns.zone.Zone複数のNameとそれに対するNodeをマッピングするクラス
dns.node.NodeRdatasetの集合をあらわすクラス
Zoneクラスにおいて1つのNameに対応する複数のクラス、タイプのRdatasetをまとめて管理するために使う
dns.rdataset.RdatasetDNSのリソースレコードセットのドメイン名以外に対応するクラス
タイプ、クラスが同じRdataの集合
dns.name.NameDNSのリソースレコードのドメイン名(owner)に対応するクラス
dns.rdataclass.RdataClassDNSのリソースレコードのクラス(class)に対応するクラス
dns.rdatatype.RdataTypeDNSのリソースレコードのタイプ(type)に対応するクラス
dns.rdata.RdataDNSのリソースレコードのデータ(RDATA)に対応するクラス

名前解決関連

名前解決を実行するスタブリゾルバに関するクラスを以下に示す。

dns.resolver.Resolverスタブリゾルバ(DNSクライアント)
dns.resolver.Answerスタブリゾルバによる名前解決結果

名前解決の実行

名前解決を行うには、Resolverを作成し、resolveメソッドを実行すれば良い。

import dns.resolver
import dns.rdatatype

myresolver = dns.resolver.Resolver()

answer = myresolver.resolve("example.com.",dns.rdatatype.A)
print(answer.rrset)
example.com. 79586 IN A 93.184.216.xx

もし細かい設定が不要ならデフォルトのResolverを使うresolve関数も用意されている。

dns.resolver.resolve("example.com.",dns.rdatatype.A)

ResolverとAnswer

dns.resolver.ResolverはDNSスタブリゾルバである。デフォルトでPOSIXシステムでは /etc/resolv.confファイルの、WindowsシステムではレジストリのDNS設定を使用する。Resolverの属性で動作を変更することができる。ここでは使用するDNSサーバを指定する例を示す。

import dns.resolver
import dns.rdatatype

myresolver = dns.resolver.Resolver()
myresolver.nameservers = ["8.8.8.8", "1.1.1.1"]

name=dns.name.from_text("www.example.com.")
answer = myresolver.resolve(name, dns.rdatatype.A)

resolveメソッドは名前解決を行い、その結果をdns.resolver.Answerで返す。Answerはrrset属性に応答結果のリソースレコードセットを持っているので、応答が見たければAnswer.rrsetを参照する。Answerが持つ主な属性を示す。

属性名クラス説明
qnamedns.name.Nameクエリ名
rdclassintクエリクラス
rdtypeintクエリタイプ
rrsetOptional[dns.rrset.RRset]応答のリソースレコードセット
expiratopnfloat応答の有効期限
responsedns.message.QueryMessage応答メッセージ

DNSの応答を読みやすいテキストとして出力したいならrrset属性を文字列にstr()で変換するだけでよい。AnswerのRdataの値だけを取得したい場合はrrsetにシーケンスの方法でアクセスする。また、Answerオブジェクトにシーケンスの方法でアクセスしても、rrset属性へのシーケンスアクセスと同じ結果になる。

import dns.resolver
import dns.rdatatype
import textwrap

resolver = dns.resolver.Resolver()
name=dns.name.from_text("example.com.")
answer = resolver.resolve(name, dns.rdatatype.NS)

print("answer.rrset:")
print(textwrap.indent(str(answer.rrset), "  ")) # rrsetを文字列に変換
print(f"answer[0]:\n  {answer[0]}")             # Answerオブジェクトにインデックスでアクセス
print(f"answer.rrset[1]:\n  {answer.rrset[1]}") # RRsetオブジェクトにインデックスでアクセス
answer.rrset:
  example.com. 84782 IN NS a.iana-servers.net.
  example.com. 84782 IN NS b.iana-servers.net.
answer[0]:
  a.iana-servers.net.
answer.rrset[1]:
  b.iana-servers.net.

DNSサーバ、ポートの指定

DNSサーバはdns.nameserver.Nameserverクラスで指定する。dns.nameserver.Nameserverは抽象クラスであり、実際はdns.nameserver.Do53Nameserver(通常のDNSサーバ)やdns.nameserver.DoTNameserver(DNS-over-TCP対応のDNSサーバ)、dns.nameserver.DoHNameserver(DNS-over-HTTPS対応のDNSサーバ)を使用する。なおDoHNameserverを使うにはhttpxモジュールが必要である。

import dns.resolver
import dns.nameserver

myresolver = dns.resolver.Resolver()

ns = dns.nameserver.Do53Nameserver(address="1.1.1.1", port=53)
myresolver.nameservers = [ ns ]
answer = myresolver.resolve( "www.example.com.", "A" )

ns = dns.nameserver.DoTNameserver(address="1.1.1.1", port=853)
myresolver.nameservers = [ ns ]
answer = myresolver.resolve( "example.com.", "MX" )

ns = dns.nameserver.DoHNameserver(url="https://cloudflare-dns.com/dns-query")
myresolver.nameservers = [ ns ]
answer = myresolver.resolve( "example.com.", "SOA" )

なお、DoTやDoHを使用しないdns.nameserver.Do53Nameserverで良ければ、IPアドレスを文字列で直接指定してもよい。nameservers属性に文字列でサーバを指定し、ポート番号をnameserver_portsで指定する。nameserver_portsにポート番号の指定がない場合はportの値が使われる。多くの場合はポートは53(デフォルト値)であるため、ポート番号を指定する必要はほとんどない。

import dns.resolver

myresolver = dns.resolver.Resolver()

myresolver.nameservers = [ "1.1.1.1" ]

# サーバごとのポート番号指定。デフォルトで53なのでこの行は実際は不要
myresolver.nameserver_ports["1.1.1.1"] = 53

# Resolverオブジェクトのデフォルトのポート番号指定。デフォルトで53なのでこの行は実際は不要
myresolver.port = 53     
                    
answer = myresolver.resolve( "www.example.com.", "A" )

キャッシュの有効化

Resolverはデフォルトではキャッシュしないが、キャッシュを有効にできる。単純なキャッシュと、キャッシュ数を指定できるLRUキャッシュが用意されている。

name = dns.name.from_text("www.example.com.")
resolver = dns.resolver.Resolver()
resolver.cache = dns.resolver.Cache(cleaning_interval=300)  # 単純なキャッシュの場合
# resolver.cache = dns.resolver.LRUCache(max_size=100)      # LRUキャッシュの場合

answer = resolver.resolve( name, "A" )
print( resolver.cache.get((name,dns.rdatatype.A,dns.rdataclass.IN)).rrset )

time.sleep(3)
answer = resolver.resolve( name, "A" )

print(f"ヒット:{resolver.cache.hits()} ミス:{resolver.cache.misses()}")
www.example.com. 54965 IN A 93.184.216.34
ヒット:2 ミス:2

DNSサフィックスの補完

ドメインを補完する機能を有効にできる。たとえばwwwを名前解決しようとしたら、自動でwww.example.com.を名前解決する機能である。DNSサフィックスの補完を有効にするには、補完するサフィックスのリストをsearch属性で指定してから、resolveメソッドでsearch=Trueとした上で、解決する名前を相対名で指定する。

import dns.resolver
import dns.name

resolver = dns.resolver.Resolver()
resolver.search = [ dns.name.from_text("example.com.") ]

name=dns.name.from_text("www", origin=None)
answer = resolver.resolve(name, dns.rdatatype.A, search=True)

相対名をつくるには、Nameを作成するときにorigin=Noneを指定するか、Name.relativizeメソッドを使う。

# 何も指定しない場合
name1 = dns.name.from_text("www")

# origin=Noneを指定
name2 = dns.name.from_text("www", origin=None)

# relativizeメソッドを使用
base = dns.name.from_text("example.com.")
fqdn = dns.name.from_text("www.example.com.")
name3 = fqdn.relativize(base)

print( f"name1 = {name1!s}" )
print( f"name2 = {name2!s}" )
print( f"name3 = {name3!s}" )
name1 = www.  # デフォルトは.が補完されてしまう
name2 = www   # origin=Noneを明示
name3 = www   # relativizeで作成

またResolver.use_search_by_defaultをTrueに指定すれば、resolveの引数でsearch=Trueと指定しなくてもDNSサフィックスの補完が有効になる。

import dns.resolver
import dns.name

resolver = dns.resolver.Resolver()
resolver.search = [ dns.name.from_text("example.com.") ]
resolver.use_search_by_default = True

name=dns.name.from_text("www", origin=None)
answer = resolver.resolve(name, dns.rdatatype.A)

逆引き

Resolverには逆引き専用のメソッドresolve_addressが用意されている。resolve_addressは逆引きしたいIPアドレスを文字列で指定する。

resolver = dns.resolver.Resolver()
answer = resolver.resolve_address("1.2.3.4")
print(answer.rrset)
1.1.1.1.in-addr.arpa. 196 IN PTR one.one.one.one.

通常のresolveメソッドで逆引きするには、逆引き用のin-addr.arpaアドレスで指定する必要がある。IPアドレスを逆引き用のアドレスに変換するdns.reversename.from_address関数が用意されている。

reversename = dns.reversename.from_address("1.2.3.4")    # 4.3.2.1.in-addr.arpa.が得られる。
answer = resolver.resolve(reversename,dns.rdatatype.PTR)

例外処理

resolverメソッドはいくつかの例外が発生する。

例外クラス例外の状態
dns.resolver.LifetimeTimeoutDNSサーバからの応答が時間内になかった。
dns.resolver.NXDOMAINNXDOMAIN応答が返された。
(その名前のレコードがない)
dns.resolver.NoAnswerNODATA応答が返された。ただし、resolveメソッドの引数でraise_on_no_answer=Trueのときのみ。デフォルトはTrue。
(その名前、クラス、型のレコードはないが他のクラス、型のレコードはある可能性がある)
dns.resolver.NoNameservers使用できるDNSサーバがない。
dns.resolver.YXDOMAIN名前解決の途中でDNS名の長さ制限を超えた。

そのため本番コードでは、これらの例外を適切に扱うべきである。例えば以下のような実装が考えられる。

try:
    resolver.resolve( "example.com", "A" )
except (dns.resolver.NoAnswer, dns.resolver.NXDOMAIN) as e:
    # 問い合わせに対応するレコードがなかった場合の処理
except (dns.resolver.LifetimeTimeout, dns.resolver.NoNameservers, dns.resolver.YXDOMAIN) as e:
    # 名前解決の実行が正しくできなかった場合の処理

発生する例外によって処理が別れないならこれらの基底クラスdns.exception.DNSExceptionを使えば良い。

try:
    resolver.resolve("example.com", "A")
except dns.exception.DNSException as e:
    # ここで例外処理
NoAnswer例外

NoAnswer例外はresolveメソッドの引数でraise_on_no_answer=Trueのとき(デフォルト)のみ発生する。raise_on_no_answer=Falseとすると、例外は発生せずAnswerのrrsetがNoneとなる。

import dns.resolver

resolver = dns.resolver.Resolver()
answer = resolver.resolve("www.example.com", "MX", raise_on_no_answer=False) # このレコードはAレコードはあるがMXレコードはない
print(answer.rrset) # これはNone

QueryMessage

Answerにも含まれるQueryMessageは実際にDNSサーバとやりとりするIDやフラグ、Opcodeなどを表すクラスである。これにより単純な名前解決の結果だけでなく、フラグなどの詳細な確認が行える。

resolver = dns.resolver.Resolver()
answer = resolver.resolve("example.com")
print(answer.response)
id 19594
opcode QUERY
rcode NOERROR
flags QR RD RA
;QUESTION
example.com. IN A
;ANSWER
example.com. 22985 IN A 93.184.216.34
;AUTHORITY
;ADDITIONAL

ゾーンファイルやリソースレコード

ここでは下図に示すクラスの概略とサンプルをいくつか示す。詳細はdnsppythonのドキュメントを参照すること。

RdataClass

RdataClassはリソースレコードのクラスを表すクラスであり、IntEnumを継承する列挙型である。dns.rdataclassモジュールに定数が用意されているのでそれを参照しても良い。

print( dns.rdataclass.RdataClass.IN )
print( dns.rdataclass.RdataClass(255) )
print( dns.rdataclass.IN )
RdataClass.IN
RdataClass.ANY
RdataClass.IN

RdataType

RdataTypeはリソースレコードのタイプを表すクラスであり、IntEnumを継承する列挙型である。dns.rdatatypeモジュールに定数が用意されているのでそれを参照しても良い。

print( dns.rdatatype.RdataType.MX )
print( dns.rdatatype.A )
print( dns.rdatatype.RdataType(28) )
RdataType.MX
RdataType.A
RdataType.AAAA

Rdata

Rdataはリソースレコードのデータ部を表すクラスである。データ部はリソースレコードのクラス、タイプによってとる値が異なる。例えばAレコードであれば192.0.2.1のようなIPアドレスになるし、MXレコードであれば10 mx.example.comのような優先度とメールサーバの組み合わせになる。そのためRdataは基底クラスであり、クラスやタイプに応じたdns.rdtype.IN.A.Aやdns.rdtype.ANY.MX.MXなどを使っている。

Rdataを作るにはdns.rdata.from_text関数を使う。

rdatas = [
            dns.rdata.from_text(dns.rdataclass.IN, dns.rdatatype.A,  '192.168.1.15'),
            dns.rdata.from_text(dns.rdataclass.IN, dns.rdatatype.MX, '10 mx.example.com'),
        ]
for rdata in rdatas:
    print(f"{type(rdata)}:{rdata}")
<class 'dns.rdtypes.IN.A.A'>:192.168.1.15
<class 'dns.rdtypes.ANY.MX.MX'>:10 mx.example.com

Rdataのクラス、タイプ、値はそれぞれrdclass属性、rdtype属性、to_textメソッドで取得できる。

print(rdatas[0].rdclass)
print(rdatas[0].rdtype)
print(rdatas[0].to_text())
RdataClass.IN
RdataType.A
192.168.1.15

Rdataset

datasetはクラス、タイプが同じRdataの集合である。Rdataに適用されるTTLの値も持つ。Rdatasetは作成済みのRdataから作る方法とテキストから作る方法がある。それぞれに可変長配列から作成するメソッドとリストから作成するメソッドが用意されている。

import dns.rdata
import dns.rdataset

rdatas = [
    dns.rdata.from_text("IN", "A", "192.0.2.100"),
    dns.rdata.from_text("IN", "A", "192.0.2.101"),
]

# rdataから作成する方法
rdataset1 = dns.rdataset.from_rdata(3600, rdatas[0], rdatas[1])
rdataset2 = dns.rdataset.from_rdata_list(3600,rdatas)  
        
# テキストから作成する方法
rdataset3 =dns.rdataset.from_text("IN","A", 3600, "192.0.2.100", "192.0.2.101")
rdataset4 =dns.rdataset.from_text_list("IN","A", 3600, ["192.0.2.100", "192.0.2.101"])

Rdatasetはイテラブルであり、シーケンスの記法で各Rdataの要素にアクセスできる。

rdataset = dns.rdataset.from_text("IN","A", 3600, "192.0.2.100", "192.0.2.101")
print( rdataset[1] ) # 192.0.2.101となる

Node

NodeはRdatasetのリストである。rdatasets属性がrdatasetのリストである。

Name

Nameはリソースレコードのドメイン名を表すクラスである。Nameはドメイン名をラベルのタプルとして管理する。

import dns.name

name = dns.name.from_text("www.example.com")
for label in name:
    print(label)
b'www'
b'example'
b'com'
b''

一般にピリオドで表されるルートドメインは、Nameでは空のラベルで表される。デフォルトでは自動でルードドメインが補完されるが、originで補完するドメインを指定できる。originをNoneとすると補完されなくなり、ルートドメインを含まない相対的なドメイン名として扱われる。

names = [
    dns.name.from_text("www.example.com."), # ルートドメインを明示した文字列
    dns.name.from_text("www.example.com"),  # ルートドメインを省略した文字列
    dns.name.from_text("www.example.com", origin=None),
    dns.name.from_text("www", origin=None),
]
for name in names:
    print(name)
    for label in name:
        print(label)
    print()
www.example.com.
b'www'
b'example'
b'com'
b''

www.example.com.
b'www'
b'example'
b'com'
b''  # ルートドメインが補完されている

www.example.com
b'www'
b'example'
b'com'

www
b'www'

Nameのメソッド

Nameには階層構造に関連するメソッドが用意されている。なお、サンプルの前提としてnamesリストを以下のように定義済みである。

names = [
    dns.name.from_text("example.com."),
    dns.name.from_text("www.example.com."),
    dns.name.from_text("example.com", origin=None),
    dns.name.from_text("test", origin=None),
]
メソッド説明サンプルサンプルの実行例
parent()ひとつ上の階層のNameを取得する。names[1].parent()example.com.
canonicalize()canonical formに直したNameを取得する。name = dns.name.from_text("EXAMPLE.com")
name.canonicalize()
example.com.
relativize(origin:Name)originからみた相対名を取得する。names[0].relativize(dns.name.root)
name.relativize(names[1])
example.com
www
derelativize(origin:Name)相対名であるNameをoriginをもとにした絶対名を取得する。
originは絶対名でなければならない。
names[3].derelativize(names[0])test.example.com.
concatenate(other:Name)Name同士を結合する。derelativizeと異なりotherは相対名でも良い。
メソッドの呼び出し側が絶対名の場合例外が発生する。
names[3].concatenate(names[0])
names[3].concatenate(names[2])
test.example.com.
test.example.com
split(depth:int)指定の深さでNameを分割する。names[1].split(2)(www.example, com.)
is_absolute()ルートから始まる絶対名であればTrue。
is_subdomain(other:Name)otherのサブドメインであればTrue。同一のときもTrueとなる。
is_superdomain(other:Name)otherのスーパードメインであればTrue。同一のときもTrueとなる。

Name定数

dns.name.rootはルートをあらわすNameである。dns.name.Name("")と等しい。また空を表すNameとしてdns.name.emptyも用意されている。

RRset

RRsetはリソースレコードセットに対応するクラスである。名前解決の応答で用いられている。Rdatasetと同じくRdataから作成する方法とテキストから作成する方法の2つがあり、それぞれにリストの方法と可変長引数の方法がある。

RRsetはリソースレコードのowner、ttl、class、type、RDATAに対応する属性を持つ。RDATAはCollectionで管理されており、for ... inやrrset[x]のような形でRDATAにアクセスできる。

rdatas = [
    dns.rdata.from_text("IN", "A", "192.0.2.100"),
    dns.rdata.from_text("IN", "A", "192.0.2.101"),
]

rrset = dns.rrset.from_rdata("www", 3600, *rdatas)
print( rrset.name )    # owner
print( rrset.ttl )     # ttl
print( rrset.rdclass ) # class
print( rrset.rdtype )  # type
print( rrset.items )   # RDATAのCollection(実態はdict)
for rdata in rrset:
    print( rdata )
www
3600
RdataClass.IN
RdataType.A
{<DNS IN A rdata: 192.0.2.100>: None, <DNS IN A rdata: 192.0.2.101>: None}
192.0.2.100
192.0.2.101

Zone

ZoneはNameとNodeを対応させるクラスである。辞書と考えても良い。

zone_text = """
example.com.     3600    IN SOA  ns1.example.com. (
                                  postmaster.example.com.
                                  2023080001      ; Serial
                                  3600            ; Refresh 1h
                                  900             ; Retry 15m
                                  604800          ; Expire 1w
                                  3600 )          ; Negative cache TTL 15m
example.com.        3600    IN NS   ns-a.example.com.
                            IN NS   ns-b.example.com.
                    1800    IN MX   10 mx-a.example.com.
                            IN MX   20 mx-b.example.com.

ns-a                3600    IN A    192.0.2.100
ns-b                3600    IN A    192.0.2.101
                            IN A    192.0.2.102
                            IN A    192.0.2.103
                            IN A    192.0.2.104
ns-b                 600    IN TXT  "example text"
mx-a                3600    IN A    192.0.2.110
mx-b                3600    IN A    192.0.2.111
www.example.com.    3600    IN A    198.51.100.30
"""

zone:dns.zone.Zone = dns.zone.from_text(zone_text, "example.com.")
for name in zone:
    print( f"\n######### {name} ##########" )
    node = zone[name]
    for rdatasetno,rdataset in enumerate(node,start=1):
        print(f"{rdatasetno}:{rdataset.rdtype!s}")
        print(rdataset)
######### @ ##########
1:RdataType.SOA
3600 IN SOA ns1 postmaster 2023080001 3600 900 604800 3600
2:RdataType.NS
3600 IN NS ns-a
3600 IN NS ns-b
3:RdataType.MX
1800 IN MX 10 mx-a
1800 IN MX 20 mx-b

######### ns-a ##########
1:RdataType.A
3600 IN A 192.0.2.100

######### ns-b ##########
1:RdataType.A
3600 IN A 192.0.2.101
3600 IN A 192.0.2.102
3600 IN A 192.0.2.103
3600 IN A 192.0.2.104
2:RdataType.TXT
600 IN TXT "example text"

######### mx-a ##########
1:RdataType.A
3600 IN A 192.0.2.110

######### mx-b ##########
1:RdataType.A
3600 IN A 192.0.2.111

######### www ##########
1:RdataType.A
3600 IN A 198.51.100.30

Zoneのキーはownerであり、zone[owner]のようにすることでNode単位でアクセスできる。それ以外にもrdata単位やrdataset単位でアクセスすることもできる。

print("iterate_rdatas")
for i in zone.iterate_rdatas():
    print(i)

print("\niterate_rdatasets")
for i in zone.iterate_rdatasets():
    print(i)
iterate_rdatas
(<DNS name @>, 3600, <DNS IN SOA rdata: ns1 postmaster 2023080001 3600 900 604800 3600>)
(<DNS name @>, 3600, <DNS IN NS rdata: ns-a>)
(<DNS name @>, 3600, <DNS IN NS rdata: ns-b>)
(<DNS name @>, 1800, <DNS IN MX rdata: 10 mx-a>)
(<DNS name @>, 1800, <DNS IN MX rdata: 20 mx-b>)
(<DNS name ns-a>, 3600, <DNS IN A rdata: 192.0.2.100>)
(<DNS name ns-b>, 3600, <DNS IN A rdata: 192.0.2.101>)
(<DNS name ns-b>, 3600, <DNS IN A rdata: 192.0.2.102>)
(<DNS name ns-b>, 3600, <DNS IN A rdata: 192.0.2.103>)
(<DNS name ns-b>, 3600, <DNS IN A rdata: 192.0.2.104>)
(<DNS name ns-b>, 600, <DNS IN TXT rdata: "example text">)
(<DNS name mx-a>, 3600, <DNS IN A rdata: 192.0.2.110>)
(<DNS name mx-b>, 3600, <DNS IN A rdata: 192.0.2.111>)
(<DNS name www>, 3600, <DNS IN A rdata: 198.51.100.30>)

iterate_rdatasets
(<DNS name @>, <DNS IN SOA rdataset: [<ns1 postmaster 2023080001 3600 900 604800 3600>]>)
(<DNS name @>, <DNS IN NS rdataset: [<ns-a>, <ns-b>]>)
(<DNS name @>, <DNS IN MX rdataset: [<10 mx-a>, <20 mx-b>]>)
(<DNS name ns-a>, <DNS IN A rdataset: [<192.0.2.100>]>)
(<DNS name ns-b>, <DNS IN A rdataset: [<192.0.2.101>, <192.0.2.102>, <192.0.2.103>, <192.0.2.104>]>)
(<DNS name ns-b>, <DNS IN TXT rdataset: [<"example text">]>)
(<DNS name mx-a>, <DNS IN A rdataset: [<192.0.2.110>]>)
(<DNS name mx-b>, <DNS IN A rdataset: [<192.0.2.111>]>)
(<DNS name www>, <DNS IN A rdataset: [<198.51.100.30>]>)

条件を満たすものを取得するfind_node、find_rrset、find_rdatasetメソッドが用意されている。これらのメソッドは一致するものがなければ例外が発生する。例外を発生させるかわりにNoneを返すget_xxxxメソッドもある。

zone.find_node("www")
zone.find_rdataset("ns-b","A")
zone.find_rrset(dns.name.empty,"NS")
<DNS node 2223993232288>

3600 IN A 192.0.2.101
3600 IN A 192.0.2.102
3600 IN A 192.0.2.103
3600 IN A 192.0.2.104

@ 3600 IN NS ns-a
@ 3600 IN NS ns-b

コメントを残す

メールアドレスが公開されることはありません。 が付いている欄は必須項目です