Introduction to Advanced Queue Management

The advanced-queue is a powerful tool that provides extensive functionalities for handling job queues efficiently. This tutorial dives deep into dozens of useful API explanations with corresponding code snippets, tailored to help you master queue management.

Getting Started with Advanced Queue

Before exploring the APIs, make sure you’ve installed the necessary package:

  
    pip install advanced-queue
  

API Examples

1. Creating a Queue

Create a new queue to start adding and processing jobs:

  
    from advanced_queue import Queue

    queue = Queue(name='my_queue')
  

2. Adding Jobs to the Queue

Add jobs to your queue with a specified priority:

  
    def example_task(data):
        print(f'Processing {data}')

    queue.add_job(example_task, data={'id': 1, 'info': 'Sample Task'})
  

3. Processing Jobs

Automatically process jobs in the queue:

  
    queue.process_jobs()
  

4. Handling Job Completion

Manage job completion with a callback function:

  
    def on_complete(job):
        print(f'Job {job.id} completed')

    queue.on_complete(on_complete)
  

5. Error Handling

Implement error handling for failed jobs:

  
    def on_error(job, error):
        print(f'Job {job.id} failed with error: {error}')

    queue.on_error(on_error)
  

6. Scheduling Jobs

Schedule jobs to run at specific times:

  
    from datetime import datetime, timedelta

    run_at = datetime.now() + timedelta(hours=1)
    queue.add_job(example_task, data={'id': 2, 'info': 'Scheduled Task'}, run_at=run_at)
  

Application Example with Advanced Queue

Below is a full-fledged example that integrates several of the APIs discussed:

  
    from advanced_queue import Queue
    from datetime import datetime, timedelta

    def example_task(data):
        print(f'Processing {data}')

    def on_complete(job):
        print(f'Job {job.id} completed')

    def on_error(job, error):
        print(f'Job {job.id} failed with error: {error}')

    queue = Queue(name='my_queue')
    queue.add_job(example_task, data={'id': 1, 'info': 'Sample Task'})
    run_at = datetime.now() + timedelta(hours=1)
    queue.add_job(example_task, data={'id': 2, 'info': 'Scheduled Task'}, run_at=run_at)
    queue.on_complete(on_complete)
    queue.on_error(on_error)
    queue.process_jobs()
  

Hash: bdb3fd26d1c4fd1d633d666c73c00fc986a9637e777c4d4da6a43d0665658dfd