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/score_delegation/service/simple_obfuscation.py

@@ -1,5 +1,5 @@
 from datetime import datetime
-from typing import Protocol, runtime_checkable, Tuple, Container, List
+from typing import Protocol, runtime_checkable, Tuple, Container, List, Optional
 
 from app.context.manager import Context
 from app.extensions import db
@@ -9,6 +9,7 @@
 from app.modules.dto import ScoreDelegationPayload
 from app.modules.score_delegation import core
 from app.modules.score_delegation.core import ActionType
+from app.modules.user.antisybil.service.passport import GitcoinAntisybilDTO
 from app.pydantic import Model
 
 from flask import current_app as app
@@ -20,7 +21,7 @@
 class Antisybil(Protocol):
  def fetch_antisybil_status(
  self, _: Context, user_address: str
- ) -> Tuple[float, datetime, any]:
+ ) -> Optional[GitcoinAntisybilDTO]:
  ...
 
  def update_antisybil_status(
@@ -85,7 +86,7 @@ def _delegation(
  hashed_addresses = get_hashed_addresses(
  payload.primary_addr, payload.secondary_addr
  )
- score, expires_at, stamps = self.antisybil.fetch_antisybil_status(
+ score, stamps = self.antisybil.fetch_antisybil_status(
  context, payload.secondary_addr
  )
  secondary_budget = self.user_deposits_service.get_user_effective_deposit(
@@ -95,10 +96,10 @@ def _delegation(
  context,
  hashed_addresses=hashed_addresses,
  payload=payload,
- score=score,
+ score=score.score,
  secondary_budget=secondary_budget,
  action_type=action,
  )
  self.antisybil.update_antisybil_status(
- context, payload.primary_addr, score, expires_at, stamps
+ context, payload.primary_addr, score.score, score.expires_at, stamps
  )

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

backend/tests/modules/score_delegation/test_simple_obfuscation.py

@@ -9,6 +9,7 @@
 )
 from app.infrastructure import database
 from app.modules.dto import ScoreDelegationPayload
+from app.modules.user.antisybil.service.passport import GitcoinAntisybilDTO
 from app.modules.score_delegation.service.simple_obfuscation import (
  SimpleObfuscationDelegation,
  SimpleObfuscationDelegationVerifier,
@@ -39,8 +40,10 @@ def test_delegation(
  user_deposits = CalculatedUserDeposits(events_generator=mock_empty_events_generator)
  antisybil = Mock()
  antisybil.fetch_antisybil_status.return_value = (
- 20,
- "4024-05-22T14:46:46.810800+00:00",
+ GitcoinAntisybilDTO(
+ score=20,
+ expires_at="4024-05-22T14:46:46.810800+00:00",
+ ),
  ["stamp"],
  )
  service = SimpleObfuscationDelegation(
@@ -59,8 +62,10 @@ def test_delegation_disabled_when_secondary_is_locking(
  user_deposits = CalculatedUserDeposits(events_generator=mock_events_generator)
  antisybil = Mock()
  antisybil.fetch_antisybil_status.return_value = (
- 20,
- "4024-05-22T14:46:46.810800+00:00",
+ GitcoinAntisybilDTO(
+ score=20,
+ expires_at="4024-05-22T14:46:46.810800+00:00",
+ ),
  ["stamp"],
  )
  payload = ScoreDelegationPayload(
@@ -83,8 +88,10 @@ def test_disable_recalculation_when_secondary_address_is_used(
  user_deposits = CalculatedUserDeposits(events_generator=mock_empty_events_generator)
  antisybil = Mock()
  antisybil.fetch_antisybil_status.return_value = (
- 20,
- "4024-05-22T14:46:46.810800+00:00",
+ GitcoinAntisybilDTO(
+ score=20,
+ expires_at="4024-05-22T14:46:46.810800+00:00",
+ ),
  ["stamp"],
  )
  service = SimpleObfuscationDelegation(
@@ -93,8 +100,7 @@ def test_disable_recalculation_when_secondary_address_is_used(
  service.delegate(context, payload)
 
  antisybil.fetch_antisybil_status.return_value = (
- 25,
- "4024-05-22T14:46:46.810800+00:00",
+ GitcoinAntisybilDTO(score=25, expires_at="4024-05-22T14:46:46.810800+00:00"),
  ["stamp"],
  )
  payload = ScoreDelegationPayload(
@@ -115,8 +121,7 @@ def test_recalculation_when_delegation_is_not_done(
  user_deposits = CalculatedUserDeposits(events_generator=mock_empty_events_generator)
  antisybil = Mock()
  antisybil.fetch_antisybil_status.return_value = (
- 20,
- "4024-05-22T14:46:46.810800+00:00",
+ GitcoinAntisybilDTO(score=20, expires_at="4024-05-22T14:46:46.810800+00:00"),
  ["stamp"],
  )
  service = SimpleObfuscationDelegation(
@@ -125,8 +130,7 @@ def test_recalculation_when_delegation_is_not_done(
  service.delegate(context, payload)
 
  antisybil.fetch_antisybil_status.return_value = (
- 25,
- "4024-05-22T14:46:46.810800+00:00",
+ GitcoinAntisybilDTO(25, "4024-05-22T14:46:46.810800+00:00"),
  ["stamp"],
  )
  payload = ScoreDelegationPayload(