Fusion Worker

git clone git@bitbucket.org:creliohealth-repo/fusion_worker.git

fusion_worker/
├── run_worker.py              # Worker entry point
├── config.py                  # Configuration loader
├── rq/                        # Custom RQ implementation
│   ├── worker.py              # Enhanced worker with retry logic
│   ├── job.py                 # Job class
│   ├── queue.py               # Queue class
│   ├── connections.py         # Redis connection management
│   └── cli/                   # CLI commands
│       └── cli.py             # Worker CLI
├── tasks/                     # Task implementations
│   ├── sms.py                 # SMS sending logic
│   ├── _email_v2.py           # Email via AWS SES
│   ├── push_notification.py   # Push notifications
│   ├── webhooks.py            # HTTP webhooks
│   ├── whatsapp.py            # WhatsApp messaging
│   ├── plugin_sms.py          # SMS provider plugins
│   ├── plugin_whatsapp.py     # WhatsApp plugins
│   ├── sms_gateways.py        # SMS gateway definitions
│   └── maintenance.py         # Queue maintenance tasks
├── wrappers/                  # Utility wrappers
│   ├── Logger.py              # Elasticsearch logging
│   └── _utils.py              # Helper functions
├── certificates/              # SSL certificates
└── instance/                  # Environment configs
    ├── production.py          # Production settings
    └── local.py               # Local development settings

# Create virtual environment
python3.6 -m venv venv
source venv/bin/activate

# Install dependencies
pip install -r requirement.txt

# Set environment variables
export CRELIO_DEPLOYMENT_ZONE="IN"
export CRELIO_DEPLOYMENT_MODE="LOCAL"

# Run worker for all queues
python run_worker.py worker -c instance --customq all

worker: CRELIO_DEPLOYMENT_ZONE="IN" CRELIO_DEPLOYMENT_MODE="LOCAL" python run_worker.py worker -c instance --customq all

python run_worker.py worker [OPTIONS]

Options:
  -c, --config TEXT       Module containing RQ settings (e.g., 'instance')
  --customq TEXT          Specific queue to work on, or 'all' for all queues
  -b, --burst             Run in burst mode (quit when queues are empty)
  -n, --name TEXT         Worker name
  --logging_level TEXT    Logging level (default: INFO)
  -v, --verbose           Show more output
  -q, --quiet             Show less output
  --results-ttl INTEGER   Result TTL in seconds
  --worker-ttl INTEGER    Worker TTL in seconds

# Get highest priority job
_items = connection.zrevrange(current_queue, 0, 0)

# Atomic removal
if connection.zrem(current_queue, _item) == 1:
    # Successfully claimed the job
    result = _item

if current_queue == 'rq:queue:scheduler':
    _item = self.lpop([current_queue], timeout, connection)
else:
    _items = connection.zrevrange(current_queue, 0, 0)

def handle_job_failure(self, job, queue, started_job_registry=None, retry_flag=False):
    if job.retryCount < 10 and retry_flag:
        # Schedule retry
        job.retryCount += 1
        job.save()
        requests.post(settings.FUSION_SERVER + '/retryTask/', json=job._args)
    else:
        # Call failure callback and move to failed queue
        if job.failed_task_handler:
            self.failed_task_callback(job)
        self.custom_failed_handle(job, started_job_registry)

# On success
def success_task_callback(self, job):
    job_meta = {
        'description': job.description,
        'job_id': job.id,
        'status': job._status,
        'retry_attempts': job.retryCount
    }
    requests.post(job.success_task_handler, data=job_meta)

# On failure
def failed_task_callback(self, job):
    job_meta = {
        'description': job.description,
        'failure_trace': job.exc_info,
        'job_id': job.id,
        'status': job._status
    }
    requests.post(job.failed_task_handler, data=job_meta)

queues = list(set(queues) - set(['dictionary', 'report']))

def isWhitelisted(job):
    if job._args[0]['payload']['task'] != 4:
        return 1  # Non-webhook tasks always pass
    
    test = re.match(settings.regex2, job._args[0]['payload']['url'])
    if 'staging' in job._args[0]['payload']['url']:
        job.retryCount += 10  # Allow more retries for staging
        return 1
    # ... additional validation

@property
def should_run_maintenance_tasks(self):
    if self.last_cleaned_at is None:
        return True
    if (utcnow() - self.last_cleaned_at) > timedelta(hours=24):
        return True
    return False

def clean_registries(self):
    def remove(job_id):
        if self.connection.delete(f"rq:job:{job_id}"):
            self.connection.lrem("rq:queue:failed", 0, job_id)
    
    for queue in self.queues:
        clean_registries(queue)
    self.last_cleaned_at = utcnow()