Backend Live Coding

# ======================================================================
# Q: You have to update 200,000 records. How?
#
# bulk_update/update does not call save() or signals. If there is logic there
# (e.g. cache invalidation), I call it explicitly.
# ======================================================================

# If the new value is computed in the database: a single UPDATE, nothing is loaded into Python
Screen.objects.filter(workspace=ws).update(playlist=new_playlist)

# If the value differs per row: batches, so you don't hold a huge transaction and locks
qs = Screen.objects.filter(needs_migration=True).only('pk', 'settings')
for batch in batched(qs.iterator(chunk_size=2000), 2000):
    for s in batch:
        s.settings = migrate_settings(s.settings)
    with transaction.atomic():
        Screen.objects.bulk_update(batch, ['settings'], batch_size=1000)

# ======================================================================
# Q: For each screen I want its latest event (e.g. the latest error). How do you
# do it without N+1?
#
# With an index on (screen_id, level, created_at DESC) the database finds the
# latest event per screen without scanning the table. For "the last 3 records" I
# use Prefetch with a window function or, on PostgreSQL, a LATERAL join in raw
# SQL.
# ======================================================================

from django.db.models import OuterRef, Subquery, Exists

last_error = (ScreenEvent.objects
    .filter(screen=OuterRef('pk'), level='error')
    .order_by('-created_at'))

screens = Screen.objects.filter(workspace=ws).annotate(
    last_error_msg=Subquery(last_error.values('message')[:1]),
    last_error_at=Subquery(last_error.values('created_at')[:1]),
    has_errors=Exists(last_error),
)

# ======================================================================
# Q: I want, for each playlist, how many screens play it and how many of those
# are online, in a single query.
#
# Watch out: two Count aggregations over different relations in the same
# annotate multiply the rows (JOIN fan-out) and give wrong numbers. In that case
# I use a Subquery or Count(..., distinct=True).
# ======================================================================

from django.db.models import Count, Q

online_since = now() - timedelta(minutes=5)
playlists = (Playlist.objects
    .filter(workspace=ws)
    .annotate(screens_total=Count('screens'),
              screens_online=Count('screens', filter=Q(screens__last_seen__gte=online_since)))
    .order_by('-screens_total'))

# ======================================================================
# Q: Customers upload videos of up to 2 GB. How do you design it?
#
# The file does not pass through Django. The API hands out a presigned URL for a
# direct upload to S3, the browser uploads (multipart for large files), and then
# calls confirm. At that point a Celery task on the heavy queue checks the file
# type, transcodes and creates a thumbnail. Screens download the file from a
# CDN.
# A quota limit per subscription plan, a check that the file really is a
# video/image (not just by its extension), and an S3 lifecycle rule for uploads
# that were never confirmed.
# ======================================================================

def create_upload(request):
    media = Media.objects.create(workspace=request.user.workspace,
                                 status='pending', name=request.data['name'])
    key = f'ws/{media.workspace_id}/media/{media.pk}/{uuid4()}'
    url = s3.generate_presigned_url(
        'put_object', Params={'Bucket': BUCKET, 'Key': key,
                              'ContentType': request.data['content_type']},
        ExpiresIn=900)
    Media.objects.filter(pk=media.pk).update(s3_key=key)
    return Response({'media_id': media.pk, 'upload_url': url}, status=201)

def confirm_upload(request, pk):
    media = get_object_or_404(Media, pk=pk, workspace=request.user.workspace)
    transaction.on_commit(lambda: process_media.delay(media.pk))
    return Response(status=202)                     # Accepted: asynchronous processing

# ======================================================================
# Q: A customer schedules a playlist "every Monday 09:00-12:00" for screens in
# different countries. How do you handle the times?
#
# USE_TZ=True and all timestamps in UTC in the database. But the schedule is the
# screen's local time, so I store the time and day (09:00, Monday) and the
# screen's time zone (Europe/Athens), and compute the UTC at the moment it is
# needed. Otherwise, at the daylight saving change everything shifts by one
# hour.
# ======================================================================

from zoneinfo import ZoneInfo

def is_slot_active(slot, screen, at_utc):
    local = at_utc.astimezone(ZoneInfo(screen.timezone))
    return (local.weekday() in slot.weekdays
            and slot.start_time <= local.time() < slot.end_time)

# Tests with time-machine/freezegun on the days of the clock change (last Sunday of March/October)