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

Refactor ChannelDetailView to use get_object_or_404 and optimize campaign fetching logic

Browse source
This commit is contained in:
Joakim Hellsén 2026-04-12 04:47:38 +02:00
commit 3070dcb296
Signed by: Joakim Hellsén
SSH key fingerprint: SHA256:/9h/CsExpFp+PRhsfA0xznFx2CGfTT5R/kpuFfUgEQk
3 changed files with 361 additions and 73 deletions
Download patch file Download diff file Expand all files Collapse all files

126
twitch/models.py
View file

@ -1,3 +1,4 @@
import datetime
import logging
from collections import OrderedDict
from typing import TYPE_CHECKING
@ -18,8 +19,6 @@ from django.utils.html import format_html
from twitch.utils import normalize_twitch_box_art_url
if TYPE_CHECKING:
import datetime
from django.db.models import QuerySet
logger: logging.Logger = logging.getLogger("ttvdrops")
@ -404,6 +403,27 @@ class Channel(auto_prefetch.Model):
campaign_count=F("allowed_campaign_count"),
).order_by("-campaign_count", "name")
@classmethod
def for_detail_view(cls) -> models.QuerySet[Channel]:
"""Return channels with only fields needed by the channel detail view."""
return cls.objects.only(
"twitch_id",
"name",
"display_name",
"added_at",
"updated_at",
)
@property
def preferred_name(self) -> str:
"""Return display name fallback used by channel-facing pages."""
return self.display_name or self.name or self.twitch_id
def detail_description(self, total_campaigns: int) -> str:
"""Return a short channel-detail description with pluralization."""
suffix: str = "s" if total_campaigns != 1 else ""
return f"{self.preferred_name} participates in {total_campaigns} drop campaign{suffix}"
# MARK: DropCampaign
class DropCampaign(auto_prefetch.Model):
@ -684,6 +704,8 @@ class DropCampaign(auto_prefetch.Model):
"distribution_type",
"image_asset_url",
"image_file",
"image_width",
"image_height",
),
),
)
@ -693,6 +715,98 @@ class DropCampaign(auto_prefetch.Model):
.get(twitch_id=twitch_id)
)
@classmethod
def for_channel_detail(cls, channel: Channel) -> models.QuerySet[DropCampaign]:
"""Return campaigns with only channel-detail-required relations/fields.
Args:
channel: Channel used for allow-list filtering.
Returns:
QuerySet ordered by newest start date.
"""
return (
cls.objects
.filter(allow_channels=channel)
.select_related("game")
.only(
"twitch_id",
"name",
"start_at",
"end_at",
"game",
"game__twitch_id",
"game__name",
"game__display_name",
)
.prefetch_related(
Prefetch(
"time_based_drops",
queryset=TimeBasedDrop.objects.only(
"twitch_id",
"campaign_id",
).prefetch_related(
Prefetch(
"benefits",
queryset=DropBenefit.objects.only(
"twitch_id",
"name",
"image_asset_url",
"image_file",
"image_width",
"image_height",
).order_by("name"),
),
),
),
)
.order_by("-start_at")
)
@staticmethod
def split_for_channel_detail(
campaigns: list[DropCampaign],
now: datetime.datetime,
) -> tuple[list[DropCampaign], list[DropCampaign], list[DropCampaign]]:
"""Split channel campaigns into active, upcoming, and expired buckets.
Args:
campaigns: List of campaigns to split.
now: Current datetime for comparison.
Returns:
Tuple containing lists of active, upcoming, and expired campaigns.
"""
sentinel: datetime.datetime = datetime.datetime.max.replace(
tzinfo=datetime.UTC,
)
active_campaigns: list[DropCampaign] = sorted(
[
campaign
for campaign in campaigns
if campaign.start_at is not None
and campaign.start_at <= now
and campaign.end_at is not None
and campaign.end_at >= now
],
key=lambda campaign: campaign.end_at or sentinel,
)
upcoming_campaigns: list[DropCampaign] = sorted(
[
campaign
for campaign in campaigns
if campaign.start_at is not None and campaign.start_at > now
],
key=lambda campaign: campaign.start_at or sentinel,
)
expired_campaigns: list[DropCampaign] = [
campaign
for campaign in campaigns
if campaign.end_at is not None and campaign.end_at < now
]
return active_campaigns, upcoming_campaigns, expired_campaigns
@staticmethod
def _countdown_text_for_drop(
drop: TimeBasedDrop,
@ -1137,8 +1251,12 @@ class DropBenefit(auto_prefetch.Model):
def image_best_url(self) -> str:
"""Return the best URL for the benefit image (local first)."""
try:
if self.image_file and getattr(self.image_file, "url", None):
return self.image_file.url
if self.image_file:
file_name: str = getattr(self.image_file, "name", "")
if file_name and self.image_file.storage.exists(file_name):
file_url: str | None = getattr(self.image_file, "url", None)
if file_url:
return file_url
except (AttributeError, OSError, ValueError) as exc:
logger.debug("Failed to resolve DropBenefit.image_file url: %s", exc)
return self.image_asset_url or ""

229
twitch/tests/test_views.py
View file

@ -525,6 +525,205 @@ class TestChannelListView:
assert "GROUP BY" not in sql
assert "ALLOWED_CAMPAIGN_COUNT" in sql
def test_channel_detail_queryset_only_selects_rendered_fields(self) -> None:
"""Channel detail queryset should defer fields not used by the template/SEO."""
channel: Channel = Channel.objects.create(
twitch_id="channel_detail_fields",
name="channeldetailfields",
display_name="Channel Detail Fields",
)
fetched_channel: Channel | None = (
Channel
.for_detail_view()
.filter(
twitch_id=channel.twitch_id,
)
.first()
)
assert fetched_channel is not None
deferred_fields: set[str] = fetched_channel.get_deferred_fields()
assert "allowed_campaign_count" in deferred_fields
assert "name" not in deferred_fields
assert "display_name" not in deferred_fields
assert "twitch_id" not in deferred_fields
assert "added_at" not in deferred_fields
assert "updated_at" not in deferred_fields
def test_channel_detail_campaign_queryset_only_selects_rendered_fields(
self,
) -> None:
"""Channel detail campaign queryset should avoid loading unused campaign fields."""
now: datetime.datetime = timezone.now()
game: Game = Game.objects.create(
twitch_id="channel_detail_game_fields",
name="Channel Detail Game Fields",
display_name="Channel Detail Game Fields",
)
channel: Channel = Channel.objects.create(
twitch_id="channel_detail_campaign_fields",
name="channeldetailcampaignfields",
display_name="Channel Detail Campaign Fields",
)
campaign: DropCampaign = DropCampaign.objects.create(
twitch_id="channel_detail_campaign",
name="Channel Detail Campaign",
game=game,
operation_names=["DropCampaignDetails"],
start_at=now - timedelta(hours=1),
end_at=now + timedelta(hours=1),
)
campaign.allow_channels.add(channel)
fetched_campaign: DropCampaign | None = DropCampaign.for_channel_detail(
channel,
).first()
assert fetched_campaign is not None
deferred_fields: set[str] = fetched_campaign.get_deferred_fields()
assert "description" in deferred_fields
assert "details_url" in deferred_fields
assert "account_link_url" in deferred_fields
assert "name" not in deferred_fields
assert "start_at" not in deferred_fields
assert "end_at" not in deferred_fields
def test_channel_detail_prefetch_avoids_dropbenefit_refresh_n_plus_one(
self,
) -> None:
"""Channel detail prefetch should not refresh each DropBenefit row for image dimensions."""
now: datetime.datetime = timezone.now()
game: Game = Game.objects.create(
twitch_id="channel_detail_n_plus_one_game",
name="Channel Detail N+1 Game",
display_name="Channel Detail N+1 Game",
)
channel: Channel = Channel.objects.create(
twitch_id="channel_detail_n_plus_one_channel",
name="channeldetailnplusone",
display_name="Channel Detail N+1",
)
campaign: DropCampaign = DropCampaign.objects.create(
twitch_id="channel_detail_n_plus_one_campaign",
name="Channel Detail N+1 Campaign",
game=game,
operation_names=["DropCampaignDetails"],
start_at=now - timedelta(hours=1),
end_at=now + timedelta(hours=1),
)
campaign.allow_channels.add(channel)
drop: TimeBasedDrop = TimeBasedDrop.objects.create(
twitch_id="channel_detail_n_plus_one_drop",
name="Channel Detail N+1 Drop",
campaign=campaign,
)
png_1x1: bytes = (
b"\x89PNG\r\n\x1a\n\x00\x00\x00\rIHDR\x00\x00\x00\x01"
b"\x00\x00\x00\x01\x08\x06\x00\x00\x00\x1f\x15\xc4\x89"
b"\x00\x00\x00\x0bIDATx\x9cc\x00\x01\x00\x00\x05\x00\x01"
b"\r\n-\xb4\x00\x00\x00\x00IEND\xaeB`\x82"
)
benefits: list[DropBenefit] = []
for i in range(3):
benefit: DropBenefit = DropBenefit.objects.create(
twitch_id=f"channel_detail_n_plus_one_benefit_{i}",
name=f"Benefit {i}",
image_asset_url=f"https://example.com/benefit_{i}.png",
)
assert benefit.image_file is not None
benefit.image_file.save(
f"channel_detail_n_plus_one_benefit_{i}.png",
ContentFile(png_1x1),
save=True,
)
benefits.append(benefit)
drop.benefits.add(*benefits)
with CaptureQueriesContext(connection) as queries:
campaigns: list[DropCampaign] = list(
DropCampaign.for_channel_detail(channel),
)
assert campaigns
_ = [
benefit.name
for campaign_row in campaigns
for drop_row in campaign_row.time_based_drops.all() # pyright: ignore[reportAttributeAccessIssue]
for benefit in drop_row.benefits.all()
]
refresh_queries: list[str] = [
query_info["sql"]
for query_info in queries.captured_queries
if query_info["sql"].lstrip().upper().startswith("SELECT")
and 'from "twitch_dropbenefit"' in query_info["sql"].lower()
and 'where "twitch_dropbenefit"."id" =' in query_info["sql"].lower()
]
assert not refresh_queries, (
"Channel detail queryset triggered per-benefit refresh SELECTs. "
f"Queries: {refresh_queries}"
)
def test_channel_detail_uses_asset_url_when_local_benefit_file_is_missing(
self,
client: Client,
) -> None:
"""Channel detail should avoid broken local image URLs when cached files are missing."""
now: datetime.datetime = timezone.now()
game: Game = Game.objects.create(
twitch_id="channel_detail_missing_local_file_game",
name="Channel Detail Missing Local File Game",
display_name="Channel Detail Missing Local File Game",
)
channel: Channel = Channel.objects.create(
twitch_id="channel_detail_missing_local_file_channel",
name="missinglocalfilechannel",
display_name="Missing Local File Channel",
)
campaign: DropCampaign = DropCampaign.objects.create(
twitch_id="channel_detail_missing_local_file_campaign",
name="Channel Detail Missing Local File Campaign",
game=game,
operation_names=["DropCampaignDetails"],
start_at=now - timedelta(hours=1),
end_at=now + timedelta(hours=1),
)
campaign.allow_channels.add(channel)
drop: TimeBasedDrop = TimeBasedDrop.objects.create(
twitch_id="channel_detail_missing_local_file_drop",
name="Channel Detail Missing Local File Drop",
campaign=campaign,
)
remote_asset_url: str = "https://example.com/benefit-missing-local-file.png"
benefit: DropBenefit = DropBenefit.objects.create(
twitch_id="channel_detail_missing_local_file_benefit",
name="Benefit Missing Local File",
image_asset_url=remote_asset_url,
)
DropBenefit.objects.filter(pk=benefit.pk).update(
image_file="benefits/images/does-not-exist.png",
)
drop.benefits.add(benefit)
response: _MonkeyPatchedWSGIResponse = client.get(
reverse("twitch:channel_detail", args=[channel.twitch_id]),
)
assert response.status_code == 200
html: str = response.content.decode("utf-8")
assert remote_asset_url in html
assert "benefits/images/does-not-exist.png" not in html
def test_channel_allowed_campaign_count_updates_on_add_remove_clear(self) -> None:
"""Counter cache should stay in sync when campaign-channel links change."""
game: Game = Game.objects.create(
@ -928,6 +1127,36 @@ class TestChannelListView:
f"Expected one of {sorted(expected_reward_indexes)}. Plan={reward_plan}"
)
@pytest.mark.django_db
def test_dashboard_active_window_composite_indexes_exist(self) -> None:
"""Dashboard active-window filters should have supporting composite indexes."""
with connection.cursor() as cursor:
drop_constraints = connection.introspection.get_constraints(
cursor,
DropCampaign._meta.db_table,
)
reward_constraints = connection.introspection.get_constraints(
cursor,
RewardCampaign._meta.db_table,
)
def _index_columns(constraints: dict[str, Any]) -> list[tuple[str, ...]]:
columns: list[tuple[str, ...]] = []
for meta in constraints.values():
if not meta.get("index"):
continue
index_columns: list[str] = meta.get("columns") or []
columns.append(tuple(index_columns))
return columns
drop_index_columns: list[tuple[str, ...]] = _index_columns(drop_constraints)
reward_index_columns: list[tuple[str, ...]] = _index_columns(
reward_constraints,
)
assert ("start_at", "end_at") in drop_index_columns
assert ("starts_at", "ends_at") in reward_index_columns
@pytest.mark.django_db
def test_dashboard_query_count_stays_flat_with_more_data(
self,

81
twitch/views.py
View file

@ -21,6 +21,7 @@ from django.db.models import Q
from django.db.models.query import QuerySet
from django.http import Http404
from django.http import HttpResponse
from django.shortcuts import get_object_or_404
from django.shortcuts import render
from django.urls import reverse
from django.utils import timezone
@ -1320,20 +1321,10 @@ class ChannelDetailView(DetailView):
Returns:
Channel: The channel object.
Raises:
Http404: If the channel is not found.
"""
if queryset is None:
queryset = self.get_queryset()
twitch_id: str | None = self.kwargs.get("twitch_id")
try:
channel: Channel = queryset.get(twitch_id=twitch_id)
except Channel.DoesNotExist as exc:
msg = "No channel found matching the query"
raise Http404(msg) from exc
queryset = queryset or Channel.for_detail_view()
twitch_id: str = str(self.kwargs.get("twitch_id", ""))
channel: Channel = get_object_or_404(queryset, twitch_id=twitch_id)
return channel
def get_context_data(self, **kwargs) -> dict[str, Any]: # noqa: PLR0914
@ -1349,66 +1340,16 @@ class ChannelDetailView(DetailView):
channel: Channel = self.object # pyright: ignore[reportAssignmentType]
now: datetime.datetime = timezone.now()
all_campaigns: QuerySet[DropCampaign] = (
DropCampaign.objects
.filter(allow_channels=channel)
.select_related("game")
.prefetch_related(
Prefetch(
"time_based_drops",
queryset=TimeBasedDrop.objects.prefetch_related(
Prefetch(
"benefits",
queryset=DropBenefit.objects.order_by("name"),
),
),
),
)
.order_by("-start_at")
campaigns_list: list[DropCampaign] = list(
DropCampaign.for_channel_detail(channel),
)
active_campaigns, upcoming_campaigns, expired_campaigns = (
DropCampaign.split_for_channel_detail(campaigns_list, now)
)
campaigns_list: list[DropCampaign] = list(all_campaigns)
active_campaigns: list[DropCampaign] = [
campaign
for campaign in campaigns_list
if campaign.start_at is not None
and campaign.start_at <= now
and campaign.end_at is not None
and campaign.end_at >= now
]
active_campaigns.sort(
key=lambda c: (
c.end_at
if c.end_at is not None
else datetime.datetime.max.replace(tzinfo=datetime.UTC)
),
)
upcoming_campaigns: list[DropCampaign] = [
campaign
for campaign in campaigns_list
if campaign.start_at is not None and campaign.start_at > now
]
upcoming_campaigns.sort(
key=lambda c: (
c.start_at
if c.start_at is not None
else datetime.datetime.max.replace(tzinfo=datetime.UTC)
),
)
expired_campaigns: list[DropCampaign] = [
campaign
for campaign in campaigns_list
if campaign.end_at is not None and campaign.end_at < now
]
name: str = channel.display_name or channel.name or channel.twitch_id
name: str = channel.preferred_name
total_campaigns: int = len(campaigns_list)
description: str = f"{name} participates in {total_campaigns} drop campaign"
if total_campaigns > 1:
description += "s"
description: str = channel.detail_description(total_campaigns)
channel_url: str = build_absolute_uri(
reverse("twitch:channel_detail", args=[channel.twitch_id]),