Chore/typing

django_thumbnail/settings/base.py

@@ -110,3 +110,4 @@
 #    Test external-facing services over the network (not via in-process Django).
 SMOKE_DJANGO_WEB_URL = os.environ.get("SMOKE_DJANGO_WEB_URL", "http://localhost:8000")
 SMOKE_JAEGER_BASE_URL = os.environ.get("SMOKE_JAEGER_BASE_URL", "http://localhost:16686")
+SMOKE_OTEL_SERVICE_NAME = os.environ.get("SMOKE_OTEL_SERVICE_NAME", "django-thumbnail-smoke")

images/decorators.py

@@ -1,12 +1,16 @@
 import json
+from collections.abc import Callable
 from functools import wraps
+from typing import Concatenate
 
-from django.http import HttpRequest, JsonResponse
+from django.http import HttpRequest, HttpResponseBase, JsonResponse
 
 
-def login_required_json(view):
+def login_required_json[**P](
+    view: Callable[Concatenate[HttpRequest, P], HttpResponseBase],
+) -> Callable[Concatenate[HttpRequest, P], HttpResponseBase]:
     @wraps(view)
-    def wrapper(request: HttpRequest, *args, **kwargs):
+    def wrapper(request: HttpRequest, *args: P.args, **kwargs: P.kwargs) -> HttpResponseBase:
         if not request.user.is_authenticated:
             return JsonResponse({"error": "auth required"}, status=401)
         return view(request, *args, **kwargs)

images/management/commands/clean_bucket.py

@@ -1,22 +1,27 @@
+from typing import TYPE_CHECKING
+
 import boto3
 from django.conf import settings
 from django.core.management.base import BaseCommand
 from django.utils.functional import cached_property
 
+if TYPE_CHECKING:
+    from mypy_boto3_s3.client import S3Client
+
 
 class Command(BaseCommand):
     help = "Clear all objects from MinIO bucket"
 
     @cached_property
-    def s3(self):
+    def s3(self) -> S3Client:
         return boto3.client(
             "s3",
             endpoint_url=settings.AWS_S3_ENDPOINT_URL,
             aws_access_key_id=settings.AWS_ACCESS_KEY_ID,
             aws_secret_access_key=settings.AWS_SECRET_ACCESS_KEY,
         )
 
-    def handle(self, *args, **options):
+    def handle(self, *args: object, **options: object) -> None:
         bucket = settings.AWS_STORAGE_BUCKET_NAME
         resp = self.s3.list_objects_v2(Bucket=bucket)
 

images/management/commands/clean_db.py

@@ -6,7 +6,7 @@
 class Command(BaseCommand):
     help = "Clear all images and tasks from database"
 
-    def handle(self, *args, **options):
+    def handle(self, *args: object, **options: object) -> None:
         img_count = Image.objects.count()
         task_count = ImageTask.objects.count()
 

images/management/commands/create_bucket.py

@@ -1,23 +1,28 @@
+from typing import TYPE_CHECKING
+
 import boto3
 from botocore.exceptions import ClientError
 from django.conf import settings
 from django.core.management.base import BaseCommand
 from django.utils.functional import cached_property
 
+if TYPE_CHECKING:
+    from mypy_boto3_s3.client import S3Client
+
 
 class Command(BaseCommand):
     help = "Create MinIO bucket if it does not exist"
 
     @cached_property
-    def s3(self):
+    def s3(self) -> S3Client:
         return boto3.client(
             "s3",
             endpoint_url=settings.AWS_S3_ENDPOINT_URL,
             aws_access_key_id=settings.AWS_ACCESS_KEY_ID,
             aws_secret_access_key=settings.AWS_SECRET_ACCESS_KEY,
         )
 
-    def handle(self, *args, **options):
+    def handle(self, *args: object, **options: object) -> None:
         bucket = settings.AWS_STORAGE_BUCKET_NAME
         try:
             self.s3.head_bucket(Bucket=bucket)

images/management/commands/create_test_users.py

@@ -18,7 +18,7 @@ class TestUser(NamedTuple):
 class Command(BaseCommand):
     help = "Create permanent test accounts"
 
-    def handle(self, *args, **options):
+    def handle(self, *args: object, **options: object) -> None:
         if self.check_created_test_accounts():
             self.stdout.write("test accounts already exist, skipping")
             return

images/management/commands/generate_placeholders.py

@@ -15,7 +15,7 @@
 class Command(BaseCommand):
     help = "Generate placeholder images for upload demo"
 
-    def handle(self, *args, **options):
+    def handle(self, *args: object, **options: object) -> None:
         OUT_DIR.mkdir(parents=True, exist_ok=True)
         for filename, w, h, color in PLACEHOLDERS:
             img = Image.new("RGB", (w, h), color)

images/models.py

@@ -1,5 +1,6 @@
 import io
 import uuid
+from datetime import datetime
 from pathlib import Path
 from typing import TYPE_CHECKING
 
@@ -13,7 +14,7 @@
     from django.db.models.fields.related_descriptors import RelatedManager
 
 
-def _fmt_ts(dt) -> str:
+def _fmt_ts(dt: datetime | None) -> str:
     return dt.strftime("%Y-%m-%d %H:%M:%S") if dt else "?"
 
 

images/storage.py

@@ -10,7 +10,7 @@
 
 
 class _S3BaseClient:
-    def __init__(self, endpoint_url: str, config: Config | None = None):
+    def __init__(self, endpoint_url: str, config: Config | None = None) -> None:
         self._client: S3Client = boto3.client(
             "s3",
             endpoint_url=endpoint_url,
@@ -21,7 +21,7 @@ def __init__(self, endpoint_url: str, config: Config | None = None):
 
 
 class InternalS3(_S3BaseClient):
-    def __init__(self):
+    def __init__(self) -> None:
         super().__init__(endpoint_url=settings.AWS_S3_ENDPOINT_URL)
 
     def download(self, bucket: str, key: str) -> io.BytesIO:
@@ -36,7 +36,7 @@ def delete(self, bucket: str, key: str) -> None:
 
 
 class PublicS3(_S3BaseClient):
-    def __init__(self):
+    def __init__(self) -> None:
         super().__init__(
             endpoint_url=settings.AWS_S3_PUBLIC_ENDPOINT_URL,
             config=Config(signature_version=settings.AWS_S3_PUBLIC_SIGNATURE_VERSION),

images/tasks.py

@@ -1,7 +1,7 @@
 import io
 from enum import StrEnum
 
-from celery import shared_task
+from celery import Task, shared_task
 from django.conf import settings
 from opentelemetry import trace
 from PIL import Image as PillowImage
@@ -10,6 +10,7 @@
 from .storage import internal_s3
 
 THUMBNAIL_SIZE = (20, 20)
+GENERATE_THUMBNAIL_MAX_RETRIES = 3
 
 
 class InvalidImageError(Exception):
@@ -43,8 +44,8 @@ def _make_thumbnail(buf: io.BytesIO) -> io.BytesIO:
     return out
 
 
-@shared_task(bind=True, max_retries=3)
-def generate_thumbnail(self, image_task_id: str) -> None:
+@shared_task(bind=True, max_retries=GENERATE_THUMBNAIL_MAX_RETRIES)
+def generate_thumbnail(self: Task, image_task_id: str) -> None:
     task = ImageTask.objects.select_related("image").get(id=image_task_id)
     celery_id = self.request.id or ""
     task.mark_processing(celery_id)

images/tests/conftest.py

@@ -1,19 +1,21 @@
 import pytest
+from django.contrib.auth.models import User
+from django.test import Client
 
-from .factories import UserFactory
+from .factories import make_user
 
 
 @pytest.fixture
-def user(db):
-    return UserFactory()
+def user(db: None) -> User:
+    return make_user()
 
 
 @pytest.fixture
-def other_user(db):
-    return UserFactory()
+def other_user(db: None) -> User:
+    return make_user()
 
 
 @pytest.fixture
-def auth_client(client, user):
+def auth_client(client: Client, user: User) -> tuple[Client, User]:
     client.force_login(user)
     return client, user

images/tests/factories.py

@@ -1,4 +1,5 @@
 import uuid
+from typing import cast
 
 import factory
 from django.contrib.auth.models import User
@@ -17,11 +18,11 @@ class Meta:
     email = factory.LazyAttribute(lambda o: o.username)
 
     @factory.post_generation
-    def password(self, create, extracted, **kwargs):
-        pwd = extracted or USER_PLAIN_PASSWORD
-        self.set_password(pwd)
+    def password(self, create: bool, extracted: str | None, **kwargs: dict) -> None:
+        user = cast(User, self)
+        user.set_password(extracted or USER_PLAIN_PASSWORD)
         if create:
-            self.save(update_fields=["password"])
+            user.save(update_fields=["password"])
 
 
 class ImageFactory(factory.django.DjangoModelFactory):
@@ -44,3 +45,15 @@ class Meta:
 
     image = factory.SubFactory(ImageFactory)
     status = ImageStatus.PENDING
+
+
+def make_user(**kwargs: object) -> User:
+    return cast(User, UserFactory(**kwargs))
+
+
+def make_image(**kwargs: object) -> Image:
+    return cast(Image, ImageFactory(**kwargs))
+
+
+def make_image_task(**kwargs: object) -> ImageTask:
+    return cast(ImageTask, ImageTaskFactory(**kwargs))

images/tests/jaeger.py

@@ -0,0 +1,113 @@
+"""Jaeger HTTP API client + typed response schemas. Test-only."""
+
+import json
+import time
+import urllib.request
+from typing import NotRequired, TypedDict
+
+from django.conf import settings
+
+JaegerValue = str | int | float | bool
+
+_POLL_MAX = 2
+_POLL_INTERVAL = 3
+_REQUEST_TIMEOUT = 5
+
+
+class KeyedValue(TypedDict):
+    key: str
+    type: str
+    value: JaegerValue
+
+
+class Process(TypedDict):
+    serviceName: str
+    tags: list[KeyedValue]
+
+
+class SpanLog(TypedDict):
+    timestamp: int
+    fields: list[KeyedValue]
+
+
+class SpanReference(TypedDict):
+    refType: str
+    traceID: str
+    spanID: str
+
+
+class Span(TypedDict):
+    traceID: str
+    spanID: str
+    operationName: str
+    references: list[SpanReference]
+    startTime: int
+    duration: int
+    tags: list[KeyedValue]
+    logs: list[SpanLog]
+    processID: str
+    warnings: NotRequired[list[str] | None]
+
+
+class TraceData(TypedDict):
+    traceID: str
+    spans: list[Span]
+    processes: dict[str, Process]
+    warnings: NotRequired[list[str] | None]
+
+
+class JaegerResponse(TypedDict):
+    data: list[TraceData]
+    total: int
+    limit: int
+    offset: int
+    errors: list[object] | None
+
+
+_EMPTY_RESPONSE: JaegerResponse = JaegerResponse(data=[], total=0, limit=0, offset=0, errors=None)
+
+
+class JaegerClient:
+    def __init__(self, base_url: str) -> None:
+        self._base_url = base_url
+
+    def fetch_trace(self, trace_id: str) -> JaegerResponse:
+        url = f"{self._base_url}/api/traces/{trace_id}"
+        with urllib.request.urlopen(url, timeout=_REQUEST_TIMEOUT) as resp:
+            return json.loads(resp.read())
+
+    def poll_trace(self, trace_id: str) -> JaegerResponse:
+        data = _EMPTY_RESPONSE
+        for _ in range(_POLL_MAX):
+            data = self.fetch_trace(trace_id)
+            if data.get("errors") is None and data.get("data"):
+                return data
+            time.sleep(_POLL_INTERVAL)
+        return data
+
+    @staticmethod
+    def extract_events(data: JaegerResponse, operation_name: str | None = None) -> list[str]:
+        """Pull `event` field values from Jaeger logs across spans in the trace.
+
+        operation_name=None scans every span (useful when the span name is
+        owned by an auto-instrumentor whose naming may shift between releases).
+        """
+        events: list[str] = []
+        for trace_obj in data.get("data", []):
+            for span in trace_obj.get("spans", []):
+                if operation_name is not None and span.get("operationName") != operation_name:
+                    continue
+                for log in span.get("logs", []):
+                    for field in log.get("fields", []):
+                        if field.get("key") == "event":
+                            events.append(str(field.get("value")))
+        return events
+
+    @staticmethod
+    def services_in_trace(data: JaegerResponse) -> set[str]:
+        if not data.get("data"):
+            return set()
+        return {p["serviceName"] for p in data["data"][0]["processes"].values()}
+
+
+jaeger_client = JaegerClient(base_url=settings.SMOKE_JAEGER_BASE_URL)