How to use the peeringdb.http.Object function in peeringdb

To help you get started, we’ve selected a few peeringdb examples, based on popular ways it is used in public projects.

Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.

github respawner / peering-manager / peeringdb / http.py View on Github external
the local database (cache built with the peeringdb_sync command). If
        the AS details are not found in the local database, they will be
        fetched online which will take more time.
        """
        try:
            # Try to get from cached data
            network = Network.objects.get(asn=asn)
        except Network.DoesNotExist:
            # If no cached data found, query the API
            search = {"asn": asn}
            result = self.lookup(NAMESPACES["network"], search)

            if not result or not result["data"]:
                return None

            network = Object(result["data"][0])

        return network
github respawner / peering-manager / peeringdb / http.py View on Github external
if network_ixlan:
            # Try to get prefixes from cache
            ix_prefixes = Prefix.objects.filter(ixlan_id=network_ixlan.ixlan_id)

            # If not cached data, try to fetch online
            if not ix_prefixes:
                search = {"ixlan_id": network_ixlan.ixlan_id}
                result = self.lookup(NAMESPACES["internet_exchange_prefix"], search)

                if not result or not result["data"]:
                    return prefixes

                ix_prefixes = []
                for ix_prefix in result["data"]:
                    ix_prefixes.append(Object(ix_prefix))

            # Build a list of prefixes
            for ix_prefix in ix_prefixes:
                prefixes.append(ipaddress.ip_network(ix_prefix.prefix))

        return prefixes
github respawner / peering-manager / peeringdb / http.py View on Github external
can come from the local database (cache built with the peeringdb_sync
        command). If the IX network is not found in the local database, it will
        be fetched online which will take more time.
        """
        try:
            # Try to get from cached data
            network_ixlan = NetworkIXLAN.objects.get(id=ix_network_id)
        except NetworkIXLAN.DoesNotExist:
            # If no cached data found, query the API
            search = {"id": ix_network_id}
            result = self.lookup(NAMESPACES["network_internet_exchange_lan"], search)

            if not result or not result["data"]:
                return None

            network_ixlan = Object(result["data"][0])

        return network_ixlan
github respawner / peering-manager / peeringdb / http.py View on Github external
Returns a list of all IX networks an AS is connected to.
        """
        # Try to get from cached data
        network_ixlans = NetworkIXLAN.objects.filter(asn=asn)

        # If nothing found in cache, try to fetch data online
        if not network_ixlans:
            search = {"asn": asn}
            result = self.lookup(NAMESPACES["network_internet_exchange_lan"], search)

            if not result or not result["data"]:
                return None

            network_ixlans = []
            for ix_network in result["data"]:
                network_ixlans.append(Object(ix_network))

        return network_ixlans
github respawner / peering-manager / peeringdb / http.py View on Github external
if ipv6_address:
            search.update({"ipaddr6": ipv6_address})
        if ipv4_address:
            search.update({"ipaddr4": ipv4_address})

        try:
            # Try to get from cached data
            network_ixlan = NetworkIXLAN.objects.get(**search)
        except NetworkIXLAN.DoesNotExist:
            # If no cached data found, query the API
            result = self.lookup(NAMESPACES["network_internet_exchange_lan"], search)

            if not result or not result["data"]:
                return None

            network_ixlan = Object(result["data"][0])

        return network_ixlan
github respawner / peering-manager / peeringdb / http.py View on Github external
calls to be made.
        """
        # Try to get from cached data
        network_ixlans = NetworkIXLAN.objects.filter(ix_id=ix_id)

        # If nothing found in cache, try to fetch data online
        if not network_ixlans:
            search = {"ix_id": ix_id}
            result = self.lookup(NAMESPACES["network_internet_exchange_lan"], search)

            if not result or not result["data"]:
                return None

            network_ixlans = []
            for data in result["data"]:
                network_ixlans.append(Object(data))

        # List potential peers
        peers = []
        for network_ixlan in network_ixlans:
            # Ignore our own ASN
            if network_ixlan.asn == settings.MY_ASN:
                continue

            # Get more details about the current network
            network = self.get_autonomous_system(network_ixlan.asn)

            # Package all gathered details
            peers.append({"network": network, "network_ixlan": network_ixlan})

        return peers
github respawner / peering-manager / peeringdb / http.py View on Github external
This function returns the number of objects that have been successfully
        synchronized to the local database.
        """
        objects_added = 0
        objects_updated = 0
        objects_deleted = 0

        # Get all network changes since the last sync
        search = {"since": last_sync, "depth": 0}
        result = self.lookup(namespace, search)

        if not result:
            return None

        for data in result["data"]:
            peeringdb_object = Object(data)
            marked_as_deleted = peeringdb_object.status == "deleted"
            marked_as_new = False

            try:
                # Get the local object by its ID
                local_object = model.objects.get(pk=peeringdb_object.id)

                # Object marked as deleted so remove it locally too
                if marked_as_deleted:
                    local_object.delete()
                    objects_deleted += 1
                    self.logger.debug(
                        "deleted %s #%s from local database",
                        model._meta.verbose_name.lower(),
                        peeringdb_object.id,
                    )