D4
/
BGP-Ranking
mirror of https://github.com/D4-project/BGP-Ranking
2
0
Fork 0
Code Issues Releases Wiki Activity

fix: typing

pull/12/head
Raphaël Vinot 2018-04-10 23:22:32 +02:00
parent 0764f066fd
commit 07f29e865e
1 changed files with 9 additions and 9 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

18
api/api.py
View File

@ -2,7 +2,7 @@
# -*- coding: utf-8 -*- # -*- coding: utf-8 -*-
from typing import TypeVar from typing import TypeVar
from datetime import datetime, date import datetime
from enum import Enum from enum import Enum
from dateutil.parser import parse from dateutil.parser import parse
@ -12,7 +12,7 @@ from redis import StrictRedis
from bgpranking.libs.helpers import get_socket_path from bgpranking.libs.helpers import get_socket_path
from bgpranking.libs.exceptions import InvalidDateFormat from bgpranking.libs.exceptions import InvalidDateFormat
Dates = TypeVar('Dates', datetime, date, str) Dates = TypeVar('Dates', datetime.datetime, datetime.date, str)
class IPVersion(Enum): class IPVersion(Enum):
@ -33,9 +33,9 @@ class BGPRanking():
self.logger.setLevel(loglevel) self.logger.setLevel(loglevel)
def __normalize_date(self, date: Dates): def __normalize_date(self, date: Dates):
if isinstance(date, datetime): if isinstance(date, datetime.datetime):
return date.date().isoformat() return date.date().isoformat()
elif isinstance(date, date): elif isinstance(date, datetime.date):
return date.isoformat() return date.isoformat()
elif isinstance(date, str): elif isinstance(date, str):
try: try:
@ -43,32 +43,32 @@ class BGPRanking():
except ValueError: except ValueError:
raise InvalidDateFormat('Unable to parse the date. Should be YYYY-MM-DD.') raise InvalidDateFormat('Unable to parse the date. Should be YYYY-MM-DD.')
def asns_global_ranking(self, date: Dates= date.today(), def asns_global_ranking(self, date: Dates=datetime.date.today(),
ipversion: IPVersion=IPVersion.v4, limit: int=100): ipversion: IPVersion=IPVersion.v4, limit: int=100):
'''Aggregated ranking of all the ASNs known in the system, weighted by source.''' '''Aggregated ranking of all the ASNs known in the system, weighted by source.'''
d = self.__normalize_date(date) d = self.__normalize_date(date)
key = f'{d}|asns|{ipversion.value}' key = f'{d}|asns|{ipversion.value}'
return self.ranking.zrevrange(key, start=0, end=limit, withscores=True) return self.ranking.zrevrange(key, start=0, end=limit, withscores=True)
def asn_details(self, asn: int, date: Dates= date.today(), ipversion: IPVersion=IPVersion.v4): def asn_details(self, asn: int, date: Dates= datetime.date.today(), ipversion: IPVersion=IPVersion.v4):
'''Aggregated ranking of all the prefixes anounced by the given ASN, weighted by source.''' '''Aggregated ranking of all the prefixes anounced by the given ASN, weighted by source.'''
d = self.__normalize_date(date) d = self.__normalize_date(date)
key = f'{d}|{asn}|{ipversion.value}' key = f'{d}|{asn}|{ipversion.value}'
return self.ranking.zrevrange(key, start=0, end=-1, withscores=True) return self.ranking.zrevrange(key, start=0, end=-1, withscores=True)
def asn_rank(self, asn: int, date: Dates= date.today(), ipversion: IPVersion=IPVersion.v4): def asn_rank(self, asn: int, date: Dates= datetime.date.today(), ipversion: IPVersion=IPVersion.v4):
'''Get the rank of a single ASN, weighted by source.''' '''Get the rank of a single ASN, weighted by source.'''
d = self.__normalize_date(date) d = self.__normalize_date(date)
key = f'{d}|asns|{ipversion.value}' key = f'{d}|asns|{ipversion.value}'
return self.ranking.zrank(key, asn) return self.ranking.zrank(key, asn)
def asn_rank_by_source(self, asn: int, source: str, date: Dates= date.today(), ipversion: IPVersion=IPVersion.v4): def asn_rank_by_source(self, asn: int, source: str, date: Dates= datetime.date.today(), ipversion: IPVersion=IPVersion.v4):
'''Get the rank of a single ASN, not weighted by source.''' '''Get the rank of a single ASN, not weighted by source.'''
d = self.__normalize_date(date) d = self.__normalize_date(date)
key = f'{d}|{source}|{asn}|rank{ipversion.value}' key = f'{d}|{source}|{asn}|rank{ipversion.value}'
return self.ranking.get(key) return self.ranking.get(key)
def asn_details_by_source(self, source: str, asn: int, date: Dates= date.today(), def asn_details_by_source(self, source: str, asn: int, date: Dates= datetime.date.today(),
ipversion: IPVersion=IPVersion.v4): ipversion: IPVersion=IPVersion.v4):
'''Get the rank of all the prefixes announced by an ASN, not weighted by source.''' '''Get the rank of all the prefixes announced by an ASN, not weighted by source.'''
d = self.__normalize_date(date) d = self.__normalize_date(date)