diff --git a/controllers/hegemony_controller.py b/controllers/hegemony_controller.py index 8e95ee6..67e2963 100644 --- a/controllers/hegemony_controller.py +++ b/controllers/hegemony_controller.py @@ -1,4 +1,7 @@ -from fastapi import APIRouter, Depends, Query, Request, Response, HTTPException +from dtos.hegemony_country_dto import HegemonyCountryDTO +from dtos.hegemony_dto import HegemonyDTO +from dtos.hegemony_prefix_dto import HegemonyPrefixDTO +from fastapi import APIRouter, Depends, Query, Request, Response, HTTPException, status from datetime import datetime, timedelta from sqlalchemy.orm import Session from services.hegemony_service import HegemonyService @@ -16,6 +19,81 @@ class HegemonyController: service = HegemonyService() + @staticmethod + @router.get("/", response_model=GenericResponseDTO[HegemonyDTO]) + async def get_hegemony( + request: Request, + db: Session = Depends(get_db), + timebin: Optional[datetime] = Query( + None, description="Timestamp of reported value."), + timebin__gte: Optional[datetime] = Query( + None, description="Timestamp of reported value."), + timebin__lte: Optional[datetime] = Query( + None, description="Timestamp of reported value."), + asn: Optional[str] = Query( + None, description="Dependency. Transit network commonly seen in BGP paths towards originasn. Can be a single value or a list of comma separated values."), + originasn: Optional[str] = Query( + None, description="Dependent network, it can be any public ASN. Can be a single value or a list of comma separated values. Retrieve all dependencies of a network by setting a single value and a timebin."), + af: Optional[int] = Query( + None, description="Address Family (IP version), values are either 4 or 6."), + hege: Optional[float] = Query( + None, description="AS Hegemony is the estimated fraction of paths towards the originasn. The values range between 0 and 1, low values represent a small number of path (low dependency) and values close to 1 represent strong dependencies."), + hege__gte: Optional[float] = Query( + None, description="AS Hegemony is the estimated fraction of paths towards the originasn. The values range between 0 and 1, low values represent a small number of path (low dependency) and values close to 1 represent strong dependencies."), + hege__lte: Optional[float] = Query( + None, description="AS Hegemony is the estimated fraction of paths towards the originasn. The values range between 0 and 1, low values represent a small number of path (low dependency) and values close to 1 represent strong dependencies."), + page: Optional[int] = Query( + 1, ge=1, description="A page number within the paginated result set"), + ordering: Optional[str] = Query( + None, description="Which field to use when ordering the results") + ) -> GenericResponseDTO[HegemonyDTO]: + """ + List AS dependencies for all ASes visible in monitored BGP data. This endpoint also provides the AS dependency to the entire IP space (a.k.a. global graph) which is available by setting the originasn parameter to 0. + + """ + timebin__gte, timebin__lte = validate_timebin_params( + timebin, timebin__gte, timebin__lte) + + # Convert comma-separated ASNs to lists + asn_list = [int(x.strip()) for x in asn.split(",")] if asn else None + originasn_list = [int(x.strip()) + for x in originasn.split(",")] if originasn else None + + # Ensure either asn or originasn is provided + if not asn and not originasn: + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail="Required parameter missing. Please provide one of the following parameters: ['originasn', 'asn']" + ) + + hegemony_data, total_count = HegemonyController.service.get_hegemony( + db, + timebin_gte=timebin__gte, + timebin_lte=timebin__lte, + asn_ids=asn_list, + originasn_ids=originasn_list, + af=af, + hege=hege, + hege_gte=hege__gte, + hege_lte=hege__lte, + page=page, + order_by=ordering + ) + + # Calculate pagination + next_page = page + 1 if (page * page_size) < total_count else None + prev_page = page - 1 if page > 1 else None + + return GenericResponseDTO( + count=total_count, + next=build_url(request, next_page), + previous=build_url(request, prev_page), + results=hegemony_data + ) + @staticmethod @router.get("/cones", response_model=GenericResponseDTO[HegemonyConeDTO]) async def get_hegemony_cones( @@ -135,3 +213,185 @@ async def get_hegemony_alarms( previous=build_url(request, prev_page), results=alarms ) + + @staticmethod + @router.get("/countries", response_model=GenericResponseDTO[HegemonyCountryDTO]) + async def get_hegemony_countries( + request: Request, + db: Session = Depends(get_db), + timebin: Optional[datetime] = Query( + None, description="Timestamp of reported value."), + timebin__gte: Optional[datetime] = Query( + None, description="Timestamp of reported value."), + timebin__lte: Optional[datetime] = Query( + None, description="Timestamp of reported value."), + asn: Optional[str] = Query( + None, description="Dependency. Network commonly seen in BGP paths towards monitored country. Can be a single value or a list of comma separated values."), + country: Optional[str] = Query( + None, description="Monitored country or region (e.g. EU and AP) as defined by its set of ASes registered in registeries delegated files. Can be a single value or a list of comma separated values. Retrieve all dependencies of a country by setting a single value and a timebin."), + af: Optional[int] = Query( + None, description="Address Family (IP version), values are either 4 or 6."), + weightscheme: Optional[str] = Query( + None, description="Scheme used to aggregate AS Hegemony scores. 'as' gives equal weight to each AS, 'eyeball' put emphasis on large eyeball networks."), + transitonly: Optional[bool] = Query( + None, description="True means that the last AS (origin AS) in BGP paths is ignored, thus focusing only on transit ASes."), + hege: Optional[float] = Query( + None, description="AS Hegemony is the estimated fraction of paths towards the monitored country. The values range between 0 and 1, low values represent a small number of path (low dependency) and values close to 1 represent strong dependencies."), + hege__gte: Optional[float] = Query( + None, description="AS Hegemony is the estimated fraction of paths towards the monitored country. The values range between 0 and 1, low values represent a small number of path (low dependency) and values close to 1 represent strong dependencies."), + hege__lte: Optional[float] = Query( + None, description="AS Hegemony is the estimated fraction of paths towards the monitored country. The values range between 0 and 1, low values represent a small number of path (low dependency) and values close to 1 represent strong dependencies."), + page: Optional[int] = Query( + 1, ge=1, description="A page number within the paginated result set"), + ordering: Optional[str] = Query( + None, description="Which field to use when ordering the results") + ) -> GenericResponseDTO[HegemonyCountryDTO]: + """ + List AS dependencies of countries. A country infrastructure is defined by its ASes registed in RIRs delegated files. Emphasis can be put on eyeball users with the eyeball weighting scheme (i.e. weightscheme='eyeball'). + + """ + timebin__gte, timebin__lte = validate_timebin_params( + timebin, timebin__gte, timebin__lte, max_days=31) + + # Ensure either `asn` or `country` is provided + if not asn and not country: + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail="Required parameter missing. Please provide one of the following parameters: ['country', 'asn']" + ) + + # Convert comma-separated values to lists + asn_list = [int(x.strip()) for x in asn.split(",")] if asn else None + country_list = [x.strip() + for x in country.split(",")] if country else None + + countries, total_count = HegemonyController.service.get_hegemony_countries( + db, + timebin_gte=timebin__gte, + timebin_lte=timebin__lte, + asn_ids=asn_list, + countries=country_list, + af=af, + weightscheme=weightscheme, + transitonly=transitonly, + hege=hege, + hege_gte=hege__gte, + hege_lte=hege__lte, + page=page, + order_by=ordering + ) + + # Calculate pagination + next_page = page + 1 if (page * page_size) < total_count else None + prev_page = page - 1 if page > 1 else None + + return GenericResponseDTO( + count=total_count, + next=build_url(request, next_page), + previous=build_url(request, prev_page), + results=countries + ) + + @staticmethod + @router.get("/prefixes", response_model=GenericResponseDTO[HegemonyPrefixDTO]) + async def get_hegemony_prefixes( + request: Request, + db: Session = Depends(get_db), + timebin: Optional[datetime] = Query( + None, description="Timestamp of reported value."), + timebin__gte: Optional[datetime] = Query( + None, description="Timestamp of reported value."), + timebin__lte: Optional[datetime] = Query( + None, description="Timestamp of reported value."), + prefix: Optional[str] = Query( + None, description="Monitored prefix, it can be any globally reachable prefix. Can be a single value or a list of comma separated values."), + asn: Optional[str] = Query( + None, description="Dependency. Network commonly seen in BGP paths towards monitored prefix. Can be a single value or a list of comma separated values."), + originasn: Optional[str] = Query( + None, description="Origin network, it can be any public ASN. Can be a single value or a list of comma separated values."), + country: Optional[str] = Query( + None, description="Country code for prefixes as reported by Maxmind's Geolite2 geolocation database. Can be a single value or a list of comma separated values. Retrieve all dependencies of a country by setting a single value and a timebin."), + rpki_status: Optional[str] = Query( + None, description="Route origin validation state for the monitored prefix and origin AS using RPKI."), + irr_status: Optional[str] = Query( + None, description="Route origin validation state for the monitored prefix and origin AS using IRR."), + delegated_prefix_status: Optional[str] = Query( + None, description="Status of the monitored prefix in the RIR's delegated stats. Status other than 'assigned' are usually considered as bogons."), + delegated_asn_status: Optional[str] = Query( + None, description="Status of the origin ASN in the RIR's delegated stats. Status other than 'assigned' are usually considered as bogons."), + af: Optional[int] = Query( + None, description="Address Family (IP version), values are either 4 or 6."), + hege: Optional[float] = Query( + None, description="AS Hegemony is the estimated fraction of paths towards the monitored prefix. The values range between 0 and 1, low values represent a small number of path (low dependency) and values close to 1 represent strong dependencies."), + hege__gte: Optional[float] = Query( + None, description="AS Hegemony is the estimated fraction of paths towards the monitored prefix. The values range between 0 and 1, low values represent a small number of path (low dependency) and values close to 1 represent strong dependencies."), + hege__lte: Optional[float] = Query( + None, description="AS Hegemony is the estimated fraction of paths towards the monitored prefix. The values range between 0 and 1, low values represent a small number of path (low dependency) and values close to 1 represent strong dependencies."), + origin_only: Optional[bool] = Query( + None, description="Filter out dependency results and provide only prefix/origin ASN results"), + page: Optional[int] = Query( + 1, ge=1, description="A page number within the paginated result set"), + ordering: Optional[str] = Query( + None, description="Which field to use when ordering the results") + ) -> GenericResponseDTO[HegemonyPrefixDTO]: + """ + List AS dependencies of prefixes. + + """ + # Ensure at least one filter is provided + if not any([prefix, originasn, country, rpki_status, irr_status, + delegated_prefix_status, delegated_asn_status]): + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail="Required parameter missing. Please provide one of the following parameter: ['prefix', 'originasn', 'country', 'rpki_status', 'irr_status', 'delegated_prefix_status', 'delegated_asn_status']" + ) + + timebin__gte, timebin__lte = validate_timebin_params( + timebin, timebin__gte, timebin__lte, max_days=3) + + # Convert comma-separated values to lists + prefix_list = [x.strip() + for x in prefix.split(",")] if prefix else None + asn_list = [int(x.strip()) for x in asn.split(",")] if asn else None + originasn_list = [int(x.strip()) + for x in originasn.split(",")] if originasn else None + country_list = [x.strip() + for x in country.split(",")] if country else None + + prefixes, total_count = HegemonyController.service.get_hegemony_prefixes( + db, + timebin_gte=timebin__gte, + timebin_lte=timebin__lte, + prefixes=prefix_list, + asn_ids=asn_list, + originasn_ids=originasn_list, + countries=country_list, + rpki_status=rpki_status, + irr_status=irr_status, + delegated_prefix_status=delegated_prefix_status, + delegated_asn_status=delegated_asn_status, + af=af, + hege=hege, + hege_gte=hege__gte, + hege_lte=hege__lte, + origin_only=origin_only, + page=page, + order_by=ordering + ) + + # Calculate pagination + next_page = page + 1 if (page * page_size) < total_count else None + prev_page = page - 1 if page > 1 else None + + return GenericResponseDTO( + count=total_count, + next=build_url(request, next_page), + previous=build_url(request, prev_page), + results=prefixes + ) diff --git a/dtos/hegemony_country_dto.py b/dtos/hegemony_country_dto.py new file mode 100644 index 0000000..43d6717 --- /dev/null +++ b/dtos/hegemony_country_dto.py @@ -0,0 +1,17 @@ +from pydantic import BaseModel +from datetime import datetime + + +class HegemonyCountryDTO(BaseModel): + timebin: datetime + country: str + asn: int + hege: float + af: int + asn_name: str + weight: float + weightscheme: str + transitonly: bool + + class Config: + from_attributes = True diff --git a/dtos/hegemony_dto.py b/dtos/hegemony_dto.py new file mode 100644 index 0000000..efb4d51 --- /dev/null +++ b/dtos/hegemony_dto.py @@ -0,0 +1,15 @@ +from pydantic import BaseModel +from datetime import datetime + + +class HegemonyDTO(BaseModel): + timebin: datetime + originasn: int + asn: int + hege: float + af: int + asn_name: str + originasn_name: str + + class Config: + from_attributes = True diff --git a/dtos/hegemony_prefix_dto.py b/dtos/hegemony_prefix_dto.py new file mode 100644 index 0000000..0264ce7 --- /dev/null +++ b/dtos/hegemony_prefix_dto.py @@ -0,0 +1,24 @@ +from pydantic import BaseModel +from datetime import datetime + + +class HegemonyPrefixDTO(BaseModel): + timebin: datetime + prefix: str + originasn: int + country: str + asn: int + hege: float + af: int + visibility: float + rpki_status: str + irr_status: str + delegated_prefix_status: str + delegated_asn_status: str + descr: str + moas: bool + originasn_name: str + asn_name: str + + class Config: + from_attributes = True diff --git a/models/hegemony.py b/models/hegemony.py index db3f22b..07ce68d 100644 --- a/models/hegemony.py +++ b/models/hegemony.py @@ -1,4 +1,4 @@ -from sqlalchemy import Column, BigInteger, Float, Integer, ForeignKey, PrimaryKeyConstraint +from sqlalchemy import Column, BigInteger, Float, Integer, PrimaryKeyConstraint from sqlalchemy.dialects.postgresql import TIMESTAMP from sqlalchemy.orm import relationship from config.database import Base @@ -47,13 +47,18 @@ class Hegemony(Base): af = Column(Integer, default=0, nullable=False, doc='Address Family (IP version), values are either 4 or 6.') - asn_id = Column(BigInteger, - nullable=False, - doc='Dependency. Transit network commonly seen in BGP paths towards originasn.') + asn = Column('asn_id', BigInteger, + nullable=False, + doc='Dependency. Transit network commonly seen in BGP paths towards originasn.') - originasn_id = Column(BigInteger, + originasn = Column('originasn_id',BigInteger, nullable=False, doc='Dependent network, it can be any public ASN. Retrieve all dependencies of a network by setting only this parameter and a timebin.') - - + asn_relation = relationship('ASN', + primaryjoin='Hegemony.asn == ASN.number', + foreign_keys=[asn]) + + originasn_relation = relationship('ASN', + primaryjoin='Hegemony.originasn == ASN.number', + foreign_keys=[originasn]) diff --git a/models/hegemony_country.py b/models/hegemony_country.py index 38c5c96..543a08d 100644 --- a/models/hegemony_country.py +++ b/models/hegemony_country.py @@ -1,4 +1,4 @@ -from sqlalchemy import Column, Integer, Float, String, Boolean, ForeignKey, BigInteger,PrimaryKeyConstraint +from sqlalchemy import Column, Integer, Float, String, Boolean, ForeignKey, BigInteger, PrimaryKeyConstraint from sqlalchemy.dialects.postgresql import TIMESTAMP from sqlalchemy.orm import relationship from config.database import Base @@ -8,7 +8,7 @@ class HegemonyCountry(Base): __tablename__ = 'ihr_hegemony_country' __table_args__ = ( - PrimaryKeyConstraint('id','timebin'), + PrimaryKeyConstraint('id', 'timebin'), ) __hypertable__ = { @@ -47,12 +47,18 @@ class HegemonyCountry(Base): transitonly = Column(Boolean, default=False, nullable=False, doc='If True, then origin ASNs of BGP path are ignored (focus only on transit networks).') - asn_id = Column(BigInteger, - nullable=False, - doc='Dependency. Network commonly seen in BGP paths towards monitored country.') + asn = Column('asn_id', BigInteger, + nullable=False, + doc='Dependency. Network commonly seen in BGP paths towards monitored country.') - country_id = Column(String(4), - nullable=False, - doc='Monitored country. Retrieve all dependencies of a country by setting only this parameter and a timebin.') + country = Column('country_id', String(4), + nullable=False, + doc='Monitored country. Retrieve all dependencies of a country by setting only this parameter and a timebin.') - \ No newline at end of file + asn_relation = relationship('ASN', + primaryjoin='HegemonyCountry.asn == ASN.number', + foreign_keys=[asn]) + + country_relation = relationship('Country', + primaryjoin='HegemonyCountry.country == Country.code', + foreign_keys=[country]) diff --git a/models/hegemony_prefix.py b/models/hegemony_prefix.py index 62df125..1487916 100644 --- a/models/hegemony_prefix.py +++ b/models/hegemony_prefix.py @@ -8,7 +8,7 @@ class HegemonyPrefix(Base): __tablename__ = 'ihr_hegemony_prefix' __table_args__ = ( - PrimaryKeyConstraint('id','timebin'), + PrimaryKeyConstraint('id', 'timebin'), ) __hypertable__ = { @@ -62,7 +62,7 @@ class HegemonyPrefix(Base): doc="Status of the monitored prefix in the RIR's delegated stats. Status other than 'assigned' are usually considered as bogons.") delegated_asn_status = Column(String(32), nullable=False, - doc="Status of the origin ASN in the RIR's delegated stats. Status other than 'assigned' are usually considered as bogons.") + doc="Status of the origin ASN in the RIR's delegated stats. Status other than 'assigned' are usually considered as bogons.") descr = Column(String(64), nullable=False, doc='Prefix description from IRR (maximum 64 characters).') @@ -70,13 +70,26 @@ class HegemonyPrefix(Base): moas = Column(Boolean, default=False, nullable=False, doc='True if the prefix is originated by multiple ASNs.') - asn_id = Column(BigInteger, - nullable=False, - doc='Dependency. Network commonly seen in BGP paths towards monitored prefix.') - originasn_id = Column(BigInteger, - nullable=False, - doc='Network seen as originating the monitored prefix.') - - country_id = Column(String(4), - nullable=False, - doc="Country for the monitored prefix identified by Maxmind's Geolite2 geolocation database.") + asn = Column('asn_id', BigInteger, + nullable=False, + doc='Dependency. Network commonly seen in BGP paths towards monitored prefix.') + + originasn = Column('originasn_id', BigInteger, + nullable=False, + doc='Network seen as originating the monitored prefix.') + + country = Column('country_id', String(4), + nullable=False, + doc="Country for the monitored prefix identified by Maxmind's Geolite2 geolocation database.") + + asn_relation = relationship('ASN', + primaryjoin='HegemonyPrefix.asn == ASN.number', + foreign_keys=[asn]) + + originasn_relation = relationship('ASN', + primaryjoin='HegemonyPrefix.originasn == ASN.number', + foreign_keys=[originasn]) + + country_relation = relationship('Country', + primaryjoin='HegemonyPrefix.country == Country.code', + foreign_keys=[country]) diff --git a/repositories/hegemony_country_repository.py b/repositories/hegemony_country_repository.py new file mode 100644 index 0000000..075a664 --- /dev/null +++ b/repositories/hegemony_country_repository.py @@ -0,0 +1,59 @@ +from datetime import datetime +from sqlalchemy.orm import Session +from models.hegemony_country import HegemonyCountry +from typing import Optional, List, Tuple +from globals import page_size + + +class HegemonyCountryRepository: + def get_all( + self, + db: Session, + timebin_gte: Optional[datetime] = None, + timebin_lte: Optional[datetime] = None, + asn_ids: Optional[List[int]] = None, + countries: Optional[List[str]] = None, + af: Optional[int] = None, + weightscheme: Optional[str] = None, + transitonly: Optional[bool] = None, + hege: Optional[float] = None, + hege_gte: Optional[float] = None, + hege_lte: Optional[float] = None, + page: int = 1, + order_by: Optional[str] = None + ) -> Tuple[List[HegemonyCountry], int]: + query = db.query(HegemonyCountry) + + # Apply filters + if timebin_gte: + query = query.filter(HegemonyCountry.timebin >= timebin_gte) + if timebin_lte: + query = query.filter(HegemonyCountry.timebin <= timebin_lte) + if asn_ids: + query = query.filter(HegemonyCountry.asn.in_(asn_ids)) + if countries: + query = query.filter(HegemonyCountry.country.in_(countries)) + if af is not None: + query = query.filter(HegemonyCountry.af == af) + if weightscheme is not None: + query = query.filter(HegemonyCountry.weightscheme == weightscheme) + if transitonly is not None: + query = query.filter(HegemonyCountry.transitonly == transitonly) + if hege is not None: + query = query.filter(HegemonyCountry.hege == hege) + if hege_gte is not None: + query = query.filter(HegemonyCountry.hege >= hege_gte) + if hege_lte is not None: + query = query.filter(HegemonyCountry.hege <= hege_lte) + + total_count = query.count() + + # Apply ordering + if order_by and hasattr(HegemonyCountry, order_by): + query = query.order_by(getattr(HegemonyCountry, order_by)) + + # Apply pagination + offset = (page - 1) * page_size + results = query.offset(offset).limit(page_size).all() + + return results, total_count diff --git a/repositories/hegemony_prefix_repository.py b/repositories/hegemony_prefix_repository.py new file mode 100644 index 0000000..8a30099 --- /dev/null +++ b/repositories/hegemony_prefix_repository.py @@ -0,0 +1,79 @@ +from datetime import datetime +from sqlalchemy.orm import Session +from models.hegemony_prefix import HegemonyPrefix +from typing import Optional, List, Tuple +from globals import page_size + + +class HegemonyPrefixRepository: + def get_all( + self, + db: Session, + timebin_gte: Optional[datetime] = None, + timebin_lte: Optional[datetime] = None, + prefixes: Optional[List[str]] = None, + asn_ids: Optional[List[int]] = None, + originasn_ids: Optional[List[int]] = None, + countries: Optional[List[str]] = None, + rpki_status: Optional[str] = None, + irr_status: Optional[str] = None, + delegated_prefix_status: Optional[str] = None, + delegated_asn_status: Optional[str] = None, + af: Optional[int] = None, + hege: Optional[float] = None, + hege_gte: Optional[float] = None, + hege_lte: Optional[float] = None, + origin_only: Optional[bool] = None, + page: int = 1, + order_by: Optional[str] = None + ) -> Tuple[List[HegemonyPrefix], int]: + query = db.query(HegemonyPrefix) + + # Apply filters + if timebin_gte: + query = query.filter(HegemonyPrefix.timebin >= timebin_gte) + if timebin_lte: + query = query.filter(HegemonyPrefix.timebin <= timebin_lte) + if prefixes: + query = query.filter(HegemonyPrefix.prefix.in_(prefixes)) + if asn_ids: + query = query.filter(HegemonyPrefix.asn.in_(asn_ids)) + if originasn_ids: + query = query.filter(HegemonyPrefix.originasn.in_(originasn_ids)) + if countries: + query = query.filter(HegemonyPrefix.country.in_(countries)) + if rpki_status: + query = query.filter( + HegemonyPrefix.rpki_status.contains(rpki_status)) + if irr_status: + query = query.filter( + HegemonyPrefix.irr_status.contains(irr_status)) + if delegated_prefix_status: + query = query.filter( + HegemonyPrefix.delegated_prefix_status.contains(delegated_prefix_status)) + if delegated_asn_status: + query = query.filter( + HegemonyPrefix.delegated_asn_status.contains(delegated_asn_status)) + if af is not None: + query = query.filter(HegemonyPrefix.af == af) + if hege is not None: + query = query.filter(HegemonyPrefix.hege == hege) + if hege_gte is not None: + query = query.filter(HegemonyPrefix.hege >= hege_gte) + if hege_lte is not None: + query = query.filter(HegemonyPrefix.hege <= hege_lte) + if origin_only: + query = query.filter( + HegemonyPrefix.originasn == HegemonyPrefix.asn) + + total_count = query.count() + + # Apply ordering + if order_by and hasattr(HegemonyPrefix, order_by): + query = query.order_by(getattr(HegemonyPrefix, order_by)) + + # Apply pagination + offset = (page - 1) * page_size + results = query.offset(offset).limit(page_size).all() + + return results, total_count diff --git a/repositories/hegemony_repository.py b/repositories/hegemony_repository.py new file mode 100644 index 0000000..4014eb5 --- /dev/null +++ b/repositories/hegemony_repository.py @@ -0,0 +1,53 @@ +from datetime import datetime +from sqlalchemy.orm import Session +from models.hegemony import Hegemony +from typing import Optional, List, Tuple +from globals import page_size + + +class HegemonyRepository: + def get_all( + self, + db: Session, + timebin_gte: Optional[datetime] = None, + timebin_lte: Optional[datetime] = None, + asn_ids: Optional[List[int]] = None, + originasn_ids: Optional[List[int]] = None, + af: Optional[int] = None, + hege: Optional[float] = None, + hege_gte: Optional[float] = None, + hege_lte: Optional[float] = None, + page: int = 1, + order_by: Optional[str] = None + ) -> Tuple[List[Hegemony], int]: + query = db.query(Hegemony) + + # Apply filters + if timebin_gte: + query = query.filter(Hegemony.timebin >= timebin_gte) + if timebin_lte: + query = query.filter(Hegemony.timebin <= timebin_lte) + if asn_ids: + query = query.filter(Hegemony.asn.in_(asn_ids)) + if originasn_ids: + query = query.filter(Hegemony.originasn.in_(originasn_ids)) + if af is not None: + query = query.filter(Hegemony.af == af) + if hege is not None: + query = query.filter(Hegemony.hege == hege) + if hege_gte: + query = query.filter(Hegemony.hege >= hege_gte) + if hege_lte: + query = query.filter(Hegemony.hege <= hege_lte) + + total_count = query.count() + + # Apply ordering + if order_by and hasattr(Hegemony, order_by): + query = query.order_by(getattr(Hegemony, order_by)) + + # Apply pagination + offset = (page - 1) * page_size + results = query.offset(offset).limit(page_size).all() + + return results, total_count diff --git a/services/hegemony_service.py b/services/hegemony_service.py index 1e3bf89..6b1eef4 100644 --- a/services/hegemony_service.py +++ b/services/hegemony_service.py @@ -3,6 +3,12 @@ from dtos.hegemony_cone_dto import HegemonyConeDTO from repositories.hegemony_alarms_repository import HegemonyAlarmsRepository from dtos.hegemony_alarms_dto import HegemonyAlarmsDTO +from repositories.hegemony_country_repository import HegemonyCountryRepository +from dtos.hegemony_country_dto import HegemonyCountryDTO +from repositories.hegemony_repository import HegemonyRepository +from dtos.hegemony_dto import HegemonyDTO +from repositories.hegemony_prefix_repository import HegemonyPrefixRepository +from dtos.hegemony_prefix_dto import HegemonyPrefixDTO from typing import Optional, List, Tuple from datetime import datetime @@ -11,6 +17,9 @@ class HegemonyService: def __init__(self): self.hegemony_cone_repository = HegemonyConeRepository() self.hegemony_alarms_repository = HegemonyAlarmsRepository() + self.hegemony_country_repository = HegemonyCountryRepository() + self.hegemony_repository = HegemonyRepository() + self.hegemony_prefix_repository = HegemonyPrefixRepository() def get_hegemony_cones( self, @@ -80,3 +89,155 @@ def get_hegemony_alarms( asn_name=alarm.asn_relation.name if alarm.asn_relation else None, originasn_name=alarm.originasn_relation.name if alarm.originasn_relation else None ) for alarm in alarms], total_count + + def get_hegemony_countries( + self, + db: Session, + timebin_gte: Optional[datetime] = None, + timebin_lte: Optional[datetime] = None, + asn_ids: Optional[List[int]] = None, + countries: Optional[List[str]] = None, + af: Optional[int] = None, + weightscheme: Optional[str] = None, + transitonly: Optional[bool] = None, + hege: Optional[float] = None, + hege_gte: Optional[float] = None, + hege_lte: Optional[float] = None, + page: int = 1, + order_by: Optional[str] = None + ) -> Tuple[List[HegemonyCountryDTO], int]: + """ + Get hegemony country data with filtering. + """ + countries_data, total_count = self.hegemony_country_repository.get_all( + db, + timebin_gte=timebin_gte, + timebin_lte=timebin_lte, + asn_ids=asn_ids, + countries=countries, + af=af, + weightscheme=weightscheme, + transitonly=transitonly, + hege=hege, + hege_gte=hege_gte, + hege_lte=hege_lte, + page=page, + order_by=order_by + ) + + return [HegemonyCountryDTO( + timebin=country.timebin, + country=country.country, + asn=country.asn, + hege=country.hege, + af=country.af, + asn_name=country.asn_relation.name if country.asn_relation else None, + weight=country.weight, + weightscheme=country.weightscheme, + transitonly=country.transitonly + ) for country in countries_data], total_count + + def get_hegemony( + self, + db: Session, + timebin_gte: Optional[datetime] = None, + timebin_lte: Optional[datetime] = None, + asn_ids: Optional[List[int]] = None, + originasn_ids: Optional[List[int]] = None, + af: Optional[int] = None, + hege: Optional[float] = None, + hege_gte: Optional[float] = None, + hege_lte: Optional[float] = None, + page: int = 1, + order_by: Optional[str] = None + ) -> Tuple[List[HegemonyDTO], int]: + """ + Get hegemony data with filtering. + """ + hegemony_data, total_count = self.hegemony_repository.get_all( + db, + timebin_gte=timebin_gte, + timebin_lte=timebin_lte, + asn_ids=asn_ids, + originasn_ids=originasn_ids, + af=af, + hege=hege, + hege_gte=hege_gte, + hege_lte=hege_lte, + page=page, + order_by=order_by + ) + + return [HegemonyDTO( + timebin=hegemony.timebin, + originasn=hegemony.originasn, + asn=hegemony.asn, + hege=hegemony.hege, + af=hegemony.af, + asn_name=hegemony.asn_relation.name if hegemony.asn_relation else None, + originasn_name=hegemony.originasn_relation.name if hegemony.originasn_relation else None + ) for hegemony in hegemony_data], total_count + + def get_hegemony_prefixes( + self, + db: Session, + timebin_gte: Optional[datetime] = None, + timebin_lte: Optional[datetime] = None, + prefixes: Optional[List[str]] = None, + asn_ids: Optional[List[int]] = None, + originasn_ids: Optional[List[int]] = None, + countries: Optional[List[str]] = None, + rpki_status: Optional[str] = None, + irr_status: Optional[str] = None, + delegated_prefix_status: Optional[str] = None, + delegated_asn_status: Optional[str] = None, + af: Optional[int] = None, + hege: Optional[float] = None, + hege_gte: Optional[float] = None, + hege_lte: Optional[float] = None, + origin_only: Optional[bool] = None, + page: int = 1, + order_by: Optional[str] = None + ) -> Tuple[List[HegemonyPrefixDTO], int]: + """ + Get hegemony prefix data with filtering. + """ + prefixes_data, total_count = self.hegemony_prefix_repository.get_all( + db, + timebin_gte=timebin_gte, + timebin_lte=timebin_lte, + prefixes=prefixes, + asn_ids=asn_ids, + originasn_ids=originasn_ids, + countries=countries, + rpki_status=rpki_status, + irr_status=irr_status, + delegated_prefix_status=delegated_prefix_status, + delegated_asn_status=delegated_asn_status, + af=af, + hege=hege, + hege_gte=hege_gte, + hege_lte=hege_lte, + origin_only=origin_only, + page=page, + order_by=order_by + ) + + return [HegemonyPrefixDTO( + timebin=prefix.timebin, + prefix=prefix.prefix, + originasn=prefix.originasn, + country=prefix.country, + asn=prefix.asn, + hege=prefix.hege, + af=prefix.af, + visibility=prefix.visibility, + rpki_status=prefix.rpki_status, + irr_status=prefix.irr_status, + delegated_prefix_status=prefix.delegated_prefix_status, + delegated_asn_status=prefix.delegated_asn_status, + descr=prefix.descr, + moas=prefix.moas, + originasn_name=prefix.originasn_relation.name if prefix.originasn_relation else None, + asn_name=prefix.asn_relation.name if prefix.asn_relation else None + ) for prefix in prefixes_data], total_count