cr: lots of minor changes

* use DTO instead of tuples
* return partial results when fetching antisybil status - they are enough for FE
* lineup declared and actual types

backend/app/infrastructure/routes/user.py

@@ -190,23 +190,24 @@ def patch(self, user_address: str):
 )
 class AntisybilStatus(OctantResource):
     @ns.doc(
-        description="Returns user's antisybil status.",
+        description="""Returns user's antisybil status.""",
     )
     @ns.marshal_with(user_antisybil_status_model)
     @ns.response(200, "User's cached antisybil status retrieved")
+    @ns.response(404, {"status": "Unknown"})
     def get(self, user_address: str):
         app.logger.debug(f"Getting user {user_address} cached antisybil status")
 
         antisybil_status = get_user_antisybil_status(user_address)
         app.logger.debug(f"User {user_address} antisybil status: {antisybil_status}")
-        if antisybil_status is None:
+        if antisybil_status == (None, None):
             return {"status": "Unknown"}, 404
-        (score, expires_at), (has_sbt, _) = antisybil_status
+        gitcoin, passport = antisybil_status
         return {
-            "holonym": has_sbt,
+            "holonym": passport.has_sbt if passport else None,
             "status": "Known",
-            "score": score,
-            "expires_at": int(expires_at.timestamp()),
+            "score": gitcoin.score if gitcoin else None,
+            "expires_at": int(gitcoin.expires_at.timestamp()) if gitcoin else None,
         }, 200
 
     @ns.doc(
@@ -219,9 +220,9 @@ def put(self, user_address: str):
         app.logger.info(f"Updating user {user_address} antisybil status")
         status = update_user_antisybil_status(user_address)
         app.logger.info(f"Got status for user {user_address} = {status}")
-        (score, expires_at), (has_sbt, _) = status
+        passport, sbt = status
         app.logger.info(
-            f"User {user_address} antisybil status refreshed {[score, has_sbt, expires_at]}"
+            f"User {user_address} antisybil status refreshed {[passport.score, sbt.has_sbt, passport.expires_at]}"
         )
 
         return {}, 204

backend/app/modules/uq/service/preliminary.py

@@ -1,6 +1,5 @@
-from datetime import datetime
 from decimal import Decimal
-from typing import Protocol, List, Optional, Tuple, runtime_checkable
+from typing import Protocol, Optional, runtime_checkable
 
 from app.context.manager import Context
 from app.infrastructure.database.uniqueness_quotient import (
@@ -9,21 +8,23 @@
 )
 from app.modules.uq.core import calculate_uq
 from app.pydantic import Model
+from app.modules.user.antisybil.service.holonym import HolonymAntisybilDTO
+from app.modules.user.antisybil.service.passport import GitcoinAntisybilDTO
 
 
 @runtime_checkable
 class Passport(Protocol):
     def get_antisybil_status(
         self, _: Context, user_address: str
-    ) -> Optional[Tuple[float, datetime]]:
+    ) -> Optional[GitcoinAntisybilDTO]:
         ...
 
 
 @runtime_checkable
 class Holonym(Protocol):
     def get_sbt_status(
         self, _: Context, user_address: str
-    ) -> Optional[Tuple[bool, List[str]]]:
+    ) -> Optional[HolonymAntisybilDTO]:
         ...
 
 

backend/app/modules/user/antisybil/controller.py

@@ -1,41 +1,37 @@
-from datetime import datetime
-from typing import List, Optional, Tuple
+from typing import Optional, Tuple
 
 from app.context.epoch_state import EpochState
 from app.context.manager import state_context
 from app.modules.registry import get_services
+from app.modules.user.antisybil.service.holonym import HolonymAntisybilDTO
+from app.modules.user.antisybil.service.passport import GitcoinAntisybilDTO
 
 
 def get_user_antisybil_status(
     user_address: str,
-) -> Optional[Tuple[Tuple[int, datetime], Tuple[bool, List[str]]]]:
+) -> Tuple[Optional[GitcoinAntisybilDTO], Optional[HolonymAntisybilDTO]]:
     context = state_context(EpochState.CURRENT)
     passport = get_services(context.epoch_state).user_antisybil_passport_service
-    passport_status = passport.get_antisybil_status(context, user_address)
+    gpscore = passport.get_antisybil_status(context, user_address)
     holonym = get_services(context.epoch_state).user_antisybil_holonym_service
-    holonym_status = holonym.get_sbt_status(context, user_address)
-    if passport_status is None or holonym_status is None:
-        return None
-
-    return (passport_status, holonym_status)
+    sbt = holonym.get_sbt_status(context, user_address)
+    return (gpscore, sbt)
 
 
 def update_user_antisybil_status(
     user_address: str,
-) -> Tuple[Tuple[int, datetime], Tuple[bool, List[str]]]:
+) -> Tuple[GitcoinAntisybilDTO, HolonymAntisybilDTO]:
     context = state_context(EpochState.CURRENT)
     passport = get_services(context.epoch_state).user_antisybil_passport_service
 
-    score, expires_at, all_stamps = passport.fetch_antisybil_status(
-        context, user_address
-    )
+    gpscore, stamps = passport.fetch_antisybil_status(context, user_address)
     passport.update_antisybil_status(
-        context, user_address, score, expires_at, all_stamps
+        context, user_address, gpscore.score, gpscore.expires_at, stamps
     )
 
     holonym = get_services(context.epoch_state).user_antisybil_holonym_service
 
-    has_sbt, cred_type = holonym.fetch_sbt_status(context, user_address)
-    holonym.update_sbt_status(context, user_address, has_sbt, cred_type)
+    sbt = holonym.fetch_sbt_status(context, user_address)
+    holonym.update_sbt_status(context, user_address, sbt.has_sbt, sbt.sbt_details)
 
-    return ((score, expires_at), (has_sbt, cred_type))
+    return (gpscore, sbt)

backend/app/modules/user/antisybil/service/holonym.py

@@ -1,21 +1,29 @@
+from dataclasses import dataclass
 from flask import current_app as app
 
 from eth_utils.address import to_checksum_address
 
 import json
-from typing import List, Optional, Tuple
+from typing import List, Optional
 
 from app.extensions import db
 from app.infrastructure import database
+from app.infrastructure.external_api.holonym.antisybil import check
 from app.context.manager import Context
 from app.pydantic import Model
 from app.exceptions import UserNotFound
 
 
+@dataclass
+class HolonymAntisybilDTO:
+    has_sbt: bool
+    sbt_details: List[str]
+
+
 class HolonymAntisybil(Model):
     def get_sbt_status(
         self, _: Context, user_address: str
-    ) -> Optional[Tuple[bool, List[str]]]:
+    ) -> Optional[HolonymAntisybilDTO]:
         user_address = to_checksum_address(user_address)
         try:
             entry = database.user_antisybil.get_sbt_by_address(user_address)
@@ -26,16 +34,15 @@ def get_sbt_status(
             raise ex
 
         if entry is not None:
-            return entry.has_sbt, json.loads(entry.sbt_details)
+            return HolonymAntisybilDTO(
+                has_sbt=entry.has_sbt, sbt_details=json.loads(entry.sbt_details)
+            )
         return None
 
-    def fetch_sbt_status(
-        self, _: Context, user_address: str
-    ) -> Optional[Tuple[bool, List[str]]]:
-        from app.infrastructure.external_api.holonym.antisybil import check
-
+    def fetch_sbt_status(self, _: Context, user_address: str) -> HolonymAntisybilDTO:
         user_address = to_checksum_address(user_address)
-        return check(user_address)
+        has_sbt, sbt_type = check(user_address)
+        return HolonymAntisybilDTO(has_sbt=has_sbt, sbt_details=sbt_type)
 
     def update_sbt_status(
         self, _: Context, user_address: str, has_sbt: bool, cred_type: List[str]

backend/app/modules/user/antisybil/service/passport.py

@@ -1,3 +1,4 @@
+from dataclasses import dataclass
 from flask import current_app as app
 
 from eth_utils.address import to_checksum_address
@@ -23,10 +24,16 @@
 )
 
 
+@dataclass
+class GitcoinAntisybilDTO:
+    score: float
+    expires_at: datetime
+
+
 class GitcoinPassportAntisybil(Model):
     def get_antisybil_status(
         self, _: Context, user_address: str
-    ) -> Optional[Tuple[float, datetime]]:
+    ) -> Optional[GitcoinAntisybilDTO]:
         user_address = to_checksum_address(user_address)
         try:
             score = database.user_antisybil.get_score_by_address(user_address)
@@ -38,12 +45,12 @@ def get_antisybil_status(
         if score is not None:
             if user_address in GUEST_LIST and not _has_guest_stamp_applied_by_gp(score):
                 score.score = score.score + 21.0
-            return score.score, score.expires_at
+            return GitcoinAntisybilDTO(score=score.score, expires_at=score.expires_at)
         return None
 
     def fetch_antisybil_status(
         self, _: Context, user_address: str
-    ) -> Tuple[float, datetime, any]:
+    ) -> Tuple[GitcoinAntisybilDTO, any]:
         score = issue_address_for_scoring(user_address)
 
         def _retry_fetch():
@@ -62,7 +69,10 @@ def _retry_fetch():
             expires_at = _parse_expiration_date(
                 min([stamp["credential"]["expirationDate"] for stamp in valid_stamps])
             )
-        return float(score["score"]), expires_at, all_stamps
+        return (
+            GitcoinAntisybilDTO(score=float(score["score"]), expires_at=expires_at),
+            all_stamps,
+        )
 
     def update_antisybil_status(
         self,

backend/migrations/versions/003a0cad5494_add_cached_status_for_holonym_sbt.py

@@ -17,7 +17,6 @@
 
 
 def upgrade():
-    # ### commands auto generated by Alembic - please adjust! ###
     op.create_table(
         "holonym_sbts",
         sa.Column("id", sa.Integer(), nullable=False),
@@ -31,10 +30,7 @@ def upgrade():
         ),
         sa.PrimaryKeyConstraint("id"),
     )
-    # ### end Alembic commands ###
 
 
 def downgrade():
-    # ### commands auto generated by Alembic - please adjust! ###
     op.drop_table("holonym_sbts")
-    # ### end Alembic commands ###

backend/tests/api-e2e/test_api_antisybil.py

@@ -7,8 +7,9 @@
 
 @pytest.mark.api
 def test_holonym(client: Client):
-    # check status for a known address with SBT before caching
     address_with_sbt = "0x76273DCC41356e5f0c49bB68e525175DC7e83417"
+
+    # check status for a known address with SBT before caching
     database.user.add_user(address_with_sbt)
     _, code = client.get_antisybil_score(address_with_sbt)
     assert code == 404  # score for this user is not cached