TheLovinator/twitch-drop-notifier
1
Fork 0
Code Issues Actions Packages Releases Activity

Add migrations

Browse Source
This commit is contained in:
Joakim Hellsén
2024-12-11 06:01:09 +01:00
parent 42e9a5a7ed
commit daead80fdc
16 changed files with 7897 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
README.md
View File

@ -23,7 +23,8 @@ pip install -r requirements-dev.txt
python -c 'from django.core.management.utils import get_random_secret_key; print(get_random_secret_key())' python -c 'from django.core.management.utils import get_random_secret_key; print(get_random_secret_key())'
# Rename .env.example to .env and fill in the required values. # Rename .env.example to .env and fill in the required values.
# DISCORD_WEBHOOK_URL and EMAIL_* can be left empty. # Only DJANGO_SECRET_KEY is required to run the server.
# EMAIL_HOST_USER, EMAIL_HOST_PASSWORD and DISCORD_WEBHOOK_URL can be left empty if not needed.
mv .env.example .env mv .env.example .env
# Run the migrations. # Run the migrations.

2
core/admin.py
View File

@ -1,3 +1,5 @@
from __future__ import annotations
from django.contrib import admin from django.contrib import admin
from core.models import Benefit, DropCampaign, Game, Owner, TimeBasedDrop from core.models import Benefit, DropCampaign, Game, Owner, TimeBasedDrop

2
core/apps.py
View File

@ -1,3 +1,5 @@
from __future__ import annotations
from django.apps import AppConfig from django.apps import AppConfig

11
core/import_json.py Normal file
View File

@ -0,0 +1,11 @@
from __future__ import annotations
from typing import Any
def import_data_from_view(data: dict[str, Any]) -> None:
"""Import data that are sent from Twitch Drop Miner.
Args:
data (dict[str, Any]): The data to import.
"""

420
core/migrations/0001_initial.py Normal file
View File

@ -0,0 +1,420 @@
# Generated by Django 5.1.4 on 2024-12-11 04:58
from __future__ import annotations
from typing import TYPE_CHECKING
import auto_prefetch
import django.contrib.auth.models
import django.contrib.auth.validators
import django.db.models.deletion
import django.db.models.manager
import django.utils.timezone
from django.db import migrations, models
if TYPE_CHECKING:
from django.db.migrations.operations.base import Operation
class Migration(migrations.Migration):
"""Initial migration for the core app.
This add the following models:
- ScrapedJson
- User
- Owner
- Game
- DropCampaign
- TimeBasedDrop
- Benefit
"""
initial = True
dependencies: list[tuple[str, str]] = [
("auth", "0012_alter_user_first_name_max_length"),
]
operations: list[Operation] = [
migrations.CreateModel(
name="ScrapedJson",
fields=[
("id", models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name="ID")),
("json_data", models.JSONField(help_text="The JSON data from the Twitch API.", unique=True)),
("created_at", models.DateTimeField(auto_now_add=True)),
("modified_at", models.DateTimeField(auto_now=True)),
("imported_at", models.DateTimeField(null=True)),
],
options={
"ordering": ["-created_at"],
"abstract": False,
"base_manager_name": "prefetch_manager",
},
managers=[
("objects", django.db.models.manager.Manager()),
("prefetch_manager", django.db.models.manager.Manager()),
],
),
migrations.CreateModel(
name="User",
fields=[
("id", models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name="ID")),
("password", models.CharField(max_length=128, verbose_name="password")),
("last_login", models.DateTimeField(blank=True, null=True, verbose_name="last login")),
(
"is_superuser",
models.BooleanField(
default=False,
help_text="Designates that this user has all permissions without explicitly assigning them.",
verbose_name="superuser status",
),
),
(
"username",
models.CharField(
error_messages={"unique": "A user with that username already exists."},
help_text="Required. 150 characters or fewer. Letters, digits and @/./+/-/_ only.",
max_length=150,
unique=True,
validators=[django.contrib.auth.validators.UnicodeUsernameValidator()],
verbose_name="username",
),
),
("first_name", models.CharField(blank=True, max_length=150, verbose_name="first name")),
("last_name", models.CharField(blank=True, max_length=150, verbose_name="last name")),
("email", models.EmailField(blank=True, max_length=254, verbose_name="email address")),
(
"is_staff",
models.BooleanField(
default=False,
help_text="Designates whether the user can log into this admin site.",
verbose_name="staff status",
),
),
(
"is_active",
models.BooleanField(
default=True,
help_text=(
"Designates whether this user should be treated as active. Unselect this instead of"
" deleting accounts."
),
verbose_name="active",
),
),
("date_joined", models.DateTimeField(default=django.utils.timezone.now, verbose_name="date joined")),
(
"groups",
models.ManyToManyField(
blank=True,
help_text=(
"The groups this user belongs to. A user will get all permissions granted to each of their"
" groups."
),
related_name="user_set",
related_query_name="user",
to="auth.group",
verbose_name="groups",
),
),
(
"user_permissions",
models.ManyToManyField(
blank=True,
help_text="Specific permissions for this user.",
related_name="user_set",
related_query_name="user",
to="auth.permission",
verbose_name="user permissions",
),
),
],
options={
"ordering": ["username"],
"abstract": False,
"base_manager_name": "prefetch_manager",
},
managers=[
("objects", django.contrib.auth.models.UserManager()),
],
),
migrations.CreateModel(
name="Owner",
fields=[
("created_at", models.DateTimeField(auto_created=True)),
(
"twitch_id",
models.TextField(help_text="The Twitch ID of the owner.", primary_key=True, serialize=False),
),
("modified_at", models.DateTimeField(auto_now=True)),
("name", models.TextField(blank=True, help_text="The name of the owner.")),
],
options={
"ordering": ["name"],
"abstract": False,
"base_manager_name": "prefetch_manager",
"indexes": [
models.Index(fields=["name"], name="owner_name_idx"),
models.Index(fields=["created_at"], name="owner_created_at_idx"),
],
},
managers=[
("objects", django.db.models.manager.Manager()),
("prefetch_manager", django.db.models.manager.Manager()),
],
),
migrations.CreateModel(
name="Game",
fields=[
(
"created_at",
models.DateTimeField(auto_created=True, help_text="When the game was first added to the database."),
),
(
"twitch_id",
models.TextField(help_text="The Twitch ID of the game.", primary_key=True, serialize=False),
),
("modified_at", models.DateTimeField(auto_now=True, help_text="When the game was last modified.")),
("game_url", models.URLField(blank=True, help_text="The URL to the game on Twitch.")),
("display_name", models.TextField(blank=True, help_text="The display name of the game.")),
("name", models.TextField(blank=True, help_text="The name of the game.")),
("box_art_url", models.URLField(blank=True, help_text="URL to the box art of the game.")),
("slug", models.TextField(blank=True)),
(
"org",
auto_prefetch.ForeignKey(
null=True,
on_delete=django.db.models.deletion.CASCADE,
related_name="games",
to="core.owner",
),
),
],
options={
"ordering": ["display_name"],
"abstract": False,
"base_manager_name": "prefetch_manager",
},
managers=[
("objects", django.db.models.manager.Manager()),
("prefetch_manager", django.db.models.manager.Manager()),
],
),
migrations.CreateModel(
name="DropCampaign",
fields=[
(
"created_at",
models.DateTimeField(
auto_created=True,
help_text="When the drop campaign was first added to the database.",
),
),
(
"twitch_id",
models.TextField(
help_text="The Twitch ID of the drop campaign.",
primary_key=True,
serialize=False,
),
),
(
"modified_at",
models.DateTimeField(auto_now=True, help_text="When the drop campaign was last modified."),
),
(
"account_link_url",
models.URLField(blank=True, help_text="The URL to link accounts for the drop campaign."),
),
("description", models.TextField(blank=True, help_text="The description of the drop campaign.")),
("details_url", models.URLField(blank=True, help_text="The URL to the details of the drop campaign.")),
("ends_at", models.DateTimeField(help_text="When the drop campaign ends.", null=True)),
("starts_at", models.DateTimeField(help_text="When the drop campaign starts.", null=True)),
("image_url", models.URLField(blank=True, help_text="The URL to the image for the drop campaign.")),
("name", models.TextField(blank=True, help_text="The name of the drop campaign.")),
("status", models.TextField(blank=True, help_text="The status of the drop campaign.")),
(
"game",
auto_prefetch.ForeignKey(
null=True,
on_delete=django.db.models.deletion.CASCADE,
related_name="drop_campaigns",
to="core.game",
),
),
(
"scraped_json",
auto_prefetch.ForeignKey(
help_text="Reference to the JSON data from the Twitch API.",
null=True,
on_delete=django.db.models.deletion.SET_NULL,
to="core.scrapedjson",
),
),
],
options={
"ordering": ["ends_at"],
"abstract": False,
"base_manager_name": "prefetch_manager",
},
managers=[
("objects", django.db.models.manager.Manager()),
("prefetch_manager", django.db.models.manager.Manager()),
],
),
migrations.CreateModel(
name="TimeBasedDrop",
fields=[
(
"created_at",
models.DateTimeField(auto_created=True, help_text="When the drop was first added to the database."),
),
(
"twitch_id",
models.TextField(help_text="The Twitch ID of the drop.", primary_key=True, serialize=False),
),
("modified_at", models.DateTimeField(auto_now=True, help_text="When the drop was last modified.")),
(
"required_subs",
models.PositiveBigIntegerField(help_text="The number of subs required for the drop.", null=True),
),
("ends_at", models.DateTimeField(help_text="When the drop ends.", null=True)),
("name", models.TextField(blank=True, help_text="The name of the drop.")),
(
"required_minutes_watched",
models.PositiveBigIntegerField(help_text="The number of minutes watched required.", null=True),
),
("starts_at", models.DateTimeField(help_text="When the drop starts.", null=True)),
(
"drop_campaign",
auto_prefetch.ForeignKey(
help_text="The drop campaign this drop is part of.",
null=True,
on_delete=django.db.models.deletion.CASCADE,
related_name="drops",
to="core.dropcampaign",
),
),
],
options={
"ordering": ["required_minutes_watched"],
"abstract": False,
"base_manager_name": "prefetch_manager",
},
managers=[
("objects", django.db.models.manager.Manager()),
("prefetch_manager", django.db.models.manager.Manager()),
],
),
migrations.CreateModel(
name="Benefit",
fields=[
("created_at", models.DateTimeField(auto_created=True, null=True)),
("twitch_id", models.TextField(primary_key=True, serialize=False)),
("modified_at", models.DateTimeField(auto_now=True)),
(
"twitch_created_at",
models.DateTimeField(help_text="When the benefit was created on Twitch.", null=True),
),
(
"entitlement_limit",
models.PositiveBigIntegerField(
help_text="The number of times the benefit can be claimed.",
null=True,
),
),
("image_asset_url", models.URLField(blank=True, help_text="The URL to the image for the benefit.")),
("is_ios_available", models.BooleanField(help_text="If the benefit is farmable on iOS.", null=True)),
("name", models.TextField(blank=True, help_text="The name of the benefit.")),
("distribution_type", models.TextField(blank=True, help_text="The distribution type of the benefit.")),
(
"game",
auto_prefetch.ForeignKey(
null=True,
on_delete=django.db.models.deletion.CASCADE,
related_name="benefits",
to="core.game",
),
),
(
"owner_organization",
auto_prefetch.ForeignKey(
null=True,
on_delete=django.db.models.deletion.CASCADE,
related_name="benefits",
to="core.owner",
),
),
(
"time_based_drop",
auto_prefetch.ForeignKey(
help_text="The time based drop this benefit is for.",
null=True,
on_delete=django.db.models.deletion.CASCADE,
related_name="benefits",
to="core.timebaseddrop",
),
),
],
options={
"ordering": ["-twitch_created_at"],
"abstract": False,
"base_manager_name": "prefetch_manager",
},
managers=[
("objects", django.db.models.manager.Manager()),
("prefetch_manager", django.db.models.manager.Manager()),
],
),
migrations.AddIndex(
model_name="game",
index=models.Index(fields=["display_name"], name="game_display_name_idx"),
),
migrations.AddIndex(
model_name="game",
index=models.Index(fields=["name"], name="game_name_idx"),
),
migrations.AddIndex(
model_name="game",
index=models.Index(fields=["created_at"], name="game_created_at_idx"),
),
migrations.AddIndex(
model_name="dropcampaign",
index=models.Index(fields=["name"], name="drop_campaign_name_idx"),
),
migrations.AddIndex(
model_name="dropcampaign",
index=models.Index(fields=["starts_at"], name="drop_campaign_starts_at_idx"),
),
migrations.AddIndex(
model_name="dropcampaign",
index=models.Index(fields=["ends_at"], name="drop_campaign_ends_at_idx"),
),
migrations.AddIndex(
model_name="timebaseddrop",
index=models.Index(fields=["name"], name="time_based_drop_name_idx"),
),
migrations.AddIndex(
model_name="timebaseddrop",
index=models.Index(fields=["starts_at"], name="time_based_drop_starts_at_idx"),
),
migrations.AddIndex(
model_name="timebaseddrop",
index=models.Index(fields=["ends_at"], name="time_based_drop_ends_at_idx"),
),
migrations.AddIndex(
model_name="benefit",
index=models.Index(fields=["name"], name="benefit_name_idx"),
),
migrations.AddIndex(
model_name="benefit",
index=models.Index(fields=["twitch_created_at"], name="benefit_twitch_created_at_idx"),
),
migrations.AddIndex(
model_name="benefit",
index=models.Index(fields=["created_at"], name="benefit_created_at_idx"),
),
migrations.AddIndex(
model_name="benefit",
index=models.Index(fields=["is_ios_available"], name="benefit_is_ios_available_idx"),
),
]

7
core/models_utils.py
View File

@ -1,8 +1,11 @@
from __future__ import annotations
import logging import logging
from datetime import datetime from datetime import datetime
from typing import Any from typing import TYPE_CHECKING, Any
from django.db import models if TYPE_CHECKING:
from django.db import models
logger: logging.Logger = logging.getLogger(__name__) logger: logging.Logger = logging.getLogger(__name__)

2
core/settings.py
View File

@ -1,3 +1,5 @@
from __future__ import annotations
import os import os
from pathlib import Path from pathlib import Path
from typing import Literal from typing import Literal

2
core/templatetags/custom_filters.py
View File

@ -1,3 +1,5 @@
from __future__ import annotations
from django import template from django import template
register = template.Library() register = template.Library()

2
core/templatetags/time_filters.py
View File

@ -1,3 +1,5 @@
from __future__ import annotations
from django import template from django import template
register = template.Library() register = template.Library()

7396
core/tests/response.json Normal file
View File

File diff suppressed because it is too large Load Diff

4
core/tests/views_test.py
View File

@ -1,13 +1,15 @@
"""Tests for the views in the core app.""" """Tests for the views in the core app."""
from __future__ import annotations
from typing import TYPE_CHECKING from typing import TYPE_CHECKING
import pytest import pytest
from django.http import HttpResponse from django.http import HttpResponse
from django.test import Client
from django.urls import reverse from django.urls import reverse
if TYPE_CHECKING: if TYPE_CHECKING:
from django.test import Client
from django.test.client import _MonkeyPatchedWSGIResponse # type: ignore[import] from django.test.client import _MonkeyPatchedWSGIResponse # type: ignore[import]

9
core/urls.py
View File

@ -4,7 +4,7 @@ from debug_toolbar.toolbar import debug_toolbar_urls # type: ignore[import-unty
from django.contrib import admin from django.contrib import admin
from django.urls import URLPattern, URLResolver, path from django.urls import URLPattern, URLResolver, path
from core.views import game_view, games_view, index from core.views import get_game, get_games, get_home, get_import
app_name: str = "core" app_name: str = "core"
@ -32,8 +32,9 @@ app_name: str = "core"
# The URL patterns for the core app. # The URL patterns for the core app.
urlpatterns: list[URLPattern | URLResolver] = [ urlpatterns: list[URLPattern | URLResolver] = [
path(route="admin/", view=admin.site.urls), path(route="admin/", view=admin.site.urls),
path(route="", view=index, name="index"), path(route="", view=get_home, name="index"),
path(route="game/<int:twitch_id>/", view=game_view, name="game"), path(route="game/<int:twitch_id>/", view=get_game, name="game"),
path(route="games/", view=games_view, name="games"), path(route="games/", view=get_games, name="games"),
path(route="import/", view=get_import, name="import"),
*debug_toolbar_urls(), *debug_toolbar_urls(),
] ]

41
core/views.py
View File

@ -1,13 +1,16 @@
from __future__ import annotations from __future__ import annotations
import json
import logging import logging
from typing import TYPE_CHECKING, Any from typing import TYPE_CHECKING, Any
from django.db.models import F, Prefetch from django.db.models import F, Prefetch
from django.http import HttpRequest, HttpResponse from django.http import HttpRequest, HttpResponse, JsonResponse
from django.template.response import TemplateResponse from django.template.response import TemplateResponse
from django.utils import timezone from django.utils import timezone
from django.views.decorators.http import require_http_methods
from core.import_json import import_data_from_view
from core.models import Benefit, DropCampaign, Game, TimeBasedDrop from core.models import Benefit, DropCampaign, Game, TimeBasedDrop
if TYPE_CHECKING: if TYPE_CHECKING:
@ -47,7 +50,8 @@ def get_games_with_drops() -> QuerySet[Game]:
) )
def index(request: HttpRequest) -> HttpResponse: @require_http_methods(request_method_list=["GET", "HEAD"])
def get_home(request: HttpRequest) -> HttpResponse:
"""Render the index page. """Render the index page.
Args: Args:
@ -58,18 +62,16 @@ def index(request: HttpRequest) -> HttpResponse:
""" """
try: try:
games: QuerySet[Game] = get_games_with_drops() games: QuerySet[Game] = get_games_with_drops()
except Exception: except Exception:
logger.exception("Error fetching reward campaigns or games.") logger.exception("Error fetching reward campaigns or games.")
return HttpResponse(status=500) return HttpResponse(status=500)
context: dict[str, Any] = { context: dict[str, Any] = {"games": games}
"games": games,
}
return TemplateResponse(request, "index.html", context) return TemplateResponse(request, "index.html", context)
def game_view(request: HttpRequest, twitch_id: int) -> HttpResponse: @require_http_methods(request_method_list=["GET", "HEAD"])
def get_game(request: HttpRequest, twitch_id: int) -> HttpResponse:
"""Render the game view page. """Render the game view page.
Args: Args:
@ -101,7 +103,8 @@ def game_view(request: HttpRequest, twitch_id: int) -> HttpResponse:
return TemplateResponse(request=request, template="game.html", context=context) return TemplateResponse(request=request, template="game.html", context=context)
def games_view(request: HttpRequest) -> HttpResponse: @require_http_methods(request_method_list=["GET", "HEAD"])
def get_games(request: HttpRequest) -> HttpResponse:
"""Render the game view page. """Render the game view page.
Args: Args:
@ -114,3 +117,25 @@ def games_view(request: HttpRequest) -> HttpResponse:
context: dict[str, QuerySet[Game] | str] = {"games": games} context: dict[str, QuerySet[Game] | str] = {"games": games}
return TemplateResponse(request=request, template="games.html", context=context) return TemplateResponse(request=request, template="games.html", context=context)
@require_http_methods(request_method_list=["POST"])
def get_import(request: HttpRequest) -> HttpResponse:
"""Import data that are sent from Twitch Drop Miner.
Args:
request (HttpRequest): The request object.
Returns:
HttpResponse: The response object.
"""
try:
data = json.loads(request.body)
logger.info(data)
# Import the data.
import_data_from_view(data)
return JsonResponse({"status": "success"}, status=200)
except json.JSONDecodeError as e:
return JsonResponse({"status": "error", "message": str(e)}, status=400)

9
core/wsgi.py
View File

@ -1,8 +1,13 @@
import os from __future__ import annotations
import os
from typing import TYPE_CHECKING
from django.core.handlers.wsgi import WSGIHandler
from django.core.wsgi import get_wsgi_application from django.core.wsgi import get_wsgi_application
if TYPE_CHECKING:
from django.core.handlers.wsgi import WSGIHandler
os.environ.setdefault(key="DJANGO_SETTINGS_MODULE", value="core.settings") os.environ.setdefault(key="DJANGO_SETTINGS_MODULE", value="core.settings")
application: WSGIHandler = get_wsgi_application() application: WSGIHandler = get_wsgi_application()

2
manage.py
View File

@ -1,6 +1,8 @@
#!/usr/bin/env python #!/usr/bin/env python
"""Django's command-line utility for administrative tasks.""" """Django's command-line utility for administrative tasks."""
from __future__ import annotations
import os import os
import sys import sys

3
pyproject.toml
View File

@ -27,6 +27,9 @@ lint.select = ["ALL"]
# https://sphinxcontrib-napoleon.readthedocs.io/en/latest/example_google.html # https://sphinxcontrib-napoleon.readthedocs.io/en/latest/example_google.html
lint.pydocstyle.convention = "google" lint.pydocstyle.convention = "google"
# Add "from __future__ import annotations" to all files
lint.isort.required-imports = ["from __future__ import annotations"]
# Ignore some rules # Ignore some rules
lint.ignore = [ lint.ignore = [
"CPY001", # Checks for the absence of copyright notices within Python files. "CPY001", # Checks for the absence of copyright notices within Python files.