wimby/tubesync
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Migrate some tasks to huey

Browse Source
This commit is contained in:
tcely authored and GitHub committed 2025-06-12 16:21:29 -04:00
1 parent 54ee219ccb
commit f455350e13
1 file changed
+32 -39
Download Patch File Download Diff File Expand all files Collapse all files
tubesync/sync/tasks.py
+32 -39
View File
@@ -25,6 +25,8 @@ from django.db.transaction import atomic
from django.utils import timezone
from django.utils.translation import gettext_lazy as _
from background_task import background
from django_huey import db_periodic_task, db_task, task # noqa
from huey import CancelExecution
from background_task.exceptions import InvalidTaskError
from background_task.models import Task, CompletedTask
from common.logger import log
@@ -63,7 +65,6 @@ def map_task_to_instance(task):
'sync.tasks.download_media': Media,
'sync.tasks.download_media_metadata': Media,
'sync.tasks.save_all_media_for_source': Source,
'sync.tasks.refresh_formats': Media,
'sync.tasks.rename_media': Media,
'sync.tasks.rename_all_media_for_source': Source,
'sync.tasks.wait_for_media_premiere': Media,
@@ -209,29 +210,22 @@ def migrate_queues():
def save_model(instance):
with atomic(durable=False):
instance.save()
if 'sqlite' != db_vendor:
with atomic(durable=False):
instance.save()
return
# work around for SQLite and its many
# "database is locked" errors
with atomic(durable=False):
instance.save()
arg = getattr(settings, 'SQLITE_DELAY_FLOAT', 1.5)
time.sleep(random.expovariate(arg))
@atomic(durable=False)
def schedule_media_servers_update():
# Schedule a task to update media servers
log.info('Scheduling media server updates')
verbose_name = _('Request media server rescan for "{}"')
for mediaserver in MediaServer.objects.all():
rescan_media_server(
str(mediaserver.pk),
verbose_name=verbose_name.format(mediaserver),
)
rescan_media_server(str(mediaserver.pk))
def wait_for_errors(model, /, *, task_name=None):
@@ -266,8 +260,9 @@ def wait_for_errors(model, /, *, task_name=None):
update_task_status(task, None)
def cleanup_old_media():
with atomic():
@db_task(queue=Val(TaskQueue.FS))
def cleanup_old_media(durable=True):
with atomic(durable=durable):
for source in qs_gen(Source.objects.filter(delete_old_media=True, days_to_keep__gt=0)):
delta = timezone.now() - timedelta(days=source.days_to_keep)
mqs = source.media_source.defer(
@@ -280,13 +275,19 @@ def cleanup_old_media():
log.info(f'Deleting expired media: {source} / {media} '
f'(now older than {source.days_to_keep} days / '
f'download_date before {delta})')
with atomic():
with atomic(durable=False):
# .delete() also triggers a pre_delete/post_delete signals that remove files
media.delete()
schedule_media_servers_update()
def cleanup_removed_media(source, video_keys):
@db_task(queue=Val(TaskQueue.FS))
def cleanup_removed_media(source_id, video_keys):
try:
source = Source.objects.get(pk=source_id)
except Source.DoesNotExist as e:
# Task triggered but the Source has been deleted, delete the task
raise CancelExecution(_('no such source'), retry=False) from e
if not source.delete_removed_media:
return
log.info(f'Cleaning up media no longer in source: {source}')
@@ -295,11 +296,12 @@ def cleanup_removed_media(source, video_keys):
).filter(
source=source,
)
for media in qs_gen(mqs):
if media.key not in video_keys:
log.info(f'{media.name} is no longer in source, removing')
with atomic():
media.delete()
with atomic(durable=True):
for media in qs_gen(mqs):
if media.key not in video_keys:
log.info(f'{media.name} is no longer in source, removing')
with atomic(durable=False):
media.delete()
schedule_media_servers_update()
@@ -519,7 +521,7 @@ def index_source_task(source_id):
save_db_batch(Metadata.objects, db_batch_data, db_fields_data)
save_db_batch(Media.objects, db_batch_media, db_fields_media)
# Cleanup of media no longer available from the source
cleanup_removed_media(source, video_keys)
cleanup_removed_media(str(source.pk), video_keys)
# Clear references to indexed data
videos = video = None
db_batch_data.clear()
@@ -552,7 +554,7 @@ def check_source_directory_exists(source_id):
source.make_directory()
@background(schedule=dict(priority=10, run_at=10), queue=Val(TaskQueue.NET))
@db_task(delay=10, priority=90, queue=Val(TaskQueue.NET))
def download_source_images(source_id):
'''
Downloads an image and save it as a local thumbnail attached to a
@@ -564,7 +566,7 @@ def download_source_images(source_id):
# Task triggered but the source no longer exists, do nothing
log.error(f'Task download_source_images(pk={source_id}) called but no '
f'source exists with ID: {source_id}')
raise InvalidTaskError(_('no such source')) from e
raise CancelExecution(_('no such source'), retry=False) from e
avatar, banner = source.get_image_url
log.info(f'Thumbnail URL for source with ID: {source_id} / {source} '
f'Avatar: {avatar} '
@@ -783,10 +785,7 @@ def download_media(media_id, override=False):
# Try refreshing formats
if media.has_metadata:
log.debug(f'Scheduling a task to refresh metadata for: {media.key}: "{media.name}"')
refresh_formats(
str(media.pk),
verbose_name=f'Refreshing metadata formats for: {media.key}: "{media.name}"',
)
refresh_formats(str(media.pk))
log.exception(str(e))
raise
else:
@@ -794,10 +793,7 @@ def download_media(media_id, override=False):
# Try refreshing formats
if media.has_metadata:
log.debug(f'Scheduling a task to refresh metadata for: {media.key}: "{media.name}"')
refresh_formats(
str(media.pk),
verbose_name=f'Refreshing metadata formats for: {media.key}: "{media.name}"',
)
refresh_formats(str(media.pk))
# Expected file doesn't exist on disk
err = (f'Failed to download media: {media} (UUID: {media.pk}) to disk, '
f'expected outfile does not exist: {filepath}')
@@ -814,7 +810,7 @@ def download_media(media_id, override=False):
schedule_media_servers_update()
@background(schedule=dict(priority=0, run_at=30), queue=Val(TaskQueue.NET), remove_existing_tasks=True)
@db_task(delay=30, expire=60, priority=100, queue=Val(TaskQueue.NET))
def rescan_media_server(mediaserver_id):
'''
Attempts to request a media rescan on a remote media server.
@@ -823,7 +819,7 @@ def rescan_media_server(mediaserver_id):
mediaserver = MediaServer.objects.get(pk=mediaserver_id)
except MediaServer.DoesNotExist as e:
# Task triggered but the media server no longer exists, do nothing
raise InvalidTaskError(_('no such server')) from e
raise CancelExecution(_('no such server'), retry=False) from e
# Request an rescan / update
log.info(f'Updating media server: {mediaserver}')
mediaserver.update()
@@ -876,10 +872,7 @@ def save_all_media_for_source(source_id):
tvn_format = '1/{:,}' + f'/{refresh_qs.count():,}'
for mn, media in enumerate(qs_gen(refresh_qs), start=1):
update_task_status(task, tvn_format.format(mn))
refresh_formats(
str(media.pk),
verbose_name=f'Refreshing metadata formats for: {media.key}: "{media.name}"',
)
refresh_formats(str(media.pk))
saved_later.add(media.uuid)
# Keep out of the way of the index task!
@@ -901,12 +894,12 @@ def save_all_media_for_source(source_id):
update_task_status(task, None)
@background(schedule=dict(priority=50, run_at=0), queue=Val(TaskQueue.NET), remove_existing_tasks=True)
@db_task(priority=50, queue=Val(TaskQueue.LIMIT))
def refresh_formats(media_id):
try:
media = Media.objects.get(pk=media_id)
except Media.DoesNotExist as e:
raise InvalidTaskError(_('no such media')) from e
raise CancelExecution(_('no such media'), retry=False) from e
try:
media.refresh_formats
except YouTubeError as e: