Skip to content

Configuration Classes

Configuration classes for creating BGP streams.

BGPStreamConfig

pybgpflux.bgpstreamconfig.BGPStreamConfig pydantic-model

Bases: BaseModel

Unified BGPStream config

Show JSON schema: 
{
  "$defs": {
    "FilterOptions": {
      "description": "A unified model for the available filter options.",
      "properties": {
        "origin_asn": {
          "anyOf": [
            {
              "type": "integer"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "description": "Filter by the origin AS number.",
          "title": "Origin Asn"
        },
        "prefix": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "description": "Filter by an exact prefix match.",
          "title": "Prefix"
        },
        "prefix_super": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "description": "Filter by the exact prefix and its more general super-prefixes.",
          "title": "Prefix Super"
        },
        "prefix_sub": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "description": "Filter by the exact prefix and its more specific sub-prefixes.",
          "title": "Prefix Sub"
        },
        "prefix_super_sub": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "description": "Filter by the exact prefix and both its super- and sub-prefixes.",
          "title": "Prefix Super Sub"
        },
        "peer_ip": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "format": "ipv4",
              "type": "string"
            },
            {
              "format": "ipv6",
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "description": "Filter by the IP address of a single BGP peer.",
          "title": "Peer Ip"
        },
        "peer_ips": {
          "anyOf": [
            {
              "items": {
                "anyOf": [
                  {
                    "type": "string"
                  },
                  {
                    "format": "ipv4",
                    "type": "string"
                  },
                  {
                    "format": "ipv6",
                    "type": "string"
                  }
                ]
              },
              "type": "array"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "description": "Filter by a list of BGP peer IP addresses.",
          "title": "Peer Ips"
        },
        "peer_asn": {
          "anyOf": [
            {
              "type": "integer"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "description": "Filter by the AS number of the BGP peer.",
          "title": "Peer Asn"
        },
        "update_type": {
          "anyOf": [
            {
              "enum": [
                "withdraw",
                "announce"
              ],
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "description": "Filter by the BGP update message type.",
          "title": "Update Type"
        },
        "as_path": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "description": "Filter by a regular expression matching the AS path.",
          "title": "As Path"
        },
        "ip_version": {
          "anyOf": [
            {
              "enum": [
                4,
                6
              ],
              "type": "integer"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "description": "Filter by ip version.",
          "title": "Ip Version"
        }
      },
      "title": "FilterOptions",
      "type": "object"
    }
  },
  "description": "Unified BGPStream config",
  "properties": {
    "start_time": {
      "anyOf": [
        {
          "format": "date-time",
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "Start of the stream",
      "title": "Start Time"
    },
    "end_time": {
      "anyOf": [
        {
          "format": "date-time",
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "End of the stream",
      "title": "End Time"
    },
    "collectors": {
      "description": "List of collectors to get data from",
      "items": {
        "type": "string"
      },
      "title": "Collectors",
      "type": "array"
    },
    "data_types": {
      "default": [
        "updates"
      ],
      "description": "List of archives files to consider (`ribs` or `updates`)",
      "items": {
        "enum": [
          "ribs",
          "updates"
        ],
        "type": "string"
      },
      "title": "Data Types",
      "type": "array"
    },
    "filters": {
      "anyOf": [
        {
          "$ref": "#/$defs/FilterOptions"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "Optional filters"
    },
    "max_concurrent_downloads": {
      "anyOf": [
        {
          "type": "integer"
        },
        {
          "type": "null"
        }
      ],
      "default": 10,
      "description": "Maximum concurrent downloads of archive files.",
      "title": "Max Concurrent Downloads"
    },
    "cache_dir": {
      "anyOf": [
        {
          "format": "directory-path",
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "Specifies the directory for caching downloaded files.",
      "title": "Cache Dir"
    },
    "ram_fetch": {
      "anyOf": [
        {
          "type": "boolean"
        },
        {
          "type": "null"
        }
      ],
      "default": false,
      "description": "If caching is disabled, fetch temp files in shared RAM memory (/dev/shm) or normal disc temp dir (/tmp).Default (False) to reduce RAM usage.",
      "title": "Ram Fetch"
    },
    "parser": {
      "default": "pybgpkit",
      "description": "MRT files parser. Default `pybgpkit` is installed but slow, the others are system dependencies.",
      "enum": [
        "pybgpkit",
        "bgpkit",
        "pybgpstream",
        "bgpdump"
      ],
      "title": "Parser",
      "type": "string"
    },
    "remote_parse": {
      "anyOf": [
        {
          "type": "boolean"
        },
        {
          "type": "null"
        }
      ],
      "default": true,
      "description": "Parse directly from a remote source without saving archives locally or in RAM. Currently supported only by the pybgpstream parser.",
      "title": "Remote Parse"
    },
    "broker": {
      "anyOf": [
        {
          "enum": [
            "bgpkit",
            "bgpstream",
            "bgpfinder"
          ],
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": "bgpkit",
      "description": "BGP archive data broker",
      "title": "Broker"
    }
  },
  "required": [
    "collectors"
  ],
  "title": "BGPStreamConfig",
  "type": "object"
}

Fields:

Validators:

Source code in src/pybgpflux/bgpstreamconfig.py 
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
class BGPStreamConfig(BaseModel):
    """Unified BGPStream config"""

    # --- Query parameters (required) ---
    start_time: datetime.datetime | None = Field(
        default=None, description="Start of the stream"
    )
    end_time: datetime.datetime | None = Field(
        default=None, description="End of the stream"
    )
    collectors: list[str] = Field(description="List of collectors to get data from")
    data_types: list[Literal["ribs", "updates"]] = Field(
        default=["updates"],
        description="List of archives files to consider (`ribs` or `updates`)",
    )
    filters: FilterOptions | None = Field(default=None, description="Optional filters")

    # --- Implementation parameters (optional) ---
    max_concurrent_downloads: int | None = Field(
        default=10, description="Maximum concurrent downloads of archive files."
    )
    cache_dir: DirectoryPath | None = Field(
        default=None,
        description="Specifies the directory for caching downloaded files.",
    )
    ram_fetch: bool | None = Field(
        default=False,
        description=(
            "If caching is disabled, fetch temp files in shared RAM memory (/dev/shm) or normal disc temp dir (/tmp)."
            "Default (False) to reduce RAM usage."
        ),
    )
    parser: Literal["pybgpkit", "bgpkit", "pybgpstream", "bgpdump"] = Field(
        default="pybgpkit",
        description=(
            "MRT files parser. Default `pybgpkit` is installed but slow, the others are system dependencies."
        ),
    )
    remote_parse: bool | None = Field(
        default=True,
        description=(
            "Parse directly from a remote source without saving archives locally or in RAM. "
            "Currently supported only by the pybgpstream parser."
        ),
    )
    broker: Literal["bgpkit", "bgpstream", "bgpfinder"] | None = Field(
        default="bgpkit", description="BGP archive data broker"
    )

    @field_validator("start_time", "end_time", mode="before")
    @classmethod
    def normalize_to_utc(cls, dt: datetime.datetime) -> datetime.datetime:
        # if naive datetime (not timezone-aware) assume it's UTC
        if dt.tzinfo is None:
            return dt.replace(tzinfo=datetime.timezone.utc)
        # if timezone-aware, convert to utc
        else:
            return dt.astimezone(datetime.timezone.utc)

    @field_validator("parser")
    @classmethod
    def check_parser_available(cls, parser: str) -> str:
        if parser == "pybgpkit":
            if find_spec("bgpkit") is None:
                raise ValueError(
                    "pybgpkit is not installed. Install with: pip install pybgpkit"
                )

        elif parser == "pybgpstream":
            if find_spec("pybgpstream") is None:
                raise ValueError(
                    "pybgpstream is not installed. "
                    "Install with: pip install pybgpstream (ensure system dependencies are met)"
                )

        elif parser == "bgpdump":
            if shutil.which("bgpdump") is None:
                raise ValueError(
                    "bgpdump binary not found in PATH. "
                    "Install with: sudo apt-get install bgpdump"
                )

        elif parser == "bgpkit":
            if shutil.which("bgpkit-parser") is None:
                raise ValueError(
                    "bgpkit binary not found in PATH. "
                    "Install from: https://github.com/bgpkit/bgpkit-parser "
                    "or use cargo: cargo install bgpkit-parser --features cli"
                )

        return parser

    @model_validator(mode="after")
    def sanitize(self) -> "BGPStreamConfig":
        if (self.start_time is None) ^ (self.end_time is None):
            raise ValueError(
                "Provide both start and end times, or leave both as None for live mode."
            )

        if not self.is_live():
            assert self.start_time < self.end_time  # type: ignore[reportOperatorIssue]
        # Force data_type to update for live mode
        else:
            self.data_types = ["updates"]

        # Disable remote parsing when caching
        if self.cache_dir:
            self.remote_parse = False

        return self

    def is_live(self) -> bool:
        return self.start_time is None and self.end_time is None

start_time = None pydantic-field

Start of the stream

end_time = None pydantic-field

End of the stream

collectors pydantic-field

List of collectors to get data from

data_types = ['updates'] pydantic-field

List of archives files to consider (ribs or updates)

filters = None pydantic-field

Optional filters

max_concurrent_downloads = 10 pydantic-field

Maximum concurrent downloads of archive files.

cache_dir = None pydantic-field

Specifies the directory for caching downloaded files.

ram_fetch = False pydantic-field

If caching is disabled, fetch temp files in shared RAM memory (/dev/shm) or normal disc temp dir (/tmp).Default (False) to reduce RAM usage.

parser = 'pybgpkit' pydantic-field

MRT files parser. Default pybgpkit is installed but slow, the others are system dependencies.

remote_parse = True pydantic-field

Parse directly from a remote source without saving archives locally or in RAM. Currently supported only by the pybgpstream parser.

broker = 'bgpkit' pydantic-field

BGP archive data broker

FilterOptions

pybgpflux.bgpstreamconfig.FilterOptions pydantic-model

Bases: BaseModel

A unified model for the available filter options.

Show JSON schema: 
{
  "description": "A unified model for the available filter options.",
  "properties": {
    "origin_asn": {
      "anyOf": [
        {
          "type": "integer"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "Filter by the origin AS number.",
      "title": "Origin Asn"
    },
    "prefix": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "Filter by an exact prefix match.",
      "title": "Prefix"
    },
    "prefix_super": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "Filter by the exact prefix and its more general super-prefixes.",
      "title": "Prefix Super"
    },
    "prefix_sub": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "Filter by the exact prefix and its more specific sub-prefixes.",
      "title": "Prefix Sub"
    },
    "prefix_super_sub": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "Filter by the exact prefix and both its super- and sub-prefixes.",
      "title": "Prefix Super Sub"
    },
    "peer_ip": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "format": "ipv4",
          "type": "string"
        },
        {
          "format": "ipv6",
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "Filter by the IP address of a single BGP peer.",
      "title": "Peer Ip"
    },
    "peer_ips": {
      "anyOf": [
        {
          "items": {
            "anyOf": [
              {
                "type": "string"
              },
              {
                "format": "ipv4",
                "type": "string"
              },
              {
                "format": "ipv6",
                "type": "string"
              }
            ]
          },
          "type": "array"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "Filter by a list of BGP peer IP addresses.",
      "title": "Peer Ips"
    },
    "peer_asn": {
      "anyOf": [
        {
          "type": "integer"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "Filter by the AS number of the BGP peer.",
      "title": "Peer Asn"
    },
    "update_type": {
      "anyOf": [
        {
          "enum": [
            "withdraw",
            "announce"
          ],
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "Filter by the BGP update message type.",
      "title": "Update Type"
    },
    "as_path": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "Filter by a regular expression matching the AS path.",
      "title": "As Path"
    },
    "ip_version": {
      "anyOf": [
        {
          "enum": [
            4,
            6
          ],
          "type": "integer"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "Filter by ip version.",
      "title": "Ip Version"
    }
  },
  "title": "FilterOptions",
  "type": "object"
}

Fields:

Source code in src/pybgpflux/bgpstreamconfig.py 
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
class FilterOptions(BaseModel):
    """A unified model for the available filter options."""

    origin_asn: int | None = Field(
        default=None, description="Filter by the origin AS number."
    )
    prefix: str | None = Field(
        default=None, description="Filter by an exact prefix match."
    )
    prefix_super: str | None = Field(
        default=None,
        description="Filter by the exact prefix and its more general super-prefixes.",
    )
    prefix_sub: str | None = Field(
        default=None,
        description="Filter by the exact prefix and its more specific sub-prefixes.",
    )
    prefix_super_sub: str | None = Field(
        default=None,
        description="Filter by the exact prefix and both its super- and sub-prefixes.",
    )
    peer_ip: str | IPv4Address | IPv6Address | None = Field(
        default=None, description="Filter by the IP address of a single BGP peer."
    )
    peer_ips: list[str | IPv4Address | IPv6Address] | None = Field(
        default=None, description="Filter by a list of BGP peer IP addresses."
    )
    peer_asn: int | None = Field(
        default=None, description="Filter by the AS number of the BGP peer."
    )
    update_type: Literal["withdraw", "announce"] | None = Field(
        default=None, description="Filter by the BGP update message type."
    )
    as_path: str | None = Field(
        default=None, description="Filter by a regular expression matching the AS path."
    )
    ip_version: Literal[4, 6] | None = Field(
        default=None, description="Filter by ip version."
    )

origin_asn = None pydantic-field

Filter by the origin AS number.

prefix = None pydantic-field

Filter by an exact prefix match.

prefix_super = None pydantic-field

Filter by the exact prefix and its more general super-prefixes.

prefix_sub = None pydantic-field

Filter by the exact prefix and its more specific sub-prefixes.

prefix_super_sub = None pydantic-field

Filter by the exact prefix and both its super- and sub-prefixes.

peer_ip = None pydantic-field

Filter by the IP address of a single BGP peer.

peer_ips = None pydantic-field

Filter by a list of BGP peer IP addresses.

peer_asn = None pydantic-field

Filter by the AS number of the BGP peer.

update_type = None pydantic-field

Filter by the BGP update message type.

as_path = None pydantic-field

Filter by a regular expression matching the AS path.

ip_version = None pydantic-field

Filter by ip version.

LiveStreamConfig

pybgpflux.bgpstreamconfig.LiveStreamConfig pydantic-model

Bases: BaseModel

Config for live mode

Show JSON schema: 
{
  "$defs": {
    "FilterOptions": {
      "description": "A unified model for the available filter options.",
      "properties": {
        "origin_asn": {
          "anyOf": [
            {
              "type": "integer"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "description": "Filter by the origin AS number.",
          "title": "Origin Asn"
        },
        "prefix": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "description": "Filter by an exact prefix match.",
          "title": "Prefix"
        },
        "prefix_super": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "description": "Filter by the exact prefix and its more general super-prefixes.",
          "title": "Prefix Super"
        },
        "prefix_sub": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "description": "Filter by the exact prefix and its more specific sub-prefixes.",
          "title": "Prefix Sub"
        },
        "prefix_super_sub": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "description": "Filter by the exact prefix and both its super- and sub-prefixes.",
          "title": "Prefix Super Sub"
        },
        "peer_ip": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "format": "ipv4",
              "type": "string"
            },
            {
              "format": "ipv6",
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "description": "Filter by the IP address of a single BGP peer.",
          "title": "Peer Ip"
        },
        "peer_ips": {
          "anyOf": [
            {
              "items": {
                "anyOf": [
                  {
                    "type": "string"
                  },
                  {
                    "format": "ipv4",
                    "type": "string"
                  },
                  {
                    "format": "ipv6",
                    "type": "string"
                  }
                ]
              },
              "type": "array"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "description": "Filter by a list of BGP peer IP addresses.",
          "title": "Peer Ips"
        },
        "peer_asn": {
          "anyOf": [
            {
              "type": "integer"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "description": "Filter by the AS number of the BGP peer.",
          "title": "Peer Asn"
        },
        "update_type": {
          "anyOf": [
            {
              "enum": [
                "withdraw",
                "announce"
              ],
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "description": "Filter by the BGP update message type.",
          "title": "Update Type"
        },
        "as_path": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "description": "Filter by a regular expression matching the AS path.",
          "title": "As Path"
        },
        "ip_version": {
          "anyOf": [
            {
              "enum": [
                4,
                6
              ],
              "type": "integer"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "description": "Filter by ip version.",
          "title": "Ip Version"
        }
      },
      "title": "FilterOptions",
      "type": "object"
    }
  },
  "description": "Config for live mode",
  "properties": {
    "collectors": {
      "description": "List of collectors to get data from (for now only RIS live collectors)",
      "items": {
        "type": "string"
      },
      "title": "Collectors",
      "type": "array"
    },
    "filters": {
      "anyOf": [
        {
          "$ref": "#/$defs/FilterOptions"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "Optional filters"
    },
    "jitter_buffer_delay": {
      "anyOf": [
        {
          "type": "number"
        },
        {
          "type": "null"
        }
      ],
      "default": 10.0,
      "description": "Jitter buffer time in seconds to make sure RIS live updates are time-sorted. Introduce a slight delay. Set to None or 0 to disable",
      "title": "Jitter Buffer Delay"
    }
  },
  "required": [
    "collectors"
  ],
  "title": "LiveStreamConfig",
  "type": "object"
}

Fields:

Source code in src/pybgpflux/bgpstreamconfig.py 
165
166
167
168
169
170
171
172
173
174
175
class LiveStreamConfig(BaseModel):
    """Config for live mode"""

    collectors: list[str] = Field(
        description="List of collectors to get data from (for now only RIS live collectors)"
    )
    filters: FilterOptions | None = Field(default=None, description="Optional filters")
    jitter_buffer_delay: float | None = Field(
        default=10.0,
        description="Jitter buffer time in seconds to make sure RIS live updates are time-sorted. Introduce a slight delay. Set to None or 0 to disable",
    )

collectors pydantic-field

List of collectors to get data from (for now only RIS live collectors)

filters = None pydantic-field

Optional filters

jitter_buffer_delay = 10.0 pydantic-field

Jitter buffer time in seconds to make sure RIS live updates are time-sorted. Introduce a slight delay. Set to None or 0 to disable