TheLovinator/ttvdrops
1
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Enhance DropCampaign detail view and optimize badge queries

Browse source
This commit is contained in:
Joakim Hellsén 2026-04-12 03:52:04 +02:00
commit 917bf8ac23
Signed by: Joakim Hellsén
SSH key fingerprint: SHA256:/9h/CsExpFp+PRhsfA0xznFx2CGfTT5R/kpuFfUgEQk
3 changed files with 240 additions and 74 deletions
Download patch file Download diff file Expand all files Collapse all files

164
twitch/models.py
View file

@ -571,6 +571,170 @@ class DropCampaign(auto_prefetch.Model):
queryset = queryset.filter(end_at__lt=now)
return queryset
@classmethod
def for_detail_view(cls, twitch_id: str) -> DropCampaign:
"""Return a campaign with only detail-view-required relations/fields loaded.
Args:
twitch_id: Campaign Twitch ID.
Returns:
Campaign object with game, owners, channels, drops, and benefits preloaded.
"""
return (
cls.objects
.select_related("game")
.only(
"twitch_id",
"name",
"description",
"details_url",
"account_link_url",
"image_url",
"image_file",
"image_width",
"image_height",
"start_at",
"end_at",
"added_at",
"updated_at",
"game__twitch_id",
"game__name",
"game__display_name",
"game__slug",
)
.prefetch_related(
Prefetch(
"game__owners",
queryset=Organization.objects.only("twitch_id", "name").order_by(
"name",
),
to_attr="owners_for_detail",
),
Prefetch(
"allow_channels",
queryset=Channel.objects.only(
"twitch_id",
"name",
"display_name",
).order_by("display_name"),
to_attr="channels_ordered",
),
Prefetch(
"time_based_drops",
queryset=TimeBasedDrop.objects
.only(
"twitch_id",
"name",
"required_minutes_watched",
"required_subs",
"start_at",
"end_at",
"campaign_id",
)
.prefetch_related(
Prefetch(
"benefits",
queryset=DropBenefit.objects.only(
"twitch_id",
"name",
"distribution_type",
"image_asset_url",
"image_file",
),
),
)
.order_by("required_minutes_watched"),
),
)
.get(twitch_id=twitch_id)
)
@staticmethod
def _countdown_text_for_drop(
drop: TimeBasedDrop,
now: datetime.datetime,
) -> str:
"""Return a display countdown for a detail-view drop row."""
if drop.end_at and drop.end_at > now:
time_diff: datetime.timedelta = drop.end_at - now
days: int = time_diff.days
hours, remainder = divmod(time_diff.seconds, 3600)
minutes, seconds = divmod(remainder, 60)
if days > 0:
return f"{days}d {hours}h {minutes}m"
if hours > 0:
return f"{hours}h {minutes}m"
if minutes > 0:
return f"{minutes}m {seconds}s"
return f"{seconds}s"
if drop.start_at and drop.start_at > now:
return "Not started"
return "Expired"
def awarded_badges_by_drop_twitch_id(
self,
) -> dict[str, ChatBadge]:
"""Return the first awarded badge per drop keyed by drop Twitch ID."""
drops: list[TimeBasedDrop] = list(self.time_based_drops.all()) # pyright: ignore[reportAttributeAccessIssue]
badge_titles: set[str] = {
benefit.name
for drop in drops
for benefit in drop.benefits.all()
if benefit.distribution_type == "BADGE" and benefit.name
}
if not badge_titles:
return {}
badges_by_title: dict[str, ChatBadge] = {
badge.title: badge
for badge in (
ChatBadge.objects
.select_related("badge_set")
.only(
"title",
"description",
"image_url_2x",
"badge_set__set_id",
)
.filter(title__in=badge_titles)
)
}
awarded_badges: dict[str, ChatBadge] = {}
for drop in drops:
for benefit in drop.benefits.all():
if benefit.distribution_type != "BADGE":
continue
badge: ChatBadge | None = badges_by_title.get(benefit.name)
if badge:
awarded_badges[drop.twitch_id] = badge
break
return awarded_badges
def enhanced_drops_for_detail(
self,
now: datetime.datetime,
) -> list[dict[str, Any]]:
"""Return campaign drops with detail-view presentation metadata."""
drops: list[TimeBasedDrop] = list(self.time_based_drops.all()) # pyright: ignore[reportAttributeAccessIssue]
awarded_badges: dict[str, ChatBadge] = self.awarded_badges_by_drop_twitch_id()
return [
{
"drop": drop,
"local_start": drop.start_at,
"local_end": drop.end_at,
"timezone_name": "UTC",
"countdown_text": self._countdown_text_for_drop(drop, now),
"awarded_badge": awarded_badges.get(drop.twitch_id),
}
for drop in drops
]
@classmethod
def active_for_dashboard(
cls,

71
twitch/tests/test_views.py
View file

@ -642,10 +642,12 @@ class TestChannelListView:
campaigns_uses_index = (
"INDEX SCAN" in campaigns_plan.upper()
or "BITMAP INDEX SCAN" in campaigns_plan.upper()
or "INDEX ONLY SCAN" in campaigns_plan.upper()
)
rewards_uses_index = (
"INDEX SCAN" in reward_plan.upper()
or "BITMAP INDEX SCAN" in reward_plan.upper()
or "INDEX ONLY SCAN" in reward_plan.upper()
)
else:
pytest.skip(
@ -1735,6 +1737,75 @@ class TestChannelListView:
html = response.content.decode("utf-8")
assert "This badge was earned by subscribing." in html
@pytest.mark.django_db
def test_drop_campaign_detail_badge_queries_stay_flat(self, client: Client) -> None:
"""Campaign detail should avoid N+1 ChatBadge lookups across many badge drops."""
now: datetime.datetime = timezone.now()
game: Game = Game.objects.create(
twitch_id="g-badge-flat",
name="Game",
display_name="Game",
)
campaign: DropCampaign = DropCampaign.objects.create(
twitch_id="c-badge-flat",
name="Campaign",
game=game,
operation_names=["DropCampaignDetails"],
start_at=now - timedelta(hours=1),
end_at=now + timedelta(hours=1),
)
badge_set = ChatBadgeSet.objects.create(set_id="badge-flat")
def _create_badge_drop(i: int) -> None:
drop = TimeBasedDrop.objects.create(
twitch_id=f"flat-drop-{i}",
name=f"Drop {i}",
campaign=campaign,
required_minutes_watched=i,
required_subs=0,
start_at=now - timedelta(hours=2),
end_at=now + timedelta(hours=2),
)
title = f"Badge {i}"
benefit = DropBenefit.objects.create(
twitch_id=f"flat-benefit-{i}",
name=title,
distribution_type="BADGE",
)
drop.benefits.add(benefit)
ChatBadge.objects.create(
badge_set=badge_set,
badge_id=str(i),
image_url_1x=f"https://example.com/{i}/1x.png",
image_url_2x=f"https://example.com/{i}/2x.png",
image_url_4x=f"https://example.com/{i}/4x.png",
title=title,
description=f"Badge description {i}",
)
def _select_count() -> int:
url: str = reverse("twitch:campaign_detail", args=[campaign.twitch_id])
with CaptureQueriesContext(connection) as capture:
response: _MonkeyPatchedWSGIResponse = client.get(url)
assert response.status_code == 200
return sum(
1
for query in capture.captured_queries
if query["sql"].lstrip().upper().startswith("SELECT")
)
_create_badge_drop(1)
baseline_selects: int = _select_count()
for i in range(2, 22):
_create_badge_drop(i)
expanded_selects: int = _select_count()
# Query volume should remain effectively constant as badge-drop count grows.
assert expanded_selects <= baseline_selects + 2
@pytest.mark.django_db
def test_games_grid_view(self, client: Client) -> None:
"""Test games grid view returns 200 and has games in context."""

79
twitch/views.py
View file

@ -515,48 +515,6 @@ def drop_campaign_list_view(request: HttpRequest) -> HttpResponse: # noqa: PLR0
return render(request, "twitch/campaign_list.html", context)
def _enhance_drops_with_context(
drops: QuerySet[TimeBasedDrop],
now: datetime.datetime,
) -> list[dict[str, Any]]:
"""Helper to enhance drops with countdown and context.
Args:
drops: QuerySet of TimeBasedDrop objects.
now: Current datetime.
Returns:
List of dicts with drop and additional context for display.
"""
enhanced: list[dict[str, Any]] = []
for drop in drops:
if drop.end_at and drop.end_at > now:
time_diff: datetime.timedelta = drop.end_at - now
days: int = time_diff.days
hours, remainder = divmod(time_diff.seconds, 3600)
minutes, seconds = divmod(remainder, 60)
if days > 0:
countdown_text: str = f"{days}d {hours}h {minutes}m"
elif hours > 0:
countdown_text = f"{hours}h {minutes}m"
elif minutes > 0:
countdown_text = f"{minutes}m {seconds}s"
else:
countdown_text = f"{seconds}s"
elif drop.start_at and drop.start_at > now:
countdown_text = "Not started"
else:
countdown_text = "Expired"
enhanced.append({
"drop": drop,
"local_start": drop.start_at,
"local_end": drop.end_at,
"timezone_name": "UTC",
"countdown_text": countdown_text,
})
return enhanced
# MARK: /campaigns/<twitch_id>/
def drop_campaign_detail_view(request: HttpRequest, twitch_id: str) -> HttpResponse: # noqa: PLR0914
"""Function-based view for a drop campaign detail.
@ -572,45 +530,20 @@ def drop_campaign_detail_view(request: HttpRequest, twitch_id: str) -> HttpRespo
Http404: If the campaign is not found.
"""
try:
campaign: DropCampaign = DropCampaign.objects.prefetch_related(
"game__owners",
Prefetch(
"allow_channels",
queryset=Channel.objects.order_by("display_name"),
to_attr="channels_ordered",
),
Prefetch(
"time_based_drops",
queryset=TimeBasedDrop.objects.prefetch_related("benefits").order_by(
"required_minutes_watched",
),
),
).get(twitch_id=twitch_id)
campaign: DropCampaign = DropCampaign.for_detail_view(twitch_id)
except DropCampaign.DoesNotExist as exc:
msg = "No campaign found matching the query"
raise Http404(msg) from exc
drops: QuerySet[TimeBasedDrop] = campaign.time_based_drops.all() # pyright: ignore[reportAttributeAccessIssue]
now: datetime.datetime = timezone.now()
enhanced_drops: list[dict[str, Any]] = _enhance_drops_with_context(drops, now)
# Attach awarded_badge to each drop in enhanced_drops
for enhanced_drop in enhanced_drops:
drop = enhanced_drop["drop"]
awarded_badge = None
for benefit in drop.benefits.all():
if benefit.distribution_type == "BADGE":
awarded_badge: ChatBadge | None = ChatBadge.objects.filter(
title=benefit.name,
).first()
break
enhanced_drop["awarded_badge"] = awarded_badge
owners: list[Organization] = list(getattr(campaign.game, "owners_for_detail", []))
enhanced_drops: list[dict[str, Any]] = campaign.enhanced_drops_for_detail(now)
context: dict[str, Any] = {
"campaign": campaign,
"now": now,
"drops": enhanced_drops,
"owners": list(campaign.game.owners.all()),
"owners": owners,
"allowed_channels": getattr(campaign, "channels_ordered", []),
}
@ -650,9 +583,7 @@ def drop_campaign_detail_view(request: HttpRequest, twitch_id: str) -> HttpRespo
campaign_event["startDate"] = campaign.start_at.isoformat()
if campaign.end_at:
campaign_event["endDate"] = campaign.end_at.isoformat()
campaign_owner: Organization | None = (
_pick_owner(list(campaign.game.owners.all())) if campaign.game else None
)
campaign_owner: Organization | None = _pick_owner(owners) if owners else None
campaign_owner_name: str = (
(campaign_owner.name or campaign_owner.twitch_id)
if campaign_owner