1. import queue
  2. import threading
  3. import time
  4. import json
  5. import requests
  7. class TaskQueue:
  8. def __init__(self, staging_url, dry_run=False):
  9. self.queue = queue.Queue() # Queue for tasks
  10. self.staging_url = staging_url # URL for staging environment
  11. self.dry_run = dry_run # Dry-run mode flag
  12. self.lock = threading.Lock() # Lock for thread safety
  14. def enqueue(self, task):
  15. """Adds a task to the queue."""
  16. with self.lock:
  17. self.queue.put(task)
  19. def dequeue(self):
  20. """Retrieves a task from the queue."""
  21. try:
  22. return self.queue.get(timeout=1) # Wait for 1 second
  23. except queue.Empty:
  24. return None
  26. def start_worker(self, worker_id):
  27. """Starts a worker thread to process tasks."""
  28. def worker():
  29. while True:
  30. task = self.dequeue()
  31. if task:
  32. try:
  33. self.process_task(task)
  34. except Exception as e:
  35. print(f"Error processing task: {e}")
  36. else:
  37. time.sleep(0.1) # avoid busy waiting
  38. thread = threading.Thread(target=worker)
  39. thread.daemon = True # Allow the main thread to exit even if workers are running
  40. thread.start()
  42. def process_task(self, task):
  43. """Processes a single web form task."""
  44. try:
  45. # Validate task data
  46. if not isinstance(task, dict) or 'form_data' not in task:
  47. raise ValueError("Invalid task format")
  49. form_data = task['form_data']
  50. form_id = task['form_id']
  52. # Construct the staging URL
  53. url = f"{self.staging_url}/{form_id}"
  55. # Prepare the request
  56. headers = {'Content-Type': 'application/x-www-form-urlencoded'}
  57. response = requests.post(url, data=form_data, headers=headers)
  59. response.raise_for_status() # Raise HTTPError for bad responses (4xx or 5xx)
  61. print(f"Task {form_id} successful. Response: {response.text}")
  63. except requests.exceptions.RequestException as e:
  64. print(f"Request error for task {form_id}: {e}")
  65. except ValueError as e:
  66. print(f"Task error {e}")
  67. except Exception as e:
  68. print(f"Unexpected error processing task {form_id}: {e}")
  71. def run(self, num_workers=2):
  72. """Starts the task queue and worker threads."""
  73. self.start_worker(1)
  74. for i in range(2, num_workers + 1):
  75. self.start_worker(i)
  78. if __name__ == '__main__':
  79. # Example usage
  80. staging_url = "https://staging.example.com/forms" # Replace with your staging URL
  81. task_queue = TaskQueue(staging_url, dry_run=False) # Set dry_run to True for testing
  83. # Sample tasks
  84. tasks = [
  85. {"form_id": "form123", "form_data": {"field1": "value1", "field2": "value2"}},
  86. {"form_id": "form456", "form_data": {"field1": "value3", "field2": "value4"}},
  87. {"form_id": "form789", "form_data": {"field1": "value5", "field2": "value6"}},
  88. ]
  90. for task in tasks:
  91. task_queue.enqueue(task)
  93. task_queue.run(num_workers=3)

Add your comment