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

Implement _save_if_changed method to optimize model updates and add tests to ensure unchanged data does not trigger updates

Browse source
This commit is contained in:
Joakim Hellsén 2026-03-17 20:03:27 +01:00
commit cd9bb0a95a
Signed by: Joakim Hellsén
SSH key fingerprint: SHA256:/9h/CsExpFp+PRhsfA0xznFx2CGfTT5R/kpuFfUgEQk
3 changed files with 273 additions and 34 deletions
Download patch file Download diff file Expand all files Collapse all files

121
twitch/management/commands/better_import_drops.py
View file

@ -36,6 +36,7 @@ from twitch.utils import parse_date
if TYPE_CHECKING:
from django.core.management.base import CommandParser
from django.db.models import Model
from json_repair import JSONReturnType
from twitch.schemas import ChannelInfoSchema
@ -544,6 +545,32 @@ class Command(BaseCommand):
return valid_responses, broken_dir
def _save_if_changed(self, obj: Model, defaults: dict[str, object]) -> bool:
"""Save the model instance only when data actually changed.
This prevents unnecessary updates and avoids touching fields like
`updated_at` when the imported values are identical.
Args:
obj: The model instance to potentially update.
defaults: Field values to apply.
Returns:
True if the object was saved, False if no changes were detected.
"""
changed_fields: list[str] = []
for field, new_value in defaults.items():
if getattr(obj, field, None) != new_value:
setattr(obj, field, new_value)
changed_fields.append(field)
if not changed_fields:
return False
obj.save(update_fields=changed_fields)
return True
def _get_or_create_organization(self, org_data: OrganizationSchema) -> Organization:
"""Get or create an organization.
@ -553,11 +580,13 @@ class Command(BaseCommand):
Returns:
Organization instance.
"""
org_obj, created = Organization.objects.update_or_create(
org_obj, created = Organization.objects.get_or_create(
twitch_id=org_data.twitch_id,
defaults={"name": org_data.name},
)
if created:
if not created:
self._save_if_changed(org_obj, {"name": org_data.name})
else:
tqdm.write(
f"{Fore.GREEN}✓{Style.RESET_ALL} Created new organization: {org_data.name}",
)
@ -589,19 +618,23 @@ class Command(BaseCommand):
if campaign_org_obj:
owner_orgs.add(campaign_org_obj)
game_obj, created = Game.objects.update_or_create(
defaults: dict[str, str] = {
"display_name": game_data.display_name or (game_data.name or ""),
"name": game_data.name or "",
"slug": game_data.slug or "",
"box_art": game_data.box_art_url or "",
}
game_obj, created = Game.objects.get_or_create(
twitch_id=game_data.twitch_id,
defaults={
"display_name": game_data.display_name or (game_data.name or ""),
"name": game_data.name or "",
"slug": game_data.slug or "",
"box_art": game_data.box_art_url or "",
},
defaults=defaults,
)
# Set owners (ManyToMany)
if created or owner_orgs:
game_obj.owners.add(*owner_orgs)
if created:
if not created:
self._save_if_changed(game_obj, defaults)
else:
tqdm.write(
f"{Fore.GREEN}✓{Style.RESET_ALL} Created new game: {game_data.display_name}",
)
@ -648,11 +681,17 @@ class Command(BaseCommand):
# Use name as display_name fallback if displayName is None
display_name: str = channel_info.display_name or channel_info.name
channel_obj, created = Channel.objects.update_or_create(
defaults: dict[str, str] = {
"name": channel_info.name,
"display_name": display_name,
}
channel_obj, created = Channel.objects.get_or_create(
twitch_id=channel_info.twitch_id,
defaults={"name": channel_info.name, "display_name": display_name},
defaults=defaults,
)
if created:
if not created:
self._save_if_changed(channel_obj, defaults)
else:
tqdm.write(
f"{Fore.GREEN}✓{Style.RESET_ALL} Created new channel: {display_name}",
)
@ -750,11 +789,13 @@ class Command(BaseCommand):
"account_link_url": drop_campaign.account_link_url,
}
campaign_obj, created = DropCampaign.objects.update_or_create(
campaign_obj, created = DropCampaign.objects.get_or_create(
twitch_id=drop_campaign.twitch_id,
defaults=defaults,
)
if created:
if not created:
self._save_if_changed(campaign_obj, defaults)
else:
tqdm.write(
f"{Fore.GREEN}✓{Style.RESET_ALL} Created new campaign: {drop_campaign.name}",
)
@ -829,11 +870,13 @@ class Command(BaseCommand):
if end_at_dt is not None:
drop_defaults["end_at"] = end_at_dt
drop_obj, created = TimeBasedDrop.objects.update_or_create(
drop_obj, created = TimeBasedDrop.objects.get_or_create(
twitch_id=drop_schema.twitch_id,
defaults=drop_defaults,
)
if created:
if not created:
self._save_if_changed(drop_obj, drop_defaults)
else:
tqdm.write(
f"{Fore.GREEN}✓{Style.RESET_ALL} Created TimeBasedDrop: {drop_schema.name}",
)
@ -859,11 +902,13 @@ class Command(BaseCommand):
if created_at_dt:
benefit_defaults["created_at"] = created_at_dt
benefit_obj, created = DropBenefit.objects.update_or_create(
benefit_obj, created = DropBenefit.objects.get_or_create(
twitch_id=benefit_schema.twitch_id,
defaults=benefit_defaults,
)
if created:
if not created:
self._save_if_changed(benefit_obj, benefit_defaults)
else:
tqdm.write(
f"{Fore.GREEN}✓{Style.RESET_ALL} Created DropBenefit: {benefit_schema.name}",
)
@ -888,12 +933,15 @@ class Command(BaseCommand):
benefit_schema=benefit_schema,
)
_edge_obj, created = DropBenefitEdge.objects.update_or_create(
defaults = {"entitlement_limit": edge_schema.entitlement_limit}
edge_obj, created = DropBenefitEdge.objects.get_or_create(
drop=drop_obj,
benefit=benefit_obj,
defaults={"entitlement_limit": edge_schema.entitlement_limit},
defaults=defaults,
)
if created:
if not created:
self._save_if_changed(edge_obj, defaults)
else:
tqdm.write(
f"{Fore.GREEN}✓{Style.RESET_ALL} Linked benefit: {benefit_schema.name} → {drop_obj.name}",
)
@ -996,22 +1044,27 @@ class Command(BaseCommand):
else "",
}
_reward_campaign_obj, created = RewardCampaign.objects.update_or_create(
reward_obj, created = RewardCampaign.objects.get_or_create(
twitch_id=reward_campaign.twitch_id,
defaults=defaults,
)
action: Literal["Imported new", "Updated"] = (
"Imported new" if created else "Updated"
)
display_name = (
f"{reward_campaign.brand}: {reward_campaign.name}"
if reward_campaign.brand
else reward_campaign.name
)
tqdm.write(
f"{Fore.GREEN}✓{Style.RESET_ALL} {action} reward campaign: {display_name}",
)
updated: bool = False
if not created:
updated = self._save_if_changed(reward_obj, defaults)
if created or updated:
action: Literal["Imported new", "Updated"] = (
"Imported new" if created else "Updated"
)
display_name = (
f"{reward_campaign.brand}: {reward_campaign.name}"
if reward_campaign.brand
else reward_campaign.name
)
tqdm.write(
f"{Fore.GREEN}✓{Style.RESET_ALL} {action} reward campaign: {display_name}",
)
def handle(self, *args, **options) -> None: # noqa: ARG002
"""Main entry point for the command.

75
twitch/tests/test_better_import_drops.py
View file

@ -165,6 +165,81 @@ class ExtractCampaignsTests(TestCase):
assert campaign.name == "Test Inventory Campaign"
assert campaign.operation_names == ["Inventory"]
def test_import_does_not_update_campaign_when_data_unchanged(self) -> None:
"""Ensure repeated imports do not modify the campaign updated_at."""
command = Command()
payload: dict[str, object] = {
"data": {
"currentUser": {
"id": "17658559",
"inventory": {
"dropCampaignsInProgress": [
{
"id": "inventory-campaign-1",
"name": "Test Inventory Campaign",
"description": "Campaign from Inventory operation",
"startAt": "2025-01-01T00:00:00Z",
"endAt": "2025-12-31T23:59:59Z",
"accountLinkURL": "https://example.com/link",
"detailsURL": "https://example.com/details",
"imageURL": "https://example.com/campaign.png",
"status": "ACTIVE",
"self": {
"isAccountConnected": True,
"__typename": "DropCampaignSelfEdge",
},
"game": {
"id": "inventory-game-1",
"displayName": "Inventory Game",
"boxArtURL": "https://example.com/boxart.png",
"slug": "inventory-game",
"name": "Inventory Game",
"__typename": "Game",
},
"owner": {
"id": "inventory-org-1",
"name": "Inventory Organization",
"__typename": "Organization",
},
"timeBasedDrops": [],
"eventBasedDrops": None,
"__typename": "DropCampaign",
},
],
"gameEventDrops": None,
"__typename": "Inventory",
},
"__typename": "User",
},
},
"extensions": {"operationName": "Inventory"},
}
# First import to create the campaign
success, _ = command.process_responses(
responses=[payload],
file_path=Path("test_inventory.json"),
options={},
)
assert success is True
campaign: DropCampaign = DropCampaign.objects.get(
twitch_id="inventory-campaign-1",
)
updated_at = campaign.updated_at
# Second import should not change updated_at since data is identical
success, _ = command.process_responses(
responses=[payload],
file_path=Path("test_inventory.json"),
options={},
)
assert success is True
campaign.refresh_from_db()
assert campaign.updated_at == updated_at
def test_handles_inventory_with_null_campaigns(self) -> None:
"""Ensure Inventory JSON with null dropCampaignsInProgress is handled correctly."""
command = Command()

111
twitch/tests/test_views.py
View file

@ -1,6 +1,7 @@
import datetime
import json
from datetime import timedelta
from pathlib import Path
from typing import TYPE_CHECKING
from typing import Any
from typing import Literal
@ -8,6 +9,7 @@ from typing import Literal
import pytest
from django.core.handlers.wsgi import WSGIRequest
from django.core.paginator import Paginator
from django.db.models import Max
from django.test import RequestFactory
from django.urls import reverse
from django.utils import timezone
@ -15,6 +17,7 @@ from django.utils import timezone
from kick.models import KickCategory
from kick.models import KickDropCampaign
from kick.models import KickOrganization
from twitch.management.commands.better_import_drops import Command
from twitch.models import Channel
from twitch.models import ChatBadge
from twitch.models import ChatBadgeSet
@ -22,6 +25,7 @@ from twitch.models import DropBenefit
from twitch.models import DropCampaign
from twitch.models import Game
from twitch.models import Organization
from twitch.models import RewardCampaign
from twitch.models import TimeBasedDrop
from twitch.views import _build_breadcrumb_schema
from twitch.views import _build_pagination_info
@ -1421,6 +1425,113 @@ class TestSitemapView:
# Ensure at least one entry includes a lastmod (there are entities created by the fixture)
assert "<lastmod>" in content
def test_import_does_not_update_lastmod_on_repeated_imports(
self,
client: Client,
sample_entities: dict[str, Any],
) -> None:
"""Ensure repeated imports do not change sitemap lastmod timestamps."""
command = Command()
payload: dict[str, object] = {
"data": {
"currentUser": {
"id": "17658559",
"inventory": {
"dropCampaignsInProgress": [
{
"id": "inventory-campaign-1",
"name": "Test Inventory Campaign",
"description": "Campaign from Inventory operation",
"startAt": "2025-01-01T00:00:00Z",
"endAt": "2025-12-31T23:59:59Z",
"accountLinkURL": "https://example.com/link",
"detailsURL": "https://example.com/details",
"imageURL": "https://example.com/campaign.png",
"status": "ACTIVE",
"self": {
"isAccountConnected": True,
"__typename": "DropCampaignSelfEdge",
},
"game": {
"id": "inventory-game-1",
"displayName": "Inventory Game",
"boxArtURL": "https://example.com/boxart.png",
"slug": "inventory-game",
"name": "Inventory Game",
"__typename": "Game",
},
"owner": {
"id": "inventory-org-1",
"name": "Inventory Organization",
"__typename": "Organization",
},
"timeBasedDrops": [],
"eventBasedDrops": None,
"__typename": "DropCampaign",
},
],
"gameEventDrops": None,
"__typename": "Inventory",
},
"__typename": "User",
},
},
"extensions": {"operationName": "Inventory"},
}
def _lastmod_values() -> tuple[
datetime.datetime | None,
datetime.datetime | None,
]:
twitch_drops_lastmod = max(
[
dt
for dt in [
DropCampaign.objects.aggregate(max=Max("updated_at"))["max"],
RewardCampaign.objects.aggregate(max=Max("updated_at"))["max"],
]
if dt is not None
],
default=None,
)
twitch_others_lastmod = max(
[
dt
for dt in [
Game.objects.aggregate(max=Max("updated_at"))["max"],
Organization.objects.aggregate(max=Max("updated_at"))["max"],
ChatBadgeSet.objects.aggregate(max=Max("updated_at"))["max"],
]
if dt is not None
],
default=None,
)
return twitch_drops_lastmod, twitch_others_lastmod
# Initial import
success, _ = command.process_responses(
responses=[payload],
file_path=Path("test_inventory.json"),
options={},
)
assert success is True
first_drops, first_others = _lastmod_values()
# Second import should not change lastmod values for related models
success, _ = command.process_responses(
responses=[payload],
file_path=Path("test_inventory.json"),
options={},
)
assert success is True
second_drops, second_others = _lastmod_values()
assert first_drops == second_drops
assert first_others == second_others
def test_sitemap_contains_static_pages(
self,
client: Client,