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

Allow Inventory

Browse source
This commit is contained in:
Joakim Hellsén 2026-01-05 19:34:31 +01:00
commit f06db7e47e
No known key found for this signature in database
5 changed files with 430 additions and 22 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -505,10 +505,23 @@ class Command(BaseCommand):
if not response.data.current_user:
continue
if not response.data.current_user.drop_campaigns:
continue
for drop_campaign in response.data.current_user.drop_campaigns:
org_obj: Organization = self._get_or_create_organization(
org_data=drop_campaign.owner,
)
# Handle campaigns without owner (e.g., from Inventory operation)
if drop_campaign.owner:
org_obj: Organization = self._get_or_create_organization(
org_data=drop_campaign.owner,
)
else:
# Create a default organization for campaigns without owner
org_obj, _ = Organization.objects.get_or_create(
twitch_id="unknown",
defaults={"name": "Unknown Organization"},
)
self.organization_cache["unknown"] = org_obj
game_obj: Game = self._get_or_create_game(
game_data=drop_campaign.game,
org_obj=org_obj,

116
twitch/schemas.py
View file

@ -5,6 +5,7 @@ from typing import Literal
from pydantic import BaseModel
from pydantic import Field
from pydantic import field_validator
from pydantic import model_validator
class OrganizationSchema(BaseModel):
@ -23,12 +24,17 @@ class OrganizationSchema(BaseModel):
class GameSchema(BaseModel):
"""Schema for Twitch Game objects."""
"""Schema for Twitch Game objects.
twitch_id: str = Field(alias="id")
display_name: str = Field(alias="displayName")
box_art_url: str = Field(alias="boxArtURL")
type_name: Literal["Game"] = Field(alias="__typename")
Handles both ViewerDropsDashboard and Inventory operation formats.
"""
twitch_id: str = Field(alias="id") # Present in both ViewerDropsDashboard and Inventory formats
display_name: str | None = Field(default=None, alias="displayName") # Present in both formats
box_art_url: str = Field(alias="boxArtURL") # Present in both formats
slug: str | None = None # Present in Inventory format
name: str | None = None # Present in Inventory format (alternative to displayName)
type_name: Literal["Game"] = Field(alias="__typename") # Present in both formats
model_config = {
"extra": "forbid",
@ -37,6 +43,24 @@ class GameSchema(BaseModel):
"populate_by_name": True,
}
@model_validator(mode="before")
@classmethod
def normalize_display_name(cls, data: dict | object) -> dict | object:
"""Normalize display_name from 'name' field if needed.
Inventory format uses 'name' instead of 'displayName'.
Args:
data: The raw input data dictionary.
Returns:
Normalized data dictionary.
"""
if isinstance(data, dict) and "displayName" not in data and "name" in data:
data["displayName"] = data["name"]
return data
class DropCampaignSelfEdge(BaseModel):
"""Schema for the 'self' edge on DropCampaign objects."""
@ -53,16 +77,19 @@ class DropCampaignSelfEdge(BaseModel):
class DropBenefitSchema(BaseModel):
"""Schema for a benefit in a DropBenefitEdge."""
"""Schema for a benefit in a DropBenefitEdge.
Handles both ViewerDropsDashboard and Inventory operation formats.
"""
twitch_id: str = Field(alias="id")
name: str
image_asset_url: str = Field(alias="imageAssetURL")
created_at: str | None = Field(alias="createdAt")
entitlement_limit: int = Field(alias="entitlementLimit")
is_ios_available: bool = Field(alias="isIosAvailable")
created_at: str | None = Field(default=None, alias="createdAt")
entitlement_limit: int = Field(default=1, alias="entitlementLimit")
is_ios_available: bool = Field(default=False, alias="isIosAvailable")
distribution_type: str = Field(alias="distributionType")
type_name: Literal["Benefit"] = Field(alias="__typename")
type_name: Literal["Benefit", "DropBenefit"] = Field(alias="__typename")
model_config = {
"extra": "forbid",
@ -73,10 +100,15 @@ class DropBenefitSchema(BaseModel):
class DropBenefitEdgeSchema(BaseModel):
"""Schema for a benefit edge in a TimeBasedDrop."""
"""Schema for a benefit edge in a TimeBasedDrop.
Handles both ViewerDropsDashboard and Inventory operation formats.
"""
benefit: DropBenefitSchema
entitlement_limit: int = Field(alias="entitlementLimit")
claim_count: int | None = Field(default=None, alias="claimCount")
type_name: Literal["DropBenefitEdge"] | None = Field(default=None, alias="__typename")
model_config = {
"extra": "forbid",
@ -87,7 +119,10 @@ class DropBenefitEdgeSchema(BaseModel):
class TimeBasedDropSchema(BaseModel):
"""Schema for a TimeBasedDrop in a DropCampaign."""
"""Schema for a TimeBasedDrop in a DropCampaign.
Handles both ViewerDropsDashboard and Inventory operation formats.
"""
twitch_id: str = Field(alias="id")
name: str
@ -97,6 +132,11 @@ class TimeBasedDropSchema(BaseModel):
end_at: str | None = Field(alias="endAt")
benefit_edges: list[DropBenefitEdgeSchema] = Field(alias="benefitEdges")
type_name: Literal["TimeBasedDrop"] = Field(alias="__typename")
# Inventory-specific fields
precondition_drops: None = Field(default=None, alias="preconditionDrops")
self_edge: dict | None = Field(default=None, alias="self")
campaign: dict | None = None
localized_content: dict | None = Field(default=None, alias="localizedContent")
model_config = {
"extra": "forbid",
@ -107,11 +147,14 @@ class TimeBasedDropSchema(BaseModel):
class DropCampaign(BaseModel):
"""Schema for Twitch DropCampaign objects."""
"""Schema for Twitch DropCampaign objects.
Handles both ViewerDropsDashboard and Inventory operation formats.
"""
twitch_id: str = Field(alias="id")
name: str
owner: OrganizationSchema
owner: OrganizationSchema | None = None
game: GameSchema
status: Literal["ACTIVE", "EXPIRED", "UPCOMING"]
start_at: str = Field(alias="startAt")
@ -121,6 +164,26 @@ class DropCampaign(BaseModel):
self: DropCampaignSelfEdge
time_based_drops: list[TimeBasedDropSchema] = Field(default=[], alias="timeBasedDrops")
type_name: Literal["DropCampaign"] = Field(alias="__typename")
# Inventory-specific fields
image_url: str | None = Field(default=None, alias="imageURL")
allow: dict | None = None
event_based_drops: list | None = Field(default=None, alias="eventBasedDrops")
model_config = {
"extra": "forbid",
"validate_assignment": True,
"strict": True,
"populate_by_name": True,
}
class InventorySchema(BaseModel):
"""Schema for the inventory field in Inventory operation responses."""
drop_campaigns_in_progress: list[DropCampaign] = Field(alias="dropCampaignsInProgress")
type_name: Literal["Inventory"] = Field(alias="__typename")
# gameEventDrops field is present in Inventory but we don't process it yet
game_event_drops: list | None = Field(default=None, alias="gameEventDrops")
model_config = {
"extra": "forbid",
@ -131,11 +194,15 @@ class DropCampaign(BaseModel):
class CurrentUser(BaseModel):
"""Schema for Twitch User objects."""
"""Schema for Twitch User objects.
Handles both ViewerDropsDashboard and Inventory operation formats.
"""
twitch_id: str = Field(alias="id")
login: str
drop_campaigns: list[DropCampaign] = Field(alias="dropCampaigns")
login: str | None = None
drop_campaigns: list[DropCampaign] | None = Field(default=None, alias="dropCampaigns")
inventory: InventorySchema | None = None
type_name: Literal["User"] = Field(alias="__typename")
model_config = {
@ -145,6 +212,21 @@ class CurrentUser(BaseModel):
"populate_by_name": True,
}
@model_validator(mode="after")
def normalize_from_inventory(self) -> CurrentUser:
"""Normalize drop_campaigns from inventory if present.
If drop_campaigns is None but inventory exists, extract
dropCampaignsInProgress and assign it to drop_campaigns.
Returns:
The CurrentUser instance with normalized drop_campaigns.
"""
if self.drop_campaigns is None and self.inventory is not None:
self.drop_campaigns = self.inventory.drop_campaigns_in_progress
return self
class Data(BaseModel):
"""Schema for the data field in Twitch API responses."""

247
twitch/tests/test_schemas.py Normal file
View file

@ -0,0 +1,247 @@
"""Tests for Pydantic schemas used in the import process."""
from __future__ import annotations
from twitch.schemas import GraphQLResponse
def test_inventory_operation_validation() -> None:
"""Test that the Inventory operation format validates correctly.
The Inventory operation has a different structure than ViewerDropsDashboard:
- No 'login' field in currentUser
- No 'dropCampaigns' field in currentUser
- Has 'inventory.dropCampaignsInProgress' instead
- Campaign data has extra fields that are now explicitly defined
- Game uses 'name' instead of 'displayName'
- Benefits have 'DropBenefit' as __typename instead of 'Benefit'
"""
# Minimal valid Inventory operation structure
payload = {
"data": {
"currentUser": {
"id": "17658559",
"inventory": {
"__typename": "Inventory",
"dropCampaignsInProgress": [
{
"id": "campaign-1",
"name": "Test Campaign",
"game": {
"id": "game-1",
"name": "Test Game",
"boxArtURL": "https://example.com/boxart.jpg",
"__typename": "Game",
},
"status": "ACTIVE",
"startAt": "2025-01-01T00:00:00Z",
"endAt": "2025-12-31T23:59:59Z",
"detailsURL": "https://example.com/details",
"accountLinkURL": "https://example.com/link",
"self": {
"isAccountConnected": True,
"__typename": "DropCampaignSelfEdge",
},
"imageURL": "https://example.com/image.png",
"allow": None,
"eventBasedDrops": [],
"timeBasedDrops": [
{
"id": "drop-1",
"name": "Test Drop",
"requiredMinutesWatched": 60,
"requiredSubs": 0,
"startAt": "2025-01-01T00:00:00Z",
"endAt": "2025-12-31T23:59:59Z",
"benefitEdges": [
{
"benefit": {
"id": "benefit-1",
"name": "Test Benefit",
"imageAssetURL": "https://example.com/benefit.png",
"entitlementLimit": 1,
"isIosAvailable": False,
"distributionType": "DIRECT_ENTITLEMENT",
"__typename": "DropBenefit",
},
"entitlementLimit": 1,
"claimCount": 0,
"__typename": "DropBenefitEdge",
},
],
"preconditionDrops": None,
"self": {},
"campaign": {},
"localizedContent": {},
"__typename": "TimeBasedDrop",
},
],
"__typename": "DropCampaign",
},
],
"gameEventDrops": [],
},
"__typename": "User",
},
},
"extensions": {
"operationName": "Inventory",
},
}
# This should not raise ValidationError
response = GraphQLResponse.model_validate(payload)
# Verify the structure was parsed correctly
assert response.data.current_user is not None
assert response.data.current_user.twitch_id == "17658559"
# Verify drop_campaigns was normalized from inventory
assert response.data.current_user.drop_campaigns is not None
assert len(response.data.current_user.drop_campaigns) == 1
campaign = response.data.current_user.drop_campaigns[0]
assert campaign.name == "Test Campaign"
assert campaign.game.display_name == "Test Game"
assert len(campaign.time_based_drops) == 1
# Verify time-based drops
first_drop = campaign.time_based_drops[0]
assert first_drop.name == "Test Drop"
assert first_drop.required_minutes_watched == 60
# Verify benefits
assert len(first_drop.benefit_edges) == 1
assert first_drop.benefit_edges[0].benefit.name == "Test Benefit"
def test_viewer_drops_dashboard_operation_still_works() -> None:
"""Test that the original ViewerDropsDashboard format still validates.
This ensures backward compatibility with the existing import data.
"""
# Minimal valid ViewerDropsDashboard structure
payload = {
"data": {
"currentUser": {
"id": "12345",
"login": "testuser",
"dropCampaigns": [
{
"id": "campaign-1",
"name": "Test Campaign",
"owner": {
"id": "org-1",
"name": "Test Org",
"__typename": "Organization",
},
"game": {
"id": "game-1",
"displayName": "Test Game",
"boxArtURL": "https://example.com/boxart.jpg",
"__typename": "Game",
},
"status": "ACTIVE",
"startAt": "2025-01-01T00:00:00Z",
"endAt": "2025-12-31T23:59:59Z",
"detailsURL": "https://example.com/details",
"accountLinkURL": "https://example.com/link",
"self": {
"isAccountConnected": True,
"__typename": "DropCampaignSelfEdge",
},
"timeBasedDrops": [],
"__typename": "DropCampaign",
},
],
"__typename": "User",
},
},
"extensions": {
"operationName": "ViewerDropsDashboard",
},
}
# This should not raise ValidationError
response = GraphQLResponse.model_validate(payload)
assert response.data.current_user is not None
assert response.data.current_user.login == "testuser"
if response.data.current_user.drop_campaigns is not None:
assert len(response.data.current_user.drop_campaigns) == 1
def test_graphql_response_with_errors() -> None:
"""Test that GraphQL responses with errors field validate correctly.
Some API responses include partial data with errors (e.g., service timeouts).
The schema should accept these responses so they can be processed.
"""
# Real-world example: Inventory operation with service timeout errors
payload = {
"errors": [
{
"message": "service timeout",
"path": ["currentUser", "inventory", "dropCampaignsInProgress", 7, "allow", "channels"],
},
{
"message": "service timeout",
"path": ["currentUser", "inventory", "dropCampaignsInProgress", 10, "allow", "channels"],
},
],
"data": {
"currentUser": {
"id": "17658559",
"inventory": {
"__typename": "Inventory",
"dropCampaignsInProgress": [
{
"id": "campaign-1",
"name": "Test Campaign",
"game": {
"id": "game-1",
"name": "Test Game",
"boxArtURL": "https://example.com/boxart.jpg",
"__typename": "Game",
},
"status": "ACTIVE",
"startAt": "2025-01-01T00:00:00Z",
"endAt": "2025-12-31T23:59:59Z",
"detailsURL": "https://example.com/details",
"accountLinkURL": "https://example.com/link",
"self": {
"isAccountConnected": True,
"__typename": "DropCampaignSelfEdge",
},
"imageURL": "https://example.com/image.png",
"allow": None,
"eventBasedDrops": [],
"timeBasedDrops": [],
"__typename": "DropCampaign",
},
],
"gameEventDrops": [],
},
"__typename": "User",
},
},
"extensions": {
"operationName": "Inventory",
},
}
# This should not raise ValidationError even with errors field present
response = GraphQLResponse.model_validate(payload)
# Verify the errors were captured
assert response.errors is not None
assert len(response.errors) == 2
assert response.errors[0].message == "service timeout"
assert response.errors[0].path == ["currentUser", "inventory", "dropCampaignsInProgress", 7, "allow", "channels"]
# Verify the data is still accessible and valid
assert response.data.current_user is not None
assert response.data.current_user.twitch_id == "17658559"
assert response.data.current_user.drop_campaigns is not None
assert len(response.data.current_user.drop_campaigns) == 1

2
twitch/tests/test_views.py
View file

@ -5,8 +5,6 @@ from typing import Any
from typing import Literal
import pytest
from django.test.client import _MonkeyPatchedWSGIResponse
from django.test.utils import ContextList
from twitch.models import DropBenefit
from twitch.models import DropCampaign