Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
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
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
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
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
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
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
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,
)