Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
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)
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
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
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)
def get_peeringdb_network(self):
try:
return Network.objects.get(asn=self.asn)
except Network.DoesNotExist:
return None
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),