19 lines
568 B
Python
19 lines
568 B
Python
from app.core.celery import celery_app
|
|
from app.services import YouTubeService
|
|
from app.schemas import DownloadRequest
|
|
|
|
|
|
@celery_app.task(
|
|
bind=True,
|
|
name="download_youtube",
|
|
autoretry_for=(Exception,),
|
|
retry_kwargs={"max_retries": 3},
|
|
retry_backoff=True,
|
|
)
|
|
def download_youtube(self, url: str, quality: int) -> dict:
|
|
print("Get Task. Try to make them")
|
|
request = DownloadRequest(url=url, quality=quality)
|
|
response = YouTubeService.download(request)
|
|
print("Task make successfully. Return response")
|
|
return response.model_dump()
|