Skip to content

BGPElement

pybgpkitstream.bgpelement.BGPElement

Bases: NamedTuple

Compatible with pybgpstream.BGPElem

Source code in src/pybgpkitstream/bgpelement.py 
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
48
49
50
51
52
53
class BGPElement(NamedTuple):
    """Compatible with pybgpstream.BGPElem"""

    time: float  # time first for sorting tuples convention
    type: Literal["R", "A", "W"]
    collector: str
    peer_asn: int
    peer_address: str
    fields: ElementFields

    def __str__(self):
        """Credit to pybgpstream"""
        return "%s|%f|%s|%s|%s|%s|%s|%s|%s|%s|%s" % (
            self.type,
            self.time,
            self.collector,
            self.peer_asn,
            self.peer_address,
            self._maybe_field("prefix"),
            self._maybe_field("next-hop"),
            self.fields["as-path"]
            if self.fields.get('as-path')
            else None,
            " ".join(self.fields["communities"])
            if self.fields.get('communities')
            else None,
            self._maybe_field("old-state"),
            self._maybe_field("new-state"),
        )

    def _maybe_field(self, field):
        """Credit to pybgpstream"""
        return self.fields[field] if field in self.fields else None

    # Useful for sorting streams
    def __lt__(self, other):
        return self.time < other.time

    def __le__(self, other):
        return self.time <= other.time

__str__()

Credit to pybgpstream

Source code in src/pybgpkitstream/bgpelement.py 
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
def __str__(self):
    """Credit to pybgpstream"""
    return "%s|%f|%s|%s|%s|%s|%s|%s|%s|%s|%s" % (
        self.type,
        self.time,
        self.collector,
        self.peer_asn,
        self.peer_address,
        self._maybe_field("prefix"),
        self._maybe_field("next-hop"),
        self.fields["as-path"]
        if self.fields.get('as-path')
        else None,
        " ".join(self.fields["communities"])
        if self.fields.get('communities')
        else None,
        self._maybe_field("old-state"),
        self._maybe_field("new-state"),
    )

Using BGPElement

Each BGP message in a stream is represented as a BGPElement namedtuple containing:

  • time (float): Unix timestamp of the message
  • type (Literal["R", "A", "W"]): Element type ("R" = RIB (routing table snapshot), "A" = Announce (new/updated prefix), "W" = Withdraw (removed prefix))
  • collector (str): Source collector name
  • peer_asn (int): BGP peer AS number
  • peer_address (str): BGP peer IP address
  • fields (ElementFields): Dictionary containing:
  • "prefix" (str): IPv4 or IPv6 prefix
  • "next-hop" (str): Next-hop IP address
  • "as-path" (str): AS path (space-separated AS numbers)
  • "communities" (list[str]): BGP communities

Example Usage

for elem in stream:
    print(f"Type: {elem.type}")
    print(f"Time: {elem.time}")
    print(f"Collector: {elem.collector}")
    print(f"Peer: {elem.peer_asn} ({elem.peer_address})")
    print(f"Prefix: {elem.fields['prefix']}")
    print(f"AS Path: {elem.fields.get('as-path')}")
    print(f"Communities: {elem.fields.get('communities', [])}")
    print("---")

String Representation

BGPElement has a compatible string format with PyBGPStream:

print(elem)
# Output: A|1283289600.000000|route-views.wide|2497|192.0.2.1|192.0.2.0/24|192.0.2.254|64512 64513|||