66 lines
1.6 KiB
Python
66 lines
1.6 KiB
Python
"""
|
|
Simple test tasks for Celery functionality.
|
|
"""
|
|
import time
|
|
import logging
|
|
from celery import shared_task
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
@shared_task(name='mainapp.test_celery_connection')
|
|
def test_celery_connection(message="Hello from Celery!"):
|
|
"""
|
|
A simple test task to verify Celery is working.
|
|
|
|
Args:
|
|
message (str): Message to return
|
|
|
|
Returns:
|
|
str: Confirmation message with task completion time
|
|
"""
|
|
logger.info(f"Test task started with message: {message}")
|
|
time.sleep(2) # Simulate some work
|
|
result = f"Task completed! Received message: {message}"
|
|
logger.info(f"Test task completed: {result}")
|
|
return result
|
|
|
|
|
|
@shared_task(name='mainapp.add_numbers')
|
|
def add_numbers(x, y):
|
|
"""
|
|
A simple addition task to test Celery functionality.
|
|
|
|
Args:
|
|
x (int): First number
|
|
y (int): Second number
|
|
|
|
Returns:
|
|
int: Sum of x and y
|
|
"""
|
|
logger.info(f"Adding {x} + {y}")
|
|
result = x + y
|
|
logger.info(f"Addition completed: {x} + {y} = {result}")
|
|
return result
|
|
|
|
|
|
@shared_task(name='mainapp.long_running_task')
|
|
def long_running_task(duration=10):
|
|
"""
|
|
A task that runs for a specified duration to test long-running tasks.
|
|
|
|
Args:
|
|
duration (int): Duration in seconds
|
|
|
|
Returns:
|
|
str: Completion message
|
|
"""
|
|
logger.info(f"Starting long running task for {duration} seconds")
|
|
for i in range(duration):
|
|
time.sleep(1)
|
|
logger.info(f"Long task progress: {i+1}/{duration}")
|
|
|
|
result = f"Long running task completed after {duration} seconds"
|
|
logger.info(result)
|
|
return result
|