Bazis-BG

Background task package for the Bazis framework. Provides an asynchronous task management system with scheduler support, handlers, periodic tasks, and specialized classes for data import/export.

Quick Start

# Install package
uv add bazis-bg

# Create task
from bazis.contrib.bg.basic.base import BgBase

class MyTask(BgBase):
    name = 'My Task'
    
    def handle(self):
        self.next_phase('Processing', expected=100)
        for i in range(100):
            self.progress(d_count=10)

# Run task
task = MyTask.delay()

# Start scheduler
python manage.py bg_scheduler

Table of Contents

• Description
• Core Concept
• Features
• Requirements
• Installation
• Architecture
• Usage
  • Creating Basic Task
  • Running Tasks
  • Periodic Tasks
  • Data Export
  • Data Import
• Task API
• Configuration
• Task Management
• Examples
• Development
• Contributing
• License
• Links

Description

Bazis-BG is an extension for the Bazis framework that provides a powerful background task management system. The package is built on top of Django ORM and provides a distributed task processing system with support for:

• Asynchronous task execution
• Task scheduler (cron-like)
• Multiple handlers
• Progress tracking
• Specialized classes for data import/export
• Management through Django admin panel

Core Concept

The background task system in Bazis-BG consists of three main components:

1. Scheduler (bg_scheduler) - monitors tasks in the database and manages handlers
2. Handlers (bg_handler) - execute tasks in separate processes
3. Tasks - classes inheriting from BgBase containing business logic

When an application calls Task.delay(), a record is created in the database. The scheduler discovers the new task and passes it to a free handler. The handler executes the task in an isolated process, updating status and progress in the database.

This approach provides:

• Task execution isolation
• Scalability through adding handlers
• Fault tolerance - tasks are not lost if a handler crashes
• Transparency - complete information about execution state

Features

• Distributed processing: Multiple handlers for parallel task execution
• Scheduler: Support for periodic tasks with cron syntax and intervals
• Progress tracking: Detailed information about task execution state
• Execution history: Saving execution phases with timestamps
• Logging: Built-in logging system for each task
• Task interruption: Ability to interrupt a running task
• Transactions: Support for transactional execution
• Files: Working with files (upload/download)
• Specialized classes:
  • BgBaseDownload — data export to Excel
  • BgBaseLoad — data import from Excel/ZIP
  • BgBaseDownloadModel — Django model export
• Healthcheck: Monitoring scheduler and handler health

Advantages Over Other Solutions

1. Django Integration: Uses Django ORM, migrations, and admin
2. Ease of Use: Minimal code to create tasks
3. Transparency: Complete execution information through admin
4. Specialized Classes: Ready solutions for import/export
5. Distribution: Easy scaling through adding handlers

Requirements

• Python: 3.12+
• Bazis: base package must be installed
• PostgreSQL: 12+
• Redis: for caching
• Dependencies:
  • openpyxl — working with Excel files
  • psutil — process management
  • crontab — parsing cron expressions
  • aiohttp — asynchronous HTTP requests

Note! The current framework implementation requires Redis as the cache backend and PostgreSQL as the database.

Installation

Using uv (recommended)

uv add bazis-bg

Using pip

pip install bazis-bg

For Development

git clone git@github.com:ecofuture-tech/bazis-bg.git
cd bazis-bg
uv sync --dev

Project Setup

1. Add application to Django settings:

INSTALLED_APPS = [
    # ...
    'bazis.contrib.bg',
    # ...
]

2. Apply migrations:

python manage.py migrate

3. Configure environment variables:

# Scheduler
BS_BAZIS_TASK_SCHEDULER_HOST=localhost
BS_BAZIS_TASK_SCHEDULER_PORT=49001
BS_BAZIS_TASK_SCHEDULER_IDLE=0.2

# Handlers
BS_BAZIS_TASK_HANDLERS_LOCAL=5
BS_BAZIS_TASK_HANDLER_PORT=49100
BS_BAZIS_TASK_HANDLER_IDLE=0.5

# Storage
BS_BAZIS_TASK_FOLDER=bg

Architecture

System Components

┌─────────────────┐
│   Application   │ ──> delay() ──> Create Task
└─────────────────┘
         │
         ▼
┌─────────────────┐
│   Scheduler     │ ──> Monitor tasks
│  (bg_scheduler) │ ──> Start handlers
└─────────────────┘
         │
         ▼
┌─────────────────┐
│    Handlers     │ ──> Execute Task
│  (bg_handler)   │ ──> Update status
└─────────────────┘

Models

• Task — main task model
• TaskCron — periodic task model
• TaskHandler — handler model

Task States

1. draft — Draft
2. waiting — Waiting to start
3. starting — Starting
4. running — Running
5. done — Completed

Package Structure

bazis.contrib.bg/
├── basic/
│   ├── base.py              # Base BgBase class
│   ├── base_download.py     # Data export class
│   ├── base_download_model.py  # Django model export
│   └── base_load.py         # Data import class
├── models_abstract.py       # Abstract models
├── models.py                # Concrete models
├── admin.py                 # Django admin
└── management/
    └── commands/
        ├── bg_scheduler.py  # Scheduler command
        └── bg_handler.py    # Handler command

Usage

Creating Basic Task

from bazis.contrib.bg.basic.base import BgBase

class DataProcessingTask(BgBase):
    # Human-readable task name
    name = 'Data Processing'
    
    # Number of simultaneously running tasks
    parallel = 2
    
    # Tasks that block execution
    blocked = ['myapp.tasks.AnotherTask']
    
    # Execute in transaction
    is_transaction = True
    
    def pre_handle(self):
        """Called before handle()"""
        self.log.info('Preliminary preparation')
    
    def handle(self):
        """Main task logic"""
        # Set phase with expectation
        self.next_phase('Loading data', expected=1000)
        
        for i in range(1000):
            # Your logic
            
            # Update progress
            self.progress(
                performed=i+1,  # Performed
                d_time=3,       # Sync every 3 sec
                d_count=100     # Or every 100 operations
            )
        
        # Next phase
        self.next_phase('Saving results', expected=100)
        
        # ... save logic
        
        # Save result
        self.set_result({'processed': 1000, 'saved': 100})
    
    def post_handle(self):
        """Called after handle()"""
        self.log.info('Finalization')
    
    def excepting(self):
        """Exception handling"""
        self.log.error('An error occurred')
    
    def interrupting(self):
        """Interruption handling"""
        self.log.warning('Task interrupted by user')
    
    def finishing(self):
        """Always executed at the end"""
        self.log.info('Resource cleanup')

Running Tasks

Asynchronous Launch

# Basic launch
task = DataProcessingTask.delay()

# With parameters
task = DataProcessingTask.delay(
    param1='value1',
    param2='value2'
)

# With custom name
task = DataProcessingTask.delay(
    name='January data processing',
    param1='value1'
)

# With file
with open('data.xlsx', 'rb') as fp:
    task = DataProcessingTask.delay(
        fp=fp,
        param1='value1'
    )

# With author
task = DataProcessingTask.delay(
    author=request.user,
    param1='value1'
)

envs= of delay() is stored with the task (Task.envs) and not applied to the process that runs it: a task reads its settings itself.

Synchronous Launch

# Waits for task completion
task = DataProcessingTask.delay_sync(param1='value1')

Periodic Tasks

Declared in the code

The periodic tasks of a product are declared in the schedules.py module of an application; migrate creates or updates them (TaskCron, marked managed), takes over an unmarked one of the same slug with a warning and disables a managed one no longer declared:

# myapp/schedules.py
from django.utils.translation import gettext_lazy as _
from bazis.contrib.bg.declare import Schedule

SCHEDULES = [
    Schedule('data_processing', 'myapp.tasks.DataProcessingTask', '0 3 * * *',
             _('Data processing'), kwargs={'param1': 'value1'}),
]

Do not create them with get_or_create in a data migration: it runs once, so a new period in the code never reaches a database that has the row. To move such a periodic task (or one made in the admin) to schedules.py, delete its old row (it has no slug) and let the declaration own the schedule, otherwise both rows run the task: in the admin, or with a data migration that depends on the old one and on ('bg', '0006_taskcron_slug_managed') (apps.get_model('bg', 'TaskCron').objects.filter(cls_path='myapp.tasks.DataProcessingTask', slug=None).delete()). The tasks of the old row are kept, no longer linked to a periodic task.

Cleanup

bazis-bg declares its own periodic task, bg_tasks_clean: every BAZIS_TASK_CRON_CLEANUP seconds (3600) it deletes the manual tasks not updated for BAZIS_TASK_LIFETIME seconds (90 days) and the periodic ones not updated for BAZIS_TASK_CRON_LIFETIME (6 hours), with their files. BS_BAZIS_TASK_CLEANUP=false turns it off.

A managed periodic task no longer declared is disabled and becomes an ordinary one, which the admin may enable again (the roles of bazis-permit stay managed instead). The name is stored translated into LANGUAGE_CODE. The system checks bg.E001 and bg.E002 validate the declarations; with a database (bazis_doctor) bg.W001 reports a database that differs from them and bg.W002 an enabled periodic task that is not managed and runs a declared schedule again. The admin shows a declared periodic task read-only.

Through Django Admin

For periodic tasks that are data of an installation rather than of the product:

1. Open "Periodic Tasks" section
2. Create new task:
   • Name: Task description
   • Task Class: Full path (e.g., myapp.tasks.DataProcessingTask)
   • Period: Cron expression or seconds
   • Args/Kwargs: JSON with parameters
   • Enabled: Yes

Period Examples

# Every 60 seconds
period = "60"

# Every day at 03:00
period = "0 3 * * *"

# Every Monday at 09:00
period = "0 9 * * 1"

# Every hour
period = "0 * * * *"

# Every 5 minutes
period = "*/5 * * * *"

Data Export

Base Export Class

from bazis.contrib.bg.basic.base_download import BgBaseDownload

class ReportDownload(BgBaseDownload):
    name = 'Report Export'
    file_name = 'report'
    
    # Fields to export
    fields_read = ['id', 'name', 'created_at', 'status']
    
    # Fields marked bold in header
    fields_marked = {'name', 'status'}
    
    # Column sizes
    fields_size = {
        'id': 10,
        'name': 30,
        'created_at': 20,
        'status': 15
    }
    
    # Row limit per file (for splitting into archive)
    limit_rows_split = 500000
    
    def get_titles(self):
        """Returns column headers"""
        return {
            'id': 'ID',
            'name': 'Name',
            'created_at': 'Created At',
            'status': 'Status'
        }
    
    def get_queryset(self):
        """Returns data for export"""
        return MyModel.objects.all().values(*self.fields_read)
    
    def get_count(self):
        """Returns number of records"""
        return MyModel.objects.count()

Export with Grouped Headers

def get_titles(self):
    return {
        'groups': [
            {
                'name_union': 'Basic Information',
                'columns': ['id', 'name'],
                'marked': True
            },
            {
                'name_union': 'Dates',
                'columns': ['created_at', 'updated_at']
            }
        ],
        'titles': {
            'id': 'ID',
            'name': 'Name',
            'created_at': 'Created',
            'updated_at': 'Updated'
        }
    }

Django Model Export

from bazis.contrib.bg.basic.base_download_model import BgBaseDownloadModel
from django.apps import apps

class VehicleDownload(BgBaseDownloadModel):
    model = apps.get_model('myapp.Vehicle')
    fields_read = ['id', 'gnum', 'vehicle_model__brand__name', 'country__name']
    
    # verbose_name of fields will be picked up automatically

The file of a task is stored as bg/<random token>_<name> and downloaded by the author of the task (or staff) from GET /api/v1/bg/task/{id}/download/.

Data Import

from bazis.contrib.bg.basic.base_load import BgBaseLoad

class DataImport(BgBaseLoad):
    name = 'Data Import'
    
    # Fields in column order
    fields_read = ['name', 'price', 'quantity']
    
    # Skip first row (headers)
    skip_titles = True
    
    # Progress update period
    progress_count = 1000
    
    # Track row grouping
    groups_check = False
    
    # Add meta-data (_meta)
    with_meta = False
    
    def load(self, reader):
        """
        Process loaded data
        
        reader - generator of dictionaries with data
        """
        self.next_phase('Saving data')
        
        for row in reader:
            # row = {'name': '...', 'price': '...', 'quantity': '...'}
            
            # Create object
            MyModel.objects.create(
                name=row['name'],
                price=float(row['price']),
                quantity=int(row['quantity'])
            )
            
            self.progress(d_count=100)

Task API

Task Management Methods

# Get task object
task = self.get_task()

# Update task data
self.task_update(
    phase='New phase',
    expected=1000,
    performed=500
)

# Save file in task
with open('result.xlsx', 'rb') as fp:
    self.task_file_save('result.xlsx', fp, file_params={'rows': 1000})

# Get file from task
fp = self.task_file_get()

# Set result
self.set_result({'status': 'success', 'count': 1000})

# Force synchronization
self.flush()

Working with Phases

# Simple phase
self.next_phase('Processing')

# Phase with expectation (number)
self.next_phase('Loading files', expected=100)

# Phase with expectation (dictionary)
self.next_phase('Processing', expected={
    'files': 10,
    'records': 1000
})

# Update progress
self.progress(
    performed=50,           # Performed
    expected=100,          # Expected (can be updated)
    d_time=5,              # Sync every 5 sec
    d_count=100,           # Or every 100 operations
    need_sync=True         # Force synchronization
)

# Progress with dictionary
self.progress(
    performed={'files': 5, 'records': 500},
    d_count=10
)

Logging

# Use self.log to write to task log
self.log.info('Information message')
self.log.warning('Warning')
self.log.error('Error')
self.log.debug('Debug information')

# Logs are automatically saved in task.log field

Following a Task from a Client

An action that starts work queues the task with author=user and answers task_accepted(task): 202 Accepted with {"data": {"type": "bg.task", "id": "<id>"}} and Location: /api/v1/bg/task/<id>/. The client polls that URL until state is done.

from bazis.contrib.bg.routes import task_accepted
from bazis.core.routes_abstract.initial import http_post
from bazis.core.schemas.enums import RouteKind

class ReportRouteSet(UserRequiredRouteBase):
    @http_post('/calc/', status_code=202, kind=RouteKind.OTHER)
    def action_calc(self, **kwargs):
        return task_accepted(CalcData.delay(author=self.inject.user))

The fields of GET /api/v1/bg/task/{id}/:

• state: waiting, starting, running, done. A done task succeeded when error is empty and interrupt is false.
• phase, expected, performed: the current phase and its progress (performed from its first sync). When a phase ends it is moved to phases_history as {phase, duration, expected, performed} and expected/performed are set to null: show the current phase from them while the task runs and the finished phases from phases_history. Once the task is done both are null.
• error: for the author of the task, Error when it failed (null or empty otherwise); staff see the traceback. log is shown to staff only (null for the author). To show the cause to the author, write a message of your own to the result of the task (set_result).
• result: what the task passed to set_result; the file is downloaded from GET /api/v1/bg/task/{id}/download/.

Configuration

Environment Variables

# Scheduler
BS_BAZIS_TASK_SCHEDULER_HOST=localhost  # Scheduler host
BS_BAZIS_TASK_SCHEDULER_PORT=49001      # Scheduler healthcheck port
BS_BAZIS_TASK_SCHEDULER_IDLE=0.2        # Delay between iterations, sec

# Handlers
BS_BAZIS_TASK_HANDLERS_LOCAL=5          # Number of local handlers
BS_BAZIS_TASK_HANDLER_PORT=49100        # Base port for handlers
BS_BAZIS_TASK_HANDLER_IDLE=0.5          # Delay between task checks, sec

# Storage
BS_BAZIS_TASK_FOLDER=bg                 # Folder for task files
BS_BAZIS_STORAGE_BG=None                # Storage class (default FileSystemStorage)

Django Settings

# settings.py

# Number of local handlers
BAZIS_TASK_HANDLERS_LOCAL = 3

# Base port for handlers
BAZIS_TASK_HANDLER_PORT = 8100

# Scheduler task check interval (seconds)
BAZIS_TASK_SCHEDULER_IDLE = 1

# Handler task check interval (seconds)
BAZIS_TASK_HANDLER_IDLE = 1

# Custom storage for task files
BAZIS_STORAGE_BG = 'myapp.storage.S3Storage'

Task Management

Through Django Admin

1. Background Tasks (Task)

   • View all tasks
   • Filter by state, date
   • Interrupt running tasks
   • View logs and results
   • Download result files

2. Periodic Tasks (TaskCron)

   • Create/edit periodic tasks
   • Enable/disable
   • View next run time

3. Handlers (TaskHandler)

   • View active handlers
   • Monitor load

Programmatically

from django.apps import apps

Task = apps.get_model('bg.Task')

# Get task
task = Task.objects.get(id=task_id)

# Interrupt task
task.interrupt = True
task.save()

# Check state
if task.is_done:
    if task.is_success:
        print('Successfully completed')
    elif task.is_error:
        print(f'Error: {task.error}')
    elif task.is_interrupt:
        print('Interrupted')

# Get result
result = task.result

# Download file
if task.file:
    with task.file.open('rb') as fp:
        data = fp.read()

Running Services

# Start scheduler
python manage.py bg_scheduler

# Start handler manually
python manage.py bg_handler --host 127.0.0.1 --port 8100

# In production, it's recommended to use supervisor or systemd

Examples

Example 1: Simple Task with Progress

from bazis.contrib.bg.basic.base import BgBase

class SimpleTask(BgBase):
    name = 'Simple Task'
    
    def handle(self):
        self.next_phase('Processing', expected=100)
        
        for i in range(100):
            # Your logic
            self.log.info(f'Processing item {i+1}')
            self.progress(d_count=10)

# Launch
task = SimpleTask.delay()

Example 2: Task with File

from bazis.contrib.bg.basic.base import BgBase

class FileProcessingTask(BgBase):
    name = 'File Processing'
    
    def handle(self):
        # Get file
        fp = self.task_file_get()
        if not fp:
            raise Exception('File not provided')
        
        self.next_phase('Reading file')
        data = fp.read()
        
        # Processing...
        result = self.process_data(data)
        
        # Save result
        from io import BytesIO
        result_fp = BytesIO(result)
        self.task_file_save('result.txt', result_fp)
    
    def process_data(self, data):
        # Your processing logic
        return data.upper()

# Launch with file
with open('input.txt', 'rb') as fp:
    task = FileProcessingTask.delay(fp=fp)

Example 3: Report Export

from bazis.contrib.bg.basic.base_download_model import BgBaseDownloadModel
from django.apps import apps

class MonthlyReport(BgBaseDownloadModel):
    model = apps.get_model('myapp.Order')
    fields_read = ['id', 'number', 'customer__name', 'total', 'created_at']
    fields_marked = {'total'}
    file_name = 'monthly_report'

# Launch
task = MonthlyReport.delay()

# Get file after completion
from django.apps import apps
Task = apps.get_model('bg.Task')
task = Task.objects.get(id=task.id)
if task.is_success:
    file_url = task.file.url

Example 4: Data Import

from bazis.contrib.bg.basic.base_load import BgBaseLoad
from django.apps import apps

class ProductImport(BgBaseLoad):
    name = 'Product Import'
    fields_read = ['name', 'sku', 'price', 'quantity']
    skip_titles = True
    
    def load(self, reader):
        Product = apps.get_model('myapp.Product')
        
        self.next_phase('Importing products')
        
        created = 0
        updated = 0
        
        for row in reader:
            product, is_created = Product.objects.update_or_create(
                sku=row['sku'],
                defaults={
                    'name': row['name'],
                    'price': row['price'],
                    'quantity': row['quantity']
                }
            )
            
            if is_created:
                created += 1
            else:
                updated += 1
            
            self.progress(d_count=100)
        
        self.set_result({'created': created, 'updated': updated})

# Launch with file
with open('products.xlsx', 'rb') as fp:
    task = ProductImport.delay(fp=fp)

Example 5: Task with Parameters

from bazis.contrib.bg.basic.base import BgBase
from django.apps import apps

class SendEmailTask(BgBase):
    name = 'Send Email'
    parallel = 5  # Maximum 5 simultaneous sends
    
    def handle(self, email_list, subject, message):
        """
        Args:
            email_list: list of email addresses
            subject: email subject
            message: email body
        """
        self.next_phase('Sending emails', expected=len(email_list))
        
        for i, email in enumerate(email_list):
            try:
                # Send email
                self.send_email(email, subject, message)
                self.log.info(f'Sent to {email}')
            except Exception as e:
                self.log.error(f'Error sending to {email}: {e}')
            
            self.progress(performed=i+1, d_count=10)
    
    def send_email(self, email, subject, message):
        from django.core.mail import send_mail
        send_mail(subject, message, 'noreply@example.com', [email])

# Launch
emails = ['user1@example.com', 'user2@example.com']
task = SendEmailTask.delay(
    email_list=emails,
    subject='Important notification',
    message='Message text'
)

Example 6: Task with Locks

from bazis.contrib.bg.basic.base import BgBase

class DataExportTask(BgBase):
    name = 'Data Export'
    parallel = 1  # Only one task simultaneously
    blocked = [
        'myapp.tasks.DataImportTask',  # Blocked by import
        'myapp.tasks.DataUpdateTask'   # Blocked by update
    ]
    
    def handle(self):
        self.next_phase('Exporting data')
        # Export logic
        pass

class DataImportTask(BgBase):
    name = 'Data Import'
    parallel = 1
    blocked = [
        'myapp.tasks.DataExportTask',
        'myapp.tasks.DataUpdateTask'
    ]
    
    def handle(self):
        self.next_phase('Importing data')
        # Import logic
        pass

Development

Setting Up Development Environment

# Clone repository
git clone git@github.com:ecofuture-tech/bazis-bg.git
cd bazis-bg

# Install dependencies
uv sync --dev

# Run tests
pytest

# Check code
ruff check .

# Format code
ruff format .

Test Structure

tests/
├── test_base.py              # Base class tests
├── test_download.py          # Export tests
├── test_load.py              # Import tests
├── test_scheduler.py         # Scheduler tests
└── test_handler.py           # Handler tests

Contributing

We welcome your contributions! Here's how you can help:

1. Fork the repository
2. Create a branch for new functionality (git checkout -b feature/amazing-feature)
3. Make changes
4. Run tests (pytest)
5. Commit changes (git commit -m 'Add amazing feature')
6. Push to branch (git push origin feature/amazing-feature)
7. Open Pull Request

Please ensure that:

• Tests are written for new functionality
• Documentation is updated
• Existing code style is maintained
• Changes are added to changelog

License

Apache License 2.0

See LICENSE file for details.

Links

• Bazis Core — framework base package
• Issue Tracker — report bug or request feature
• Bazis Documentation — main documentation

Support

If you have questions or problems:

• Check documentation
• Search in existing issues
• Create new issue with detailed description

Package Ecosystem

Bazis supports extensions through additional packages:

• bazis — framework core
• bazis-bg — background task system
• bazis-test-utils — testing utilities
• bazis-<n> — other extensions (add bazis- prefix to name)

All extension packages require installation of the base bazis package.

---

Made with ❤️ by Bazis team