Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
# Must not exist
self.assertIsNone(api.get_ix_network(0))
# Using an API call (no cached data)
ix_network = api.get_ix_network(ix_network_id)
self.assertEqual(ix_network.id, ix_network_id)
# Save the data inside the cache
details = {
"id": ix_network.id,
"asn": ix_network.asn,
"name": ix_network.name,
"ix_id": ix_network.ix_id,
"ixlan_id": ix_network.ixlan_id,
}
network_ixlan = NetworkIXLAN(**details)
network_ixlan.save()
# Using no API calls (cached data)
ix_network = api.get_ix_network(ix_network_id)
self.assertEqual(ix_network.id, ix_network_id)
ix_network = api.get_ix_network_by_ip_address(
ipv6_address=ipv6_address, ipv4_address=ipv4_address
)
self.assertEqual(ix_network.id, ix_network_id)
# Save the data inside the cache
details = {
"id": ix_network.id,
"asn": ix_network.asn,
"name": ix_network.name,
"ipaddr6": ipv6_address,
"ipaddr4": ipv4_address,
"ix_id": ix_network.ix_id,
"ixlan_id": ix_network.ixlan_id,
}
network_ixlan = NetworkIXLAN(**details)
network_ixlan.save()
# Using no API calls (cached data)
ix_network = api.get_ix_network_by_ip_address(ipv6_address=ipv6_address)
self.assertEqual(ix_network.id, ix_network_id)
ix_network = api.get_ix_network_by_ip_address(ipv4_address=ipv4_address)
self.assertEqual(ix_network.id, ix_network_id)
ix_network = api.get_ix_network_by_ip_address(
ipv6_address=ipv6_address, ipv4_address=ipv4_address
)
self.assertEqual(ix_network.id, ix_network_id)
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),
"deleted": sum(deleted for _, _, deleted in list_of_changes),
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_ix_network(self, ix_network_id):
"""
Return an IX network (and its details) given an IP address. The result
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