How to use the peeringdb.models.Network 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 / tests.py View on Github external
asn = 15169

        # Must not exist
        self.assertIsNone(api.get_autonomous_system(64500))

        # Using an API call (no cached data)
        autonomous_system = api.get_autonomous_system(asn)
        self.assertEqual(autonomous_system.asn, asn)

        # Save the data inside the cache
        details = {
            "id": autonomous_system.id,
            "asn": autonomous_system.asn,
            "name": autonomous_system.name,
        }
        network = Network(**details)
        network.save()

        # Using no API calls (cached data)
        autonomous_system = api.get_autonomous_system(asn)
        self.assertEqual(autonomous_system.asn, asn)
github respawner / peering-manager / peeringdb / api / views.py View on Github external
detail=False, methods=["post", "put", "patch"], url_path="index-peer-records"
    )
    def index_peer_records(self, request):
        return Response(
            {"peer-record-count": PeeringDB().force_peer_records_discovery()}
        )


class ContactsViewSet(ReadOnlyModelViewSet):
    queryset = Contact.objects.all()
    serializer_class = ContactSerializer
    filterset_class = ContactFilterSet


class NetworksViewSet(ReadOnlyModelViewSet):
    queryset = Network.objects.all()
    serializer_class = NetworkSerializer
    filterset_class = NetworkFilterSet


class PeerRecordViewSet(ModelViewSet):
    queryset = PeerRecord.objects.all()
    serializer_class = PeerRecordSerializer
    filterset_class = PeerRecordFilterSet


class SynchronizationViewSet(ReadOnlyModelViewSet):
    queryset = Synchronization.objects.all()
    serializer_class = SynchronizationSerializer
    filterset_class = SynchronizationFilterSet
github respawner / peering-manager / peeringdb / http.py View on Github external
def get_autonomous_system(self, asn):
        """
        Return an AS (and its details) given its ASN. The result can come from
        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 / admin.py View on Github external
from django.contrib import admin
from .models import Network, NetworkIXLAN, PeerRecord, Prefix, Synchronization

admin.site.register(Network)
admin.site.register(NetworkIXLAN)
admin.site.register(PeerRecord)
admin.site.register(Prefix)
admin.site.register(Synchronization)
github respawner / peering-manager / peering / models.py View on Github external
def get_peeringdb_network(self):
        try:
            return Network.objects.get(asn=self.asn)
        except Network.DoesNotExist:
            return None
github respawner / peering-manager / peeringdb / http.py View on Github external
def update_local_database(self, last_sync):
        """
        Update the local database by synchronizing all PeeringDB API's
        namespaces that we are actually caring about.
        """
        # Set time of sync
        time_of_sync = timezone.now()
        objects_to_sync = [
            (NAMESPACES["network_contact"], Contact),
            (NAMESPACES["network"], Network),
            (NAMESPACES["network_internet_exchange_lan"], NetworkIXLAN),
            (NAMESPACES["internet_exchange_prefix"], Prefix),
        ]
        list_of_changes = []

        # Make a single transaction, avoid too much database commits (poor
        # speed) and fail the whole synchronization if something goes wrong
        with transaction.atomic():
            # Try to sync objects
            for (namespace, object_type) in objects_to_sync:
                changes = self.synchronize_objects(last_sync, namespace, object_type)
                list_of_changes.append(changes)

        objects_changes = {
            "added": sum(added for added, _, _ in list_of_changes),
            "updated": sum(updated for _, updated, _ in list_of_changes),