diff --git a/api/api.py b/api/api.py index a0788b6..c7514c4 100644 --- a/api/api.py +++ b/api/api.py @@ -2,7 +2,7 @@ # -*- coding: utf-8 -*- from typing import TypeVar -from datetime import datetime, date +import datetime from enum import Enum from dateutil.parser import parse @@ -12,7 +12,7 @@ from redis import StrictRedis from bgpranking.libs.helpers import get_socket_path from bgpranking.libs.exceptions import InvalidDateFormat -Dates = TypeVar('Dates', datetime, date, str) +Dates = TypeVar('Dates', datetime.datetime, datetime.date, str) class IPVersion(Enum): @@ -33,9 +33,9 @@ class BGPRanking(): self.logger.setLevel(loglevel) def __normalize_date(self, date: Dates): - if isinstance(date, datetime): + if isinstance(date, datetime.datetime): return date.date().isoformat() - elif isinstance(date, date): + elif isinstance(date, datetime.date): return date.isoformat() elif isinstance(date, str): try: @@ -43,32 +43,32 @@ class BGPRanking(): except ValueError: 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): '''Aggregated ranking of all the ASNs known in the system, weighted by source.''' d = self.__normalize_date(date) key = f'{d}|asns|{ipversion.value}' 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.''' d = self.__normalize_date(date) key = f'{d}|{asn}|{ipversion.value}' 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.''' d = self.__normalize_date(date) key = f'{d}|asns|{ipversion.value}' 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.''' d = self.__normalize_date(date) key = f'{d}|{source}|{asn}|rank{ipversion.value}' 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): '''Get the rank of all the prefixes announced by an ASN, not weighted by source.''' d = self.__normalize_date(date)