Added /hegemony, /hegemony/countries and /hegemony/prefixes endpoints

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

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

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

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])

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.')
 
-
+    asn_relation = relationship('ASN',
+                                primaryjoin='HegemonyCountry.asn == ASN.number',
+                                foreign_keys=[asn])
+
+    country_relation = relationship('Country',
+                                      primaryjoin='HegemonyCountry.country == Country.code',
+                                      foreign_keys=[country])

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,21 +62,34 @@ 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).')
 
     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])

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

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

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