### Starting a Worker Instance

Source: https://github.com/django-commons/django-tasks-scheduler/blob/master/_autodocs/api_reference_worker.md

Example of how to create and start a worker instance using the 'create_worker' utility function. The worker is configured to run in burst mode and will exit after processing all available jobs.

```python
from scheduler.worker import create_worker

worker = create_worker(['default', 'high-priority'], burst=True)
exit_code = worker.work()
```

--------------------------------

### Start Redis Server

Source: https://github.com/django-commons/django-tasks-scheduler/blob/master/_autodocs/quick_start.md

Start a Redis server instance using Docker for the scheduler to use as a broker.

```bash
docker run -d -p 6379:6379 redis:latest
```

--------------------------------

### Start Local Redis Broker

Source: https://github.com/django-commons/django-tasks-scheduler/blob/master/CLAUDE.md

Start a local Redis instance using Docker Compose for testing or development.

```bash
docker compose up redis8 -d
```

--------------------------------

### Start a Worker

Source: https://github.com/django-commons/django-tasks-scheduler/blob/master/_autodocs/README.md

Start a scheduler worker process using the `manage.py` command. Use the `-q` flag to specify which queues the worker should listen to.

```bash
python manage.py scheduler_worker -q default -q high
```

--------------------------------

### Install Pre-commit Hook

Source: https://github.com/django-commons/django-tasks-scheduler/blob/master/README.md

Install the pre-commit hook to enforce code style and quality checks before committing changes.

```bash
pre-commit install
```

--------------------------------

### SchedulerConfiguration Example

Source: https://github.com/django-commons/django-tasks-scheduler/blob/master/_autodocs/types.md

Provides an example of how to instantiate the SchedulerConfiguration dataclass with custom values for various settings. This allows fine-tuning of scheduler behavior like job timeouts and result TTLs.

```python
from scheduler.types import SchedulerConfiguration, Broker

SCHEDULER_CONFIG = SchedulerConfiguration(
    EXECUTIONS_IN_PAGE=50,
    SCHEDULER_INTERVAL=5,
    BROKER=Broker.REDIS,
    DEFAULT_JOB_TIMEOUT=600,  # 10 minutes
    DEFAULT_SUCCESS_TTL=3600,  # 1 hour
    DEFAULT_FAILURE_TTL=24 * 3600,  # 1 day
)
```

--------------------------------

### Host-based QueueConfiguration Example

Source: https://github.com/django-commons/django-tasks-scheduler/blob/master/_autodocs/types.md

Shows how to configure a queue using host, port, and database number. This is an alternative to using a URL.

```python
from scheduler.types import QueueConfiguration

# Host-based configuration
SCHEDULER_QUEUES = {
    'default': QueueConfiguration(HOST='localhost', PORT=6379, DB=0),
}
```

--------------------------------

### Install Django Tasks Scheduler

Source: https://github.com/django-commons/django-tasks-scheduler/blob/master/_autodocs/SUMMARY.txt

Use pip to install the django-tasks-scheduler package.

```bash
pip install django-tasks-scheduler
```

--------------------------------

### Systemd Service Configuration for Multiple Workers

Source: https://github.com/django-commons/django-tasks-scheduler/blob/master/docs/usage.md

Example configuration file for running multiple scheduler workers as systemd services. This setup allows for managing workers as background services on Linux systems.

```ini
# /etc/systemd/system/scheduler_worker@.service
[Unit]
Description = scheduler_worker daemon
After = network.target

[Service]
WorkingDirectory = {{ path_to_your_project_folder } }
ExecStart = /home/ubuntu/.virtualenv/{ { your_virtualenv } }/bin/python \
            {{ path_to_your_project_folder } }/manage.py \
            scheduler_worker high default low
# Optional
# {{user to run scheduler_worker as}}
User = ubuntu
# {{group to run scheduler_worker as}}
Group = www-data
# Redirect logs to syslog
StandardOutput = syslog
StandardError = syslog
SyslogIdentifier = scheduler_worker
Environment = OBJC_DISABLE_INITIALIZE_FORK_SAFETY = YES
Environment = LC_ALL = en_US.UTF-8
Environment = LANG = en_US.UTF-8

[Install]
WantedBy = multi-user.target
```

--------------------------------

### URL-based QueueConfiguration Examples

Source: https://github.com/django-commons/django-tasks-scheduler/blob/master/_autodocs/types.md

Demonstrates how to configure queues using a direct Redis/Valkey URL. Supports basic URLs and those with authentication credentials.

```python
from scheduler.types import QueueConfiguration

# URL-based configuration
SCHEDULER_QUEUES = {
    'default': QueueConfiguration(URL='redis://localhost:6379/0'),
    'high': QueueConfiguration(URL='redis://user:pass@redis.example.com:6380/1'),
}
```

--------------------------------

### Using Callback with Decorator Example

Source: https://github.com/django-commons/django-tasks-scheduler/blob/master/_autodocs/api_reference_callback.md

An example demonstrating how to use the `Callback` helper with the `@job` decorator to specify callbacks for success, failure, and stopped events by referencing their string paths.

```python
from scheduler import job
from scheduler.helpers.callback import Callback

@job(
    queue='default',
    timeout=300,
    on_success=Callback('myapp.callbacks.log_success'),
    on_failure=Callback('myapp.callbacks.log_failure'),
    on_stopped=Callback('myapp.callbacks.log_stopped'),
)
def my_async_task(data):
    return process(data)
```

--------------------------------

### Start Worker with Multiple Queues and Options

Source: https://github.com/django-commons/django-tasks-scheduler/blob/master/_autodocs/quick_start.md

Start a worker that listens to multiple queues ('default', 'high-priority') and includes scheduler and burst options. Use the --burst option for short-lived tasks.

```bash
python manage.py scheduler_worker \
  -q default -q high \
  --with-scheduler \
  --burst
```

--------------------------------

### JSON Output Example for Exported Tasks

Source: https://github.com/django-commons/django-tasks-scheduler/blob/master/docs/commands.md

Example structure of a JSON file generated by the 'export' command, representing a single scheduled task.

```json
[
  {
    "model": "CronTaskType",
    "name": "Scheduled Task 1",
    "callable": "scheduler.tests.test_job",
    "callable_args": [
      {
        "arg_type": "datetime",
        "val": "2022-02-01"
      }
    ],
    "callable_kwargs": [],
    "enabled": true,
    "queue": "default",
    "at_front": false,
    "timeout": null,
    "result_ttl": null,
    "scheduled_time": "2023-02-21T14:06:06"
  },
  ...
]
```

--------------------------------

### Sentinel-based QueueConfiguration Example

Source: https://github.com/django-commons/django-tasks-scheduler/blob/master/_autodocs/types.md

Demonstrates setting up a queue configuration for high availability using Redis Sentinel. Requires specifying sentinel addresses and the master name.

```python
from scheduler.types import QueueConfiguration

# Sentinel configuration
SCHEDULER_QUEUES = {
    'default': QueueConfiguration(
        SENTINELS=[('sentinel1.example.com', 26379), ('sentinel2.example.com', 26379)],
        MASTER_NAME='mymaster',
        SENTINEL_KWARGS={'password': 'sentinel_password'},
    ),
}
```

--------------------------------

### Unix Socket QueueConfiguration Example

Source: https://github.com/django-commons/django-tasks-scheduler/blob/master/_autodocs/types.md

Illustrates configuring a queue connection via a Unix domain socket path. This is useful for local connections when a socket file is available.

```python
from scheduler.types import QueueConfiguration

# Unix socket
SCHEDULER_QUEUES = {
    'default': QueueConfiguration(UNIX_SOCKET_PATH='/tmp/redis.sock'),
}
```

--------------------------------

### Configuration Reference

Source: https://github.com/django-commons/django-tasks-scheduler/blob/master/_autodocs/MANIFEST.md

Comprehensive guide to configuring the Django Tasks Scheduler, including Django settings integration and detailed parameter documentation.

```APIDOC
## Configuration Reference

### Description
This document provides a detailed guide on how to configure the Django Tasks Scheduler, covering integration with Django settings, queue configurations, and various operational parameters.

### Django Settings Integration
- **`SCHEDULER_CONFIG`**: A dictionary in Django's `settings.py` to configure the scheduler. Contains approximately 15 parameters.
- **`SCHEDULER_QUEUES`**: A dictionary defining multiple job queues, each with its own configuration. Includes 10+ parameters per queue.

### Parameter Documentation
For both `SCHEDULER_CONFIG` and `SCHEDULER_QUEUES`, detailed tables are provided, including:
- **Parameter Name**: The name of the configuration setting.
- **Type**: The expected data type (e.g., string, integer, boolean, list).
- **Default Value**: The default value if not explicitly set.
- **Description**: An explanation of the parameter's purpose and effect.

### Validation Rules
- Rules governing the validity and constraints of configuration parameters.

### Environment Variables
- Patterns and usage for configuring settings via environment variables.

### Settings Patterns
- Recommended settings configurations for different environments: development, testing, and production.

### URL Registration
- Guidance on registering URLs related to the scheduler's functionality.

### Migrations Guide
- Instructions on applying database migrations required by the scheduler.

### Usage Examples
```python
# settings.py

SCHEDULER_CONFIG = {
    'broker': 'redis',
    'redis_url': 'redis://localhost:6379/0',
    'default_queue': 'celery',
    # ... other scheduler settings
}

SCHEDULER_QUEUES = {
    'celery': {
        'tasks_max_per_worker': 10,
        'max_concurrent_tasks': 5,
        # ... other queue settings
    },
    'high_priority': {
        # ... settings for high priority queue
    }
}
```
```

--------------------------------

### Start Scheduler Worker

Source: https://github.com/django-commons/django-tasks-scheduler/blob/master/_autodocs/SUMMARY.txt

Run the scheduler worker from the command line, specifying the queue(s) to process.

```bash
python manage.py scheduler_worker -q default
```

--------------------------------

### Handle CallbackSetupError Exception

Source: https://github.com/django-commons/django-tasks-scheduler/blob/master/_autodocs/errors.md

Use this handler to manage errors during Callback setup, including invalid callback functions or import issues.

```python
from scheduler.helpers.callback import Callback, CallbackSetupError

try:
    callback = Callback(123)  # Invalid: not callable
except CallbackSetupError as e:
    print(f"Invalid callback: {e}")

try:
    callback = Callback('nonexistent.module.func')
except CallbackSetupError as e:
    print(f"Cannot import callback function: {e}")
```

--------------------------------

### Custom Connection Arguments QueueConfiguration Example

Source: https://github.com/django-commons/django-tasks-scheduler/blob/master/_autodocs/types.md

Shows how to pass custom arguments to the underlying redis-py/valkey-py connection. This allows for fine-tuning connection behavior like socket keep-alive.

```python
from scheduler.types import QueueConfiguration

# Custom connection options
SCHEDULER_QUEUES = {
    'default': QueueConfiguration(
        URL='redis://localhost:6379/0',
        CONNECTION_KWARGS={'socket_keepalive': True, 'socket_keepalive_options': {...}}
    ),
}
```

--------------------------------

### WorkerScheduler.start()

Source: https://github.com/django-commons/django-tasks-scheduler/blob/master/_autodocs/api_reference_worker.md

Starts the scheduler by acquiring necessary locks and spawning a background thread to execute the main work loop.

```APIDOC
## start()

### Description
Acquires locks and starts the scheduler thread. It attempts to acquire locks on all queues and, if successful, spawns a background thread to run the `.work()` method, setting the status to `STARTED`.

### Signature
```python
def start(self) -> None
```

### Example
```python
scheduler = WorkerScheduler(queues=[default_queue], worker_name='worker-1')
scheduler.start()
```
```

--------------------------------

### Configure Redis Queues

Source: https://github.com/django-commons/django-tasks-scheduler/blob/master/docs/installation.md

Define Redis queues in settings.py using QueueConfiguration for robust setup. Supports default, sentinel, and URL-based configurations.

```python
import os
from typing import Dict
from scheduler.types import QueueConfiguration

SCHEDULER_QUEUES: Dict[str, QueueConfiguration] = {
 'default': QueueConfiguration(
 HOST='localhost',
 PORT=6379,
 USERNAME='some-user',
 PASSWORD='some-password',
 CONNECTION_KWARGS={  # Eventual additional Broker connection arguments
 'ssl_cert_reqs': 'required',
 'ssl': True,
 },
 ),
 'with-sentinel': QueueConfiguration(
 SENTINELS= [('localhost', 26736), ('localhost', 26737)],
 MASTER_NAME= 'redismaster',
 DB= 0,
 USERNAME= 'redis-user',
 PASSWORD= 'secret',
 CONNECTION_KWARGS= {
 'ssl': True},
 SENTINEL_KWARGS= {
 'username': 'sentinel-user',
 'password': 'secret',
      }),
 'high': QueueConfiguration(URL=os.getenv('REDISTOGO_URL', 'redis://localhost:6379/0')),
 'low': QueueConfiguration(HOST='localhost', PORT=6379, DB=0, ASYNC=False),
}
```

--------------------------------

### Simple Logging Callback Example

Source: https://github.com/django-commons/django-tasks-scheduler/blob/master/_autodocs/api_reference_callback.md

A simple callback function to log the job result upon successful completion. This is useful for basic monitoring.

```python
from scheduler import job

def log_job_result(job_model, connection, result):
    print(f"✓ {job_model.name}: {result}")

@job(queue='default', on_success=log_job_result)
def fetch_data():
    return "Data fetched successfully"
```

--------------------------------

### Failure Callback Example

Source: https://github.com/django-commons/django-tasks-scheduler/blob/master/_autodocs/api_reference_callback.md

Define a function to execute when a job fails. This callback receives the job instance, connection, and exception information.

```python
import logging
from scheduler import job
from scheduler.helpers.callback import Callback

logger = logging.getLogger(__name__)

def on_failure(job_model, connection, exc_info):
    exc_type, exc_value, exc_tb = exc_info
    logger.error(f"Job {job_model.name} failed with {exc_type.__name__}: {exc_value}")
    # Send alert, cleanup resources, etc.

@job(
    queue='default',
    on_failure=Callback(on_failure, timeout=10)
)
def risky_operation():
    # ... code that may fail ...
    pass
```

--------------------------------

### Success Callback Example

Source: https://github.com/django-commons/django-tasks-scheduler/blob/master/_autodocs/api_reference_callback.md

Define a function to execute when a job completes successfully. This callback receives the job instance, connection, and the result of the job.

```python
from scheduler import job

def on_success(job_model, connection, result):
    print(f"Job {job_model.name} succeeded with result: {result}")
    # Log to database, send notification, etc.

@job(queue='default', on_success=on_success)
def process_order(order_id):
    # ... process order ...
    return f"Order {order_id} processed"
```

--------------------------------

### Cleanup Callback Example

Source: https://github.com/django-commons/django-tasks-scheduler/blob/master/_autodocs/api_reference_callback.md

A callback function to perform cleanup operations when a job is stopped or canceled. This ensures resources are released properly.

```python
from scheduler import job
from scheduler.helpers.callback import Callback

def cleanup_resources(job_model, connection):
    """Called when job is stopped or canceled"""
    # Rollback database transaction, release locks, etc.
    print(f"Cleaning up for {job_model.name}")

@job(
    queue='default',
    on_stopped=Callback(cleanup_resources, timeout=30)
)
def transactional_job():
    pass
```

--------------------------------

### Test Task Execution

Source: https://github.com/django-commons/django-tasks-scheduler/blob/master/_autodocs/quick_start.md

Write Django tests to verify task enqueuing and synchronous execution. This example shows how to enqueue a job and run it synchronously for assertion.

```python
from django.test import TestCase
from scheduler.models import Task
from scheduler.helpers.queues import get_queue

class TaskTests(TestCase):
    def test_task_execution(self):
        queue = get_queue('default')
        
        # Enqueue job
        job = queue.create_and_enqueue_job(
            'myapp.tasks.my_function',
            args=('test_data',),
        )
        
        # Run synchronously for testing
        result = queue.run_sync(job)
        
        self.assertEqual(result.status, 'finished')
```

--------------------------------

### Worker.work()

Source: https://github.com/django-commons/django-tasks-scheduler/blob/master/_autodocs/api_reference_worker.md

Starts the worker's main loop to process jobs from queues. Can be configured to run for a maximum number of jobs or seconds.

```APIDOC
#### work()

```python
def work(
    self,
    with_results: bool = False,
    max_jobs: Optional[int] = None,
    max_seconds: Optional[int] = None,
) -> int
```

Starts the worker main loop, processing jobs from queues.

**Parameters**

| Parameter | Type | Default | Description |
|-----------|------|---------|-------------|
| with_results | `bool` | `False` | Unused; kept for API compatibility |
| max_jobs | `Optional[int]` | `None` | Maximum jobs to process before exiting; `None` = unlimited |
| max_seconds | `Optional[int]` | `None` | Maximum seconds to run before exiting; `None` = unlimited |

**Returns**: Exit code (0 = success, 1 = error)

**Behavior**:
- Acquires lock on queues
- Enters infinite loop (or until `burst=True` and queue is empty)
- Dequeues jobs using the configured strategy
- Executes jobs (via forked process if `fork_job_execution=True`)
- Refreshes locks and runs maintenance tasks
- Graceful shutdown on SIGTERM/SIGINT

**Example**

```python
from scheduler.worker import create_worker

worker = create_worker(['default', 'high-priority'], burst=True)
exit_code = worker.work()
```
```

--------------------------------

### TaskKwarg value() Method Example

Source: https://github.com/django-commons/django-tasks-scheduler/blob/master/_autodocs/api_reference_models.md

Shows how to retrieve a keyword argument's key and parsed value using the TaskKwarg model. The kwarg object should be fetched and its attributes correctly assigned.

```python
kwarg = TaskKwarg.objects.get(id=1)
kwarg.key = 'email'
kwarg.arg_type = 'str'
kwarg.val = 'user@example.com'
key, value = kwarg.value()
```

--------------------------------

### Stopped Callback Example

Source: https://github.com/django-commons/django-tasks-scheduler/blob/master/_autodocs/api_reference_callback.md

Define a function to execute when a job is manually stopped or canceled. This callback receives the job instance and connection.

```python
from scheduler import job

def on_stopped(job_model, connection):
    print(f"Job {job_model.name} was stopped")
    # Cleanup, rollback, etc.

@job(queue='default', on_stopped=on_stopped)
def long_running_task():
    # ... code that may be stopped ...
    pass
```

--------------------------------

### TaskArg value() Method Example

Source: https://github.com/django-commons/django-tasks-scheduler/blob/master/_autodocs/api_reference_models.md

Demonstrates how to retrieve and use the parsed value of a TaskArg. Ensure the TaskArg object is properly fetched and its attributes are set.

```python
arg = TaskArg.objects.get(id=1)
arg.arg_type = 'int'
arg.val = '42'
print(arg.value())  # Returns 42 (int)
```

--------------------------------

### GET /scheduler/queues/

Source: https://github.com/django-commons/django-tasks-scheduler/blob/master/_autodocs/endpoints.md

Displays a dashboard with queue statistics, including job counts per registry and worker status for each queue. Provides real-time statistics.

```APIDOC
## GET /scheduler/queues/

### Description
Queue statistics dashboard.

### Method
GET

### Endpoint
/scheduler/queues/

### Response
#### Success Response (200)
- HTML page showing: Queue counts per registry (queued, active, scheduled, finished, failed), Worker status for each queue, Real-time statistics

### Request Example
```bash
curl http://localhost:8000/scheduler/queues/
```
```

--------------------------------

### Catch Broker-Specific Exceptions

Source: https://github.com/django-commons/django-tasks-scheduler/blob/master/_autodocs/types.md

Example of how to use the defined error type aliases in an except clause to handle connection or response errors during Redis operations.

```python
from scheduler.types import ConnectionErrorTypes, ResponseErrorTypes

try:
    # Redis operation
    result = connection.get('key')
except ConnectionErrorTypes:
    print("Connection failed")
except ResponseErrorTypes:
    print("Invalid command or response")
```

--------------------------------

### Error Notification Callback Example

Source: https://github.com/django-commons/django-tasks-scheduler/blob/master/_autodocs/api_reference_callback.md

A callback function to send an email notification to administrators when a job fails. It utilizes Django's email system.

```python
from scheduler import job
import smtplib

def notify_admin_on_failure(job_model, connection, exc_info):
    exc_type, exc_value, exc_tb = exc_info
    
    # Send error email
    message = f"""
    Job {job_model.name} failed with error:
    {exc_type.__name__}: {exc_value}
    """
    
    # Use Django's mail system
    from django.core.mail import send_mail
    send_mail(
        subject=f"Job Failure: {job_model.name}",
        message=message,
        from_email='alerts@example.com',
        recipient_list=['admin@example.com'],
    )

@job(queue='default', on_failure=notify_admin_on_failure)
def critical_task():
    pass
```

--------------------------------

### Implement Retry Logic with Callback

Source: https://github.com/django-commons/django-tasks-scheduler/blob/master/_autodocs/api_reference_callback.md

Implement a custom `on_failure` callback to handle job retries. This example limits retries to 3 and re-enqueues the job with an incremented retry count.

```python
from scheduler import job
from scheduler.helpers.queues import get_queue

def retry_on_failure(job_model, connection, exc_info):
    """Retry failed job after delay"""
    exc_type, exc_value, exc_tb = exc_info
    
    if job_model.meta.get('retry_count', 0) < 3:
        # Enqueue retry
        queue = get_queue(job_model.queue_name)
        queue.create_and_enqueue_job(
            job_model.func_name,
            args=job_model.args,
            kwargs=job_model.kwargs,
            meta={'retry_count': job_model.meta.get('retry_count', 0) + 1},
        )

@job(queue='default', on_failure=retry_on_failure)
def retryable_task():
    pass
```

--------------------------------

### Get Active Jobs Before Timestamp

Source: https://github.com/django-commons/django-tasks-scheduler/blob/master/_autodocs/api_reference_redis_models.md

Retrieves the names of jobs that started executing before a specified timestamp from the ActiveJobRegistry. Useful for detecting and handling job timeouts.

```python
get_job_names_before(connection, timestamp)
```

--------------------------------

### Run Background Worker for Queued Jobs

Source: https://github.com/django-commons/django-tasks-scheduler/blob/master/docs/usage.md

Start a worker process to handle jobs from specified queues in the background. This command offers various configuration options for worker behavior and integration.

```shell
usage: manage.py scheduler_worker [-h] [--pid PIDFILE] [--name NAME] [--worker-ttl WORKER_TTL] [--fork-job-execution FORK_JOB_EXECUTION] [--sentry-dsn SENTRY_DSN] [--sentry-debug] [--sentry-ca-certs SENTRY_CA_CERTS] [--burst]
                                  [--max-jobs MAX_JOBS] [--max-idle-time MAX_IDLE_TIME] [--with-scheduler] [--version] [-v {0,1,2,3}] [--settings SETTINGS] [--pythonpath PYTHONPATH] [--traceback] [--no-color] [--force-color]
                                  [--skip-checks]
                                  [queues ...]
```

--------------------------------

### JobModel get Method Signature

Source: https://github.com/django-commons/django-tasks-scheduler/blob/master/_autodocs/api_reference_redis_models.md

Signature for the get class method, used to retrieve a job from Redis by its name.

```python
@classmethod
def get(
    cls,
    name: str,
    connection: ConnectionType
) -> Optional[JobModel]
```

--------------------------------

### Configure Scheduler Queues with Host, Port, and DB

Source: https://github.com/django-commons/django-tasks-scheduler/blob/master/_autodocs/configuration.md

Configure queues using individual host, port, and database parameters. All three must be provided together.

```python
SCHEDULER_QUEUES = {
    'default': QueueConfiguration(HOST='localhost', PORT=6379, DB=0),
    'production': QueueConfiguration(HOST='redis.prod.internal', PORT=6379, DB=1),
}
```

--------------------------------

### Run Job Immediately

Source: https://github.com/django-commons/django-tasks-scheduler/blob/master/docs/commands.md

Executes a specified method (callable) and its arguments in a queue immediately. Ensure the callable and its arguments are correctly formatted.

```shell
python manage.py run_job {callable} {callable args ...}
```

--------------------------------

### Run Scheduler Migrations

Source: https://github.com/django-commons/django-tasks-scheduler/blob/master/_autodocs/quick_start.md

Apply database migrations for the scheduler app.

```bash
python manage.py migrate scheduler
```

--------------------------------

### Get All Jobs

Source: https://github.com/django-commons/django-tasks-scheduler/blob/master/_autodocs/api_reference_queue.md

Retrieves all `JobModel` instances from all job registries. This provides a comprehensive list of all managed jobs.

```python
def get_all_jobs(self) -> List[JobModel]
```

--------------------------------

### Get Absolute URL for Task

Source: https://github.com/django-commons/django-tasks-scheduler/blob/master/_autodocs/api_reference_models.md

Retrieves the URL path to the task's representation in the Django admin interface.

```python
get_absolute_url()
```

--------------------------------

### Run Database Migrations

Source: https://github.com/django-commons/django-tasks-scheduler/blob/master/docs/installation.md

Execute Django migrations to create the necessary database tables for the scheduler.

```shell
python manage.py migrate
```

--------------------------------

### SchedulerStatus Enum

Source: https://github.com/django-commons/django-tasks-scheduler/blob/master/_autodocs/api_reference_worker.md

Represents the possible states of the WorkerScheduler. Use STARTED, WORKING, or STOPPED to track the scheduler's lifecycle.

```python
class SchedulerStatus(str, Enum):
    STARTED = "started"
    WORKING = "working"
    STOPPED = "stopped"
```

--------------------------------

### Get All Job Names

Source: https://github.com/django-commons/django-tasks-scheduler/blob/master/_autodocs/api_reference_queue.md

Retrieves a list of names for all jobs currently present in all registries. This can be useful for monitoring or auditing purposes.

```python
def get_all_job_names(self) -> List[str]
```

--------------------------------

### Start WorkerScheduler

Source: https://github.com/django-commons/django-tasks-scheduler/blob/master/_autodocs/api_reference_worker.md

Acquires necessary locks and initiates the scheduler's background thread. Ensure queues are properly configured before calling.

```python
scheduler = WorkerScheduler(queues=[default_queue], worker_name='worker-1')
scheduler.start()
```

--------------------------------

### Prepare Job for Execution

Source: https://github.com/django-commons/django-tasks-scheduler/blob/master/_autodocs/api_reference_redis_models.md

Prepares a job for execution by moving it to the active registry and setting the worker name.

```python
def prepare_for_execution(
    self,
    worker_name: str,
    active_job_registry: ActiveJobRegistry,
    connection: ConnectionType
) -> None:
    ...
```

--------------------------------

### Get Worker Detail

Source: https://github.com/django-commons/django-tasks-scheduler/blob/master/_autodocs/endpoints.md

Retrieve detailed information about a specific worker by its name. This includes metadata, assigned queues, and current/recent jobs.

```bash
curl http://localhost:8000/scheduler/workers/worker-1/
```

--------------------------------

### Get Callable Function from Task

Source: https://github.com/django-commons/django-tasks-scheduler/blob/master/_autodocs/api_reference_models.md

Retrieves the actual Python callable associated with a task. This is useful for dynamically executing task functions.

```python
task = Task.objects.get(name='send_email')
func = task.callable_func()
result = func()
```

--------------------------------

### Queue Class and Helpers

Source: https://github.com/django-commons/django-tasks-scheduler/blob/master/_autodocs/MANIFEST.md

API documentation for the Queue class and associated helper functions for managing jobs and queues. Covers creating, running, canceling, and retrieving jobs.

```APIDOC
## Queue Class and Helper Functions

### Description
The `Queue` class provides an interface for interacting with job queues. Helper functions allow for easy access to queue instances and the current job context.

### Queue Class Methods
- **`create_and_enqueue_job(func, *args, **kwargs)`**: Creates a new job from the given function and arguments, and enqueues it.
- **`run_sync(func, *args, **kwargs)`**: Executes a job synchronously in the current process.
- **`cancel_job(job_id)`**: Cancels a job with the specified ID.
- **`dequeue_any()`**: Retrieves and removes the next available job from any queue.
- **`get_all_jobs()`**: Returns a list of all jobs currently in the queue.
- **`get_registry()`**: Accesses the job registry associated with the queue.

### Helper Functions
- **`get_queue(name='default')`**: Returns a `Queue` instance for the specified queue name.
- **`get_current_job()`**: Returns the `Job` object for the currently executing job, if any.

### Exception Classes
- **`InvalidJobOperation`**: Raised for invalid operations on a job.
- **`NoSuchJobError`**: Raised when a specified job ID does not exist.
- **`NoSuchRegistryError`**: Raised when a specified registry does not exist.

### Usage Examples
```python
from django_tasks_scheduler.queue import Queue, get_queue

queue = get_queue('my_queue')
job_id = queue.create_and_enqueue_job(my_task, 'arg1', kwarg1='value')

try:
    queue.cancel_job(job_id)
except NoSuchJobError:
    print(f'Job {job_id} not found.')
```
```

--------------------------------

### Basic Django Settings for Scheduler

Source: https://github.com/django-commons/django-tasks-scheduler/blob/master/_autodocs/configuration.md

Integrates the scheduler app and sets up the main configuration dictionary with various parameters. This includes defining broker type, execution limits, and time-to-live values for job results and metadata.

```python
from scheduler.types import SchedulerConfiguration, Broker

INSTALLED_APPS = [
    # ...
    'scheduler',
    # ...
]

SCHEDULER_CONFIG = SchedulerConfiguration(
    EXECUTIONS_IN_PAGE=20,
    SCHEDULER_INTERVAL=10,
    BROKER=Broker.REDIS,
    CALLBACK_TIMEOUT=60,
    DEFAULT_SUCCESS_TTL=10 * 60,
    DEFAULT_FAILURE_TTL=365 * 24 * 60 * 60,
    DEFAULT_JOB_TTL=10 * 60,
    DEFAULT_JOB_TIMEOUT=5 * 60,
    DEFAULT_WORKER_TTL=10 * 60,
    DEFAULT_MAINTENANCE_TASK_INTERVAL=10 * 60,
    DEFAULT_JOB_MONITORING_INTERVAL=30,
    SCHEDULER_FALLBACK_PERIOD_SECS=120,
)
```

--------------------------------

### Configure Django Settings

Source: https://github.com/django-commons/django-tasks-scheduler/blob/master/_autodocs/quick_start.md

Add 'scheduler' to INSTALLED_APPS and configure SCHEDULER_CONFIG and SCHEDULER_QUEUES in your Django settings.py. This sets up execution parameters, broker type, timeouts, and queue configurations.

```python
# settings.py
from scheduler.types import SchedulerConfiguration, QueueConfiguration, Broker

INSTALLED_APPS = [
    # ...
    'scheduler',
]

SCHEDULER_CONFIG = SchedulerConfiguration(
    EXECUTIONS_IN_PAGE=20,
    SCHEDULER_INTERVAL=10,
    BROKER=Broker.REDIS,
    DEFAULT_JOB_TIMEOUT=5 * 60,
    DEFAULT_SUCCESS_TTL=10 * 60,
)

SCHEDULER_QUEUES = {
    'default': QueueConfiguration(URL='redis://localhost:6379/0'),
}
```

--------------------------------

### Enqueue a Job via Command Line

Source: https://github.com/django-commons/django-tasks-scheduler/blob/master/docs/usage.md

Use this command to queue a job for execution from the command line. Specify the queue, timeout, result TTL, callable, and arguments.

```shell
python manage.py run_job -q {queue} -t {timeout} -r {result_ttl} {callable} {args}
```

--------------------------------

### Configure Scheduler Queues with Connection Keywords

Source: https://github.com/django-commons/django-tasks-scheduler/blob/master/_autodocs/configuration.md

Pass additional arguments to the redis-py/valkey-py connection. Useful for fine-tuning connection behavior like timeouts and keepalive.

```python
SCHEDULER_QUEUES = {
    'default': QueueConfiguration(
        URL='redis://localhost:6379/0',
        CONNECTION_KWARGS={
            'socket_keepalive': True,
            'socket_connect_timeout': 5,
            'socket_timeout': 10,
        }
    ),
}
```

--------------------------------

### Check Redis Queue Count

Source: https://github.com/django-commons/django-tasks-scheduler/blob/master/_autodocs/quick_start.md

Use the Django shell to get the count of jobs in the 'default' queue, useful for diagnosing connection issues.

```python
# Or check settings
python manage.py shell
>>> from scheduler.helpers.queues import get_queue
>>> queue = get_queue('default')
>>> queue.count
```

--------------------------------

### prepare_for_execution

Source: https://github.com/django-commons/django-tasks-scheduler/blob/master/_autodocs/api_reference_redis_models.md

Prepares a job for execution by moving it to the active registry and setting the worker name.

```APIDOC
## prepare_for_execution

### Description
Prepares a job for execution by moving it to the active registry and setting the worker name.

### Method Signature
```python
def prepare_for_execution(
    self,
    worker_name: str,
    active_job_registry: ActiveJobRegistry,
    connection: ConnectionType
) -> None
```

### Parameters
#### Arguments
- **worker_name** (str) - Required - The name of the worker executing the job.
- **active_job_registry** (ActiveJobRegistry) - Required - The registry for active jobs.
- **connection** (ConnectionType) - Required - Broker connection
```

--------------------------------

### Configure Scheduler and Queues

Source: https://github.com/django-commons/django-tasks-scheduler/blob/master/_autodocs/README.md

Configure the scheduler's behavior and queue settings using `SCHEDULER_CONFIG` and `SCHEDULER_QUEUES` dictionaries. Specify broker type, intervals, timeouts, and queue URLs.

```python
from scheduler.types import SchedulerConfiguration, QueueConfiguration, Broker

SCHEDULER_CONFIG = SchedulerConfiguration(
    BROKER=Broker.REDIS,                    # Redis, Valkey, or FakeRedis
    SCHEDULER_INTERVAL=10,                  # Seconds between scheduler checks
    DEFAULT_JOB_TIMEOUT=300,                # 5 minutes default job timeout
    DEFAULT_SUCCESS_TTL=600,                # Keep results for 10 minutes
    DEFAULT_FAILURE_TTL=31536000,           # Keep failures for 1 year
    CALLBACK_TIMEOUT=60,                    # Callback timeout
)

SCHEDULER_QUEUES = {
    'default': QueueConfiguration(
        URL='redis://localhost:6379/0',     # Or HOST/PORT/DB
        ASYNC=True,                         # True = async, False = sync
    ),
}
```

--------------------------------

### Get Queue and Job Statistics

Source: https://github.com/django-commons/django-tasks-scheduler/blob/master/_autodocs/endpoints.md

Retrieves statistics for all queues or a specific queue programmatically. Use this to monitor job counts and worker activity.

```python
from scheduler.views import get_statistics

def get_statistics(queue_name: Optional[str] = None) -> Dict[str, Any]
```

```python
from scheduler.views import get_statistics

stats = get_statistics()
print(f"Total jobs: {stats['total_jobs']}")
print(f"Active workers: {stats['active_workers']}")

# For specific queue
default_stats = get_statistics('default')
print(f"Queued jobs: {default_stats['queued']}")
```

--------------------------------

### GET /scheduler/queues/stats.json

Source: https://github.com/django-commons/django-tasks-scheduler/blob/master/_autodocs/endpoints.md

Retrieves queue statistics in JSON format, detailing job counts across different registries and worker availability for each queue.

```APIDOC
## GET /scheduler/queues/stats.json

### Description
Queue statistics in JSON format.

### Method
GET

### Endpoint
/scheduler/queues/stats.json

### Response
#### Success Response (200)
- **queues** (array) - List of queue statistics.
  - **name** (string) - Name of the queue.
  - **total_jobs** (integer) - Total number of jobs in the queue.
  - **queued** (integer) - Number of jobs currently queued.
  - **active** (integer) - Number of jobs currently active.
  - **scheduled** (integer) - Number of jobs scheduled for future execution.
  - **finished** (integer) - Number of jobs that have finished successfully.
  - **failed** (integer) - Number of jobs that have failed.
  - **canceled** (integer) - Number of jobs that have been canceled.
  - **workers_active** (integer) - Number of active workers for the queue.
  - **workers_idle** (integer) - Number of idle workers for the queue.
- **timestamp** (string) - Timestamp of when the statistics were generated.

### Request Example
```bash
curl http://localhost:8000/scheduler/queues/stats.json
```
```

--------------------------------

### Configure Scheduler Queues with Redis URL

Source: https://github.com/django-commons/django-tasks-scheduler/blob/master/_autodocs/configuration.md

Define a default queue using a Redis URL. Ensure the URL format is correct.

```python
SCHEDULER_QUEUES = {
    'default': QueueConfiguration(URL='redis://localhost:6379/0'),
}
```

--------------------------------

### Retrieve a Job by Name

Source: https://github.com/django-commons/django-tasks-scheduler/blob/master/_autodocs/api_reference_redis_models.md

Example of how to retrieve a specific job from Redis using its name and a connection. Checks if the job exists before accessing its status.

```python
from scheduler.redis_models import JobModel

job = JobModel.get('default:123:20250101000000000000', connection=redis_conn)
if job:
    print(f"Job status: {job.status}")
```

--------------------------------

### Create and Enqueue Job

Source: https://github.com/django-commons/django-tasks-scheduler/blob/master/_autodocs/api_reference_queue.md

Use this method to create a new job and add it to the queue. You can specify the function to run, arguments, scheduling time, timeouts, and callbacks. Jobs can be enqueued immediately or scheduled for a future time.

```python
def create_and_enqueue_job(
    self,
    func: FunctionReferenceType,
    args: Union[Tuple[Any, ...], List[Any], None] = None,
    kwargs: Optional[Dict[str, Any]] = None,
    when: Optional[datetime] = None,
    timeout: Optional[int] = None,
    result_ttl: Optional[int] = None,
    job_info_ttl: Optional[int] = None,
    description: Optional[str] = None,
    name: Optional[str] = None,
    at_front: bool = False,
    meta: Optional[Dict[str, Any]] = None,
    on_success: Optional[Callback] = None,
    on_failure: Optional[Callback] = None,
    on_stopped: Optional[Callback] = None,
    task_type: Optional[str] = None,
    scheduled_task_id: Optional[int] = None,
) -> JobModel
```

```python
from scheduler.helpers.queues import get_queue
from datetime import datetime, timedelta

queue = get_queue('default')

# Enqueue immediately
job = queue.create_and_enqueue_job(
    func='myapp.tasks.process_data',
    args=('data_file.csv',),
    kwargs={'format': 'csv'},
    timeout=600,
    result_ttl=3600
)
print(f"Job queued: {job.name}")

# Schedule for future execution
future_time = datetime.now() + timedelta(hours=2)
job = queue.create_and_enqueue_job(
    func='myapp.tasks.generate_report',
    when=future_time,
    timeout=300
)
```

--------------------------------

### Run Full Test Suite

Source: https://github.com/django-commons/django-tasks-scheduler/blob/master/CLAUDE.md

Execute the complete test suite for the scheduler, excluding multiprocess tests.

```bash
cd testproject/
uv sync --extra yaml

# Full test suite
uv run python manage.py test --exclude-tag multiprocess scheduler
```

--------------------------------

### Get Job Statistics

Source: https://github.com/django-commons/django-tasks-scheduler/blob/master/_autodocs/quick_start.md

Retrieve and display statistics about jobs, such as the total number of jobs and failed jobs. This is useful for monitoring job execution health.

```python
from scheduler.views import get_statistics

stats = get_statistics()
print(f"Total jobs: {stats['total_jobs']}")
print(f"Failed jobs: {stats['failed']}")
```

--------------------------------

### Create a Scheduled Task

Source: https://github.com/django-commons/django-tasks-scheduler/blob/master/_autodocs/SUMMARY.txt

Create a scheduled task by instantiating the Task model with task details like name, type, callable, cron string, and queue.

```python
from scheduler.models import Task

Task.objects.create(
    name='daily_job',
    task_type='CronTaskType',
    callable='myapp.tasks.run_daily',
    cron_string='0 * * * *',
    queue='default'
)
```

--------------------------------

### Get Formatted Task Callable String

Source: https://github.com/django-commons/django-tasks-scheduler/blob/master/_autodocs/api_reference_models.md

Generates a string representation of the task's callable function, including its arguments. This is useful for logging or debugging.

```python
function_string()
```

--------------------------------

### Configure Scheduler Queues with Username and Password

Source: https://github.com/django-commons/django-tasks-scheduler/blob/master/_autodocs/configuration.md

Set authentication credentials for Redis. Username support requires Redis 6.0+.

```python
SCHEDULER_QUEUES = {
    'default': QueueConfiguration(
        HOST='redis.example.com',
        PORT=6379,
        DB=0,
        USERNAME='app_user',
        PASSWORD='secure_password',
    ),
}
```

--------------------------------

### Conditional Callback Example

Source: https://github.com/django-commons/django-tasks-scheduler/blob/master/_autodocs/api_reference_callback.md

A callback that updates application status based on the job's result. It checks for specific conditions in the result before updating the database.

```python
from scheduler import job

def update_status_callback(job_model, connection, result):
    from myapp.models import Task as TaskModel
    
    # Update application state based on job result
    if isinstance(result, dict) and result.get('status') == 'ok':
        TaskModel.objects.filter(id=job_model.scheduled_task_id).update(
            status='completed'
        )

@job(queue='default', on_success=update_status_callback)
def validate_and_process():
    return {'status': 'ok', 'message': 'Validation passed'}
```

--------------------------------

### Set BROKER in SCHEDULER_CONFIG for Production and Testing

Source: https://github.com/django-commons/django-tasks-scheduler/blob/master/_autodocs/configuration.md

Specifies the broker library to use for task management. Use Broker.REDIS or Broker.VALKEY for production, and Broker.FAKEREDIS for in-memory testing without a Redis server.

```python
from scheduler.types import Broker, SchedulerConfiguration

# Production
SCHEDULER_CONFIG = SchedulerConfiguration(BROKER=Broker.REDIS)

# Testing
SCHEDULER_CONFIG = SchedulerConfiguration(BROKER=Broker.FAKEREDIS)
```

--------------------------------

### Register New Worker for Queues

Source: https://github.com/django-commons/django-tasks-scheduler/blob/master/docs/drt-model.md

This diagram illustrates the process of registering a new worker for specified queues. It includes checking queue existence and creating a worker key with associated information.

```mermaid
sequenceDiagram
    autonumber

        participant worker as WorkerProcess

        participant qlist as QueueHash<br/>name -> key
        participant wlist as WorkerList
        participant wkey as WorkerKey
        participant queue as QueueKey
        participant job as JobHash


    note over worker,qlist: Checking sanity

    break when a queue-name in the args is not in queue-list
        worker ->>+ qlist: Query queue names
        qlist -->>- worker: All queue names
        worker ->> worker: check that queue names exists in the system
    end

    note over worker,wkey: register
    worker ->> wkey: Create workerKey with all info (new id, queues, status)
    worker ->> wlist: Add new worker to list, last heartbeat set to now()
```

--------------------------------

### Get Jobs Ready to Schedule

Source: https://github.com/django-commons/django-tasks-scheduler/blob/master/_autodocs/api_reference_redis_models.md

Retrieves job names from the ScheduledJobRegistry that are ready to be executed at or before the given 'timestamp'. Used to move jobs from scheduled to active states.

```python
get_jobs_to_schedule(connection, timestamp)
```

--------------------------------

### Get Queue Instance by Name

Source: https://github.com/django-commons/django-tasks-scheduler/blob/master/_autodocs/api_reference_queue.md

Retrieves a Queue instance using its configured name from Django settings. This is useful for interacting with specific queues defined in SCHEDULER_QUEUES.

```python
from scheduler.helpers.queues import get_queue

def get_queue(queue_name: str) -> Queue:
```

```python
default_queue = get_queue('default')
high_priority = get_queue('high-priority')
```

--------------------------------

### Configure Scheduler Settings

Source: https://github.com/django-commons/django-tasks-scheduler/blob/master/docs/configuration.md

Set up default scheduler configurations including broker type, timeouts, and time-to-live for job results and information. This is the primary dictionary for customizing scheduler behavior.

```python
import os
from typing import Dict
from scheduler.types import SchedulerConfiguration, Broker, QueueConfiguration

SCHEDULER_CONFIG = SchedulerConfiguration(
    EXECUTIONS_IN_PAGE=20,
    SCHEDULER_INTERVAL=10,
    BROKER=Broker.REDIS,
    CALLBACK_TIMEOUT=60,  # Callback timeout in seconds (success/failure/stopped)
    # Default values, can be overridden per task/job
    DEFAULT_SUCCESS_TTL=10 * 60,  # Time To Live (TTL) in seconds to keep successful job results
    DEFAULT_FAILURE_TTL=365 * 24 * 60 * 60,  # Time To Live (TTL) in seconds to keep job failure information
    DEFAULT_JOB_TTL=10 * 60,  # Time To Live (TTL) in seconds to keep job information
    DEFAULT_JOB_TIMEOUT=5 * 60,  # timeout (seconds) for a job
    # General configuration values
    DEFAULT_WORKER_TTL=10 * 60,  # Time To Live (TTL) in seconds to keep worker information after last heartbeat
    DEFAULT_MAINTENANCE_TASK_INTERVAL=10 * 60,  # The interval to run maintenance tasks in seconds. 10 minutes.
    DEFAULT_JOB_MONITORING_INTERVAL=30,  # The interval to monitor jobs in seconds.
    SCHEDULER_FALLBACK_PERIOD_SECS=120,  # Period (secs) to wait before requiring to reacquire locks
)
SCHEDULER_QUEUES: Dict[str, QueueConfiguration] = {
    'default': QueueConfiguration(
        HOST='localhost',
        PORT=6379,
        USERNAME='some-user',
        PASSWORD='some-password',
        CONNECTION_KWARGS={
            'ssl_cert_reqs': 'required',
            'ssl': True,
        },
    ),
    'high': QueueConfiguration(URL=os.getenv('REDISTOGO_URL', 'redis://localhost:6379/0')),
    'low': QueueConfiguration(HOST='localhost', PORT=6379, DB=0, ASYNC=False),
}
```

--------------------------------

### List All Active Workers

Source: https://github.com/django-commons/django-tasks-scheduler/blob/master/_autodocs/endpoints.md

Use this endpoint to get a list of all currently active workers in the scheduler. The response is an HTML page detailing worker status and assignments.

```bash
curl http://localhost:8000/scheduler/workers/
```

--------------------------------

### Export Task Configuration to Dictionary

Source: https://github.com/django-commons/django-tasks-scheduler/blob/master/_autodocs/api_reference_models.md

Serializes the task's configuration into a dictionary, suitable for backups or inter-process communication. Includes all relevant settings.

```python
task_dict = task.to_dict()
import json
with open('task_backup.json', 'w') as f:
    json.dump(task_dict, f, default=str)
```

--------------------------------

### Chained Jobs Callback Example

Source: https://github.com/django-commons/django-tasks-scheduler/blob/master/_autodocs/api_reference_callback.md

A callback that triggers the next job in a pipeline upon successful completion of the current job. This is useful for creating task sequences.

```python
from scheduler import job
from scheduler.helpers.queues import get_queue

def trigger_next_stage(job_model, connection, result):
    """On success, enqueue the next job in pipeline"""
    queue = get_queue(job_model.queue_name)
    next_job = queue.create_and_enqueue_job(
        'myapp.tasks.process_result',
        args=(result,),
    )
    print(f"Enqueued next job: {next_job.name}")

@job(queue='default', on_success=trigger_next_stage)
def stage_one():
    return "Stage 1 complete"

@job(queue='default')
def process_result(data):
    return f"Processed: {data}"
```

--------------------------------

### Configure Scheduler Settings

Source: https://github.com/django-commons/django-tasks-scheduler/blob/master/_autodocs/SUMMARY.txt

Configure the scheduler in your Django settings.py file, specifying broker and queue details.

```python
from scheduler.types import SchedulerConfiguration, QueueConfiguration, Broker

INSTALLED_APPS = ['scheduler', ...]

SCHEDULER_CONFIG = SchedulerConfiguration(BROKER=Broker.REDIS)
SCHEDULER_QUEUES = {'default': QueueConfiguration(URL='redis://localhost:6379/0')}
```

--------------------------------

### Configure Scheduler Queues with Unix Socket Path

Source: https://github.com/django-commons/django-tasks-scheduler/blob/master/_autodocs/configuration.md

Connect to Redis using a Unix domain socket by specifying the socket file path.

```python
SCHEDULER_QUEUES = {
    'default': QueueConfiguration(UNIX_SOCKET_PATH='/tmp/redis.sock'),
}
```

--------------------------------

### QueuedJobRegistry Methods

Source: https://github.com/django-commons/django-tasks-scheduler/blob/master/_autodocs/api_reference_redis_models.md

Details the specific methods for the QueuedJobRegistry.

```APIDOC
### QueuedJobRegistry

Jobs waiting for execution.

#### enqueue

Add job to the queue.

#### Method

```python
def enqueue(connection, job_name)
```

#### dequeue

Remove and return the first job from the queue.

#### Method

```python
def dequeue(connection)
```

#### compact

Remove expired entries from the registry.

#### Method

```python
def compact(connection)
```
```

--------------------------------

### Manage Systemd Workers

Source: https://github.com/django-commons/django-tasks-scheduler/blob/master/docs/usage.md

Commands to reload systemd settings and manage scheduler worker services. Use these to start, stop, or manage individual or multiple worker instances.

```shell
sudo systemctl daemon-reload
sudo systemctl start scheduler_worker@{1..3}
```

```shell
sudo systemctl stop scheduler_worker@2
```

--------------------------------

### Create and Run Worker Programmatically

Source: https://github.com/django-commons/django-tasks-scheduler/blob/master/_autodocs/quick_start.md

Instantiate and run a worker programmatically, specifying the queues it should listen to. This offers more control than command-line arguments.

```python
from scheduler.worker import create_worker

worker = create_worker(['default', 'high-priority'])
worker.work()
```

--------------------------------

### Environment Variables for Scheduler Configuration

Source: https://github.com/django-commons/django-tasks-scheduler/blob/master/_autodocs/configuration.md

Load scheduler configuration from environment variables. Ensures default values if variables are not set.

```python
import os

SCHEDULER_CONFIG = SchedulerConfiguration(
    SCHEDULER_INTERVAL=int(os.getenv('SCHEDULER_INTERVAL', '10')),
    DEFAULT_JOB_TIMEOUT=int(os.getenv('DEFAULT_JOB_TIMEOUT', '300')),
)

SCHEDULER_QUEUES = {
    'default': QueueConfiguration(
        URL=os.getenv('REDIS_URL', 'redis://localhost:6379/0'),
    ),
}
```

--------------------------------

### Get Job Detail

Source: https://github.com/django-commons/django-tasks-scheduler/blob/master/_autodocs/endpoints.md

Fetch detailed information about a specific job using its full identifier. The response includes job status, timing, configuration, and results or errors.

```bash
curl http://localhost:8000/scheduler/jobs/default:42:20250115123456000000/
```

--------------------------------

### Create a Worker Instance

Source: https://github.com/django-commons/django-tasks-scheduler/blob/master/_autodocs/api_reference_worker.md

Use this factory function to create and configure a Worker. Provide a single queue name, a list of queue names, or a Queue object. Optional parameters include a worker name and additional Worker constructor arguments.

```python
from scheduler.worker import create_worker

def create_worker(
    queues: Union[str, List[str], List[Queue], Queue],
    worker_name: Optional[str] = None,
    **kwargs
) -> Worker:
```

```python
from scheduler.worker import create_worker

# Simple usage
worker = create_worker(['default', 'high'])

# With options
worker = create_worker(
    queues=['default'],
    worker_name='worker-1',
    fork_job_execution=True,
    burst=False
)

# Start worker
worker.work()
```

--------------------------------

### Lint and Format Code

Source: https://github.com/django-commons/django-tasks-scheduler/blob/master/CLAUDE.md

Check code for style issues and automatically format it using Ruff and MyPy.

```bash
ruff check --fix
ruff format
mypy scheduler/
```

--------------------------------

### Worker request_stop() Method Example

Source: https://github.com/django-commons/django-tasks-scheduler/blob/master/_autodocs/api_reference_worker.md

Demonstrates how to request a graceful shutdown of a worker instance within a signal handler. This ensures the worker finishes its current job before exiting.

```python
# In a signal handler
def handle_sigterm(signum, frame):
    worker.request_stop()

signal.signal(signal.SIGTERM, handle_sigterm)
```

--------------------------------

### Configure FakeRedis for Testing

Source: https://github.com/django-commons/django-tasks-scheduler/blob/master/_autodocs/quick_start.md

Use FakeRedis for testing to avoid actual Redis connections. Configure this in your test settings file.

```python
# settings_test.py
from scheduler.types import SchedulerConfiguration, Broker

SCHEDULER_CONFIG = SchedulerConfiguration(
    BROKER=Broker.FAKEREDIS,
    DEFAULT_JOB_TIMEOUT=30,
)
```