feat: v2 history endpoint with permissions

api/account/admin.py

@@ -158,7 +158,13 @@ def edit_selected(modeladmin, request, queryset):
 
 
 class APIKeyPermissionsAdmin(ScorerModelAdmin):
-    list_display = ("id", "submit_passports", "read_scores", "create_scorers")
+    list_display = (
+        "id",
+        "submit_passports",
+        "read_scores",
+        "create_scorers",
+        "historical_endpoint",
+    )
 
 
 svg_widget = AceWidget(

api/account/migrations/0038_accountapikey_historical_endpoint.py

@@ -0,0 +1,17 @@
+# Generated by Django 4.2.6 on 2024-10-21 20:57
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("account", "0037_customization_partner_name"),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name="accountapikey",
+            name="historical_endpoint",
+            field=models.BooleanField(default=False),
+        ),
+    ]

api/account/models.py

@@ -270,6 +270,7 @@ class AccountAPIKey(AbstractAPIKey):
     submit_passports = models.BooleanField(default=True)
     read_scores = models.BooleanField(default=True)
     create_scorers = models.BooleanField(default=False)
+    historical_endpoint = models.BooleanField(default=False)
 
     @admin.display(description="Rate Limit")
     def rate_limit_display(self):

api/registry/api/common.py

@@ -11,7 +11,11 @@
     ErrorMessageResponse,
 )
 from registry.api.utils import ApiKey, check_rate_limit, with_read_db
-from registry.exceptions import InvalidLimitException, api_get_object_or_404
+from registry.exceptions import (
+    InvalidAPIKeyPermissions,
+    InvalidLimitException,
+    api_get_object_or_404,
+)
 from registry.models import Event, Score
 from registry.utils import (
     decode_cursor,
@@ -30,6 +34,9 @@ def get_score_history(
     token: str = None,
     limit: int = 1000,
 ) -> CursorPaginatedHistoricalScoreResponse:
+    if not request.api_key.historical_endpoint:
+        raise InvalidAPIKeyPermissions()
+
     check_rate_limit(request)
 
     if limit > 1000:
@@ -79,7 +86,7 @@ def get_score_history(
             )
             return response
 
-        # Scenario 2 - Snapshot for 1 addresses and timestamp
+        # Scenario 2 - Snapshot for 1 address and timestamp
         # the user has passed in the created_at and address
         # In this case only 1 result will be returned
         if address and created_at:
@@ -246,6 +253,9 @@ def get_score_history(
 If a specific address is provided, then the list will be of max length 1.\n
 \n
 Note: results will be sorted descending by `["created_at", "id"]`
+\n
+\n
+To access this endpoint, you must submit your use case and be approved by the Passport team. To do so, please fill out the following form, making sure to provide a detailed description of your use case. The Passport team typically reviews and responds to form responses within 48 hours. <a href="https://forms.gle/4GyicBfhtHW29eEu8" target="_blank">https://forms.gle/4GyicBfhtHW29eEu8</a>
     """,
     "handler": get_score_history,
 }

api/registry/api/v1.py

@@ -1,9 +1,15 @@
 from typing import List, Optional
 from urllib.parse import urljoin
 
-import api_logging as logging
 import django_filters
 import requests
+from django.conf import settings
+from django.core.cache import cache
+from ninja import Router
+from ninja.pagination import paginate
+from ninja_extra.exceptions import APIException
+
+import api_logging as logging
 from account.api import UnauthorizedException, create_community_for_account
 
 # --- Deduplication Modules
@@ -14,11 +20,6 @@
     Rules,
 )
 from ceramic_cache.models import CeramicCache
-from django.conf import settings
-from django.core.cache import cache
-from ninja import Router
-from ninja.pagination import paginate
-from ninja_extra.exceptions import APIException
 from registry.api import common
 from registry.api.schema import (
     CursorPaginatedHistoricalScoreResponse,

api/scorer/test/conftest.py

@@ -69,6 +69,7 @@ def scorer_api_key(scorer_account):
         name="Token for user 1",
         rate_limit="3/30seconds",
         analysis_rate_limit="3/30seconds",
+        historical_endpoint=True,
     )
     return secret
 
@@ -80,6 +81,7 @@ def scorer_api_key_no_permissions(scorer_account):
         name="Token for user 1",
         rate_limit="3/30seconds",
         read_scores=False,
+        historical_endpoint=False,
     )
     return secret
 

api/v2/test/test_historical_score_endpoint.py

@@ -0,0 +1,203 @@
+from datetime import datetime, timedelta
+from unittest.mock import MagicMock, patch
+
+import pytest
+from django.test import Client
+from freezegun import freeze_time
+from web3 import Web3
+
+from ceramic_cache.models import CeramicCache
+from registry.models import Event
+
+pytestmark = pytest.mark.django_db
+
+
+class TestPassportGetHistoricalScore:
+    base_url = "/v2/stamps"
+
+    def test_get_historical_score_response_with_no_scores(
+        self,
+        scorer_account,
+        scorer_api_key,
+        scorer_community_with_binary_scorer,
+    ):
+        client = Client()
+        response = client.get(
+            f"{self.base_url}/{scorer_community_with_binary_scorer.id}/score/{scorer_account.address}/history",
+            HTTP_AUTHORIZATION="Token " + scorer_api_key,
+        )
+        response_data = response.json()
+
+        assert response.status_code == 200
+        assert len(response_data["items"]) == 0
+
+    def test_get_historical_score_response_with_scores(
+        self,
+        scorer_account,
+        scorer_api_key,
+        scorer_community_with_binary_scorer,
+    ):
+        i = 1
+        while True:
+            Event.objects.create(
+                action=Event.Action.SCORE_UPDATE,
+                address=scorer_account.address,
+                community=scorer_community_with_binary_scorer,
+                data={
+                    "score": 1,
+                    "evidence": {
+                        "rawScore": 20 + i,
+                        "type": "binary",
+                        "success": True,
+                        "threshold": 5,
+                    },
+                },
+            )
+            i += 1
+            if i > 10:
+                break
+
+        client = Client()
+        response = client.get(
+            f"{self.base_url}/{scorer_community_with_binary_scorer.id}/score/{scorer_account.address}/history",
+            HTTP_AUTHORIZATION="Token " + scorer_api_key,
+        )
+        response_data = response.json()
+
+        assert response.status_code == 200
+        assert len(response_data["items"]) == 10
+
+    def test_get_historical_score_limit_exceeding_1000(
+        self,
+        scorer_account,
+        scorer_api_key,
+        scorer_community_with_binary_scorer,
+    ):
+        client = Client()
+        response = client.get(
+            f"{self.base_url}/{scorer_community_with_binary_scorer.id}/score/{scorer_account.address}/history?limit=1001",
+            HTTP_AUTHORIZATION="Token " + scorer_api_key,
+        )
+        assert response.status_code == 400
+        assert "Invalid limit" in response.json()["detail"]
+
+    def test_get_historical_score_invalid_api_key_permissions(
+        self,
+        scorer_account,
+        scorer_api_key_no_permissions,
+        scorer_community_with_binary_scorer,
+    ):
+        client = Client()
+        response = client.get(
+            f"{self.base_url}/{scorer_community_with_binary_scorer.id}/score/{scorer_account.address}/history",
+            HTTP_AUTHORIZATION="Token " + scorer_api_key_no_permissions,
+        )
+        assert response.status_code == 403
+
+    def test_get_historical_score_address_no_created_at(
+        self,
+        scorer_account,
+        scorer_api_key,
+        scorer_community_with_binary_scorer,
+    ):
+        # Create a score event
+        Event.objects.create(
+            action=Event.Action.SCORE_UPDATE,
+            address=scorer_account.address,
+            community=scorer_community_with_binary_scorer,
+            data={
+                "score": 1,
+                "evidence": {
+                    "rawScore": 20,
+                    "type": "binary",
+                    "success": True,
+                    "threshold": 5,
+                },
+            },
+        )
+
+        client = Client()
+        response = client.get(
+            f"{self.base_url}/{scorer_community_with_binary_scorer.id}/score/{scorer_account.address}/history",
+            HTTP_AUTHORIZATION="Token " + scorer_api_key,
+        )
+        response_data = response.json()
+
+        assert response.status_code == 200
+        assert len(response_data["items"]) == 1
+        assert (
+            response_data["items"][0]["address"].lower()
+            == scorer_account.address.lower()
+        )
+
+    @freeze_time("2023-01-01")
+    def test_get_historical_score_address_and_created_at(
+        self,
+        scorer_account,
+        scorer_api_key,
+        scorer_community_with_binary_scorer,
+    ):
+        # Create multiple score events
+        for i in range(3):
+            Event.objects.create(
+                action=Event.Action.SCORE_UPDATE,
+                address=scorer_account.address,
+                community=scorer_community_with_binary_scorer,
+                data={
+                    "score": i,
+                    "evidence": {
+                        "rawScore": 20 + i,
+                        "type": "binary",
+                        "success": True,
+                        "threshold": 5,
+                    },
+                },
+                created_at=datetime.now() - timedelta(days=i),
+            )
+
+        client = Client()
+        response = client.get(
+            f"{self.base_url}/{scorer_community_with_binary_scorer.id}/score/{scorer_account.address}/history?created_at=2023-01-01T00:00:00",
+            HTTP_AUTHORIZATION="Token " + scorer_api_key,
+        )
+        response_data = response.json()
+
+        assert response.status_code == 200
+        assert len(response_data["items"]) == 1
+        assert response_data["items"][0]["score"] == "2"
+
+    @freeze_time("2023-01-01")
+    def test_get_historical_score_created_at_no_address(
+        self,
+        scorer_account,
+        scorer_api_key,
+        scorer_community_with_binary_scorer,
+    ):
+        # Create multiple score events for different addresses
+        addresses = [Web3.to_checksum_address(f"0x{i:040x}") for i in range(5)]
+        for i, address in enumerate(addresses):
+            Event.objects.create(
+                action=Event.Action.SCORE_UPDATE,
+                address=address,
+                community=scorer_community_with_binary_scorer,
+                data={
+                    "score": i,
+                    "evidence": {
+                        "rawScore": 20 + i,
+                        "type": "binary",
+                        "success": True,
+                        "threshold": 5,
+                    },
+                },
+                created_at=datetime.now() - timedelta(days=i),
+            )
+
+        client = Client()
+        response = client.get(
+            f"{self.base_url}/{scorer_community_with_binary_scorer.id}/score/{scorer_account.address}/history?created_at=2023-01-01T00:00:00",
+            HTTP_AUTHORIZATION="Token " + scorer_api_key,
+        )
+        response_data = response.json()
+
+        assert response.status_code == 200
+        assert len(response_data["items"]) == 0