feat(scores): only return best score per user

This commit is contained in:
Loïc Gremaud 2026-06-20 08:25:28 +02:00
parent ec58b74db5
commit fcb995dbd4
Signed by: Legrems
GPG Key ID: D4620E6DF3E0121D
4 changed files with 71 additions and 16 deletions

View File

@ -1,7 +1,7 @@
from app.api_base import BaseSchema
from ninja.pagination import PageNumberPagination as _PageNumberPagination
from app.api_base import BaseSchema
class PageNumberPagination(_PageNumberPagination):
class Input(_PageNumberPagination.Input, BaseSchema):

View File

@ -8,8 +8,8 @@ from ninja.pagination import paginate
from app.api_base import CamelCaseRouter
from scores.models import Score, Version
from .schemas import ScoreIn, ScoreOut
from .pagination import PageNumberPagination
from .schemas import ScoreIn, ScoreOut
router = CamelCaseRouter()
@ -42,4 +42,4 @@ def list_scores(
x_game_version: GameVersionHeader = None,
) -> QuerySet[Score]:
version = resolve_version(x_game_version)
return Score.objects.filter(version=version)
return Score.objects.filter(version=version).best_per_username().with_rank()

View File

@ -1,7 +1,7 @@
from django.db import models
from django.db.models import F
from django.db.models.expressions import Window
from django.db.models.functions import Rank
from django.db.models.functions import Lower, Rank, RowNumber
class Version(models.Model):
@ -12,19 +12,37 @@ class Version(models.Model):
return self.name
class ScoreManager(models.Manager):
def get_queryset(self) -> models.QuerySet["Score"]:
return (
super()
.get_queryset()
.select_related("version")
.annotate(
rank=Window(
expression=Rank(),
order_by=F("points").desc(),
class ScoreQuerySet(models.QuerySet):
def best_per_username(self) -> "ScoreQuerySet":
"""Keep only each user's highest score, matching usernames case-insensitively.
Lower scores are excluded from the result but remain in the database.
"""
best_ids = (
self.alias(
_row=Window(
expression=RowNumber(),
partition_by=Lower("username"),
order_by=[F("points").desc(), F("created_at").asc()],
)
)
.filter(_row=1)
.values_list("pk", flat=True)
)
return self.filter(pk__in=best_ids)
def with_rank(self) -> "ScoreQuerySet":
return self.annotate(
rank=Window(
expression=Rank(),
order_by=F("points").desc(),
)
)
class ScoreManager(models.Manager.from_queryset(ScoreQuerySet)):
def get_queryset(self) -> ScoreQuerySet:
return super().get_queryset().select_related("version")
class Score(models.Model):

View File

@ -1 +1,38 @@
# Create your tests here.
from django.test import TestCase
from scores.models import Score, Version
class BestPerUsernameTests(TestCase):
@classmethod
def setUpTestData(cls) -> None:
cls.version = Version.objects.create(name="default")
cls.other = Version.objects.create(name="other")
Score.objects.create(username="legrems", points=120, version=cls.version)
Score.objects.create(username="Legrems", points=119, version=cls.version)
Score.objects.create(username="legrems", points=50, version=cls.version)
Score.objects.create(username="alice", points=200, version=cls.version)
# Same user, different version: must not collide across versions.
Score.objects.create(username="legrems", points=999, version=cls.other)
def _best(self) -> list[Score]:
return list(Score.objects.filter(version=self.version).best_per_username().with_rank())
def test_keeps_only_highest_per_case_insensitive_username(self) -> None:
results = self._best()
self.assertEqual(len(results), 2)
by_name = {s.username.lower(): s for s in results}
self.assertEqual(by_name["legrems"].points, 120)
self.assertEqual(by_name["alice"].points, 200)
def test_ranks_recomputed_over_deduplicated_set(self) -> None:
results = self._best()
ranks = {s.username.lower(): s.rank for s in results}
self.assertEqual(ranks["alice"], 1)
self.assertEqual(ranks["legrems"], 2)
def test_lower_scores_remain_in_db(self) -> None:
self.assertEqual(Score.objects.filter(version=self.version).count(), 4)