#1157: Writing custom automation plugins with Python

Table of Contents

Part I. Fundamentals

1. What you can program and which mechanism to choose.
2. How the execution environment works.
3. A minimal script and the Runner class.
4. Parameters, data types, and additional context.
5. Your first working script: loading, web form, and verifying the result.

Part II. Launch Triggers

6. Event-driven execution and working with event.
7. Scheduled execution and business hours.

Part III. Working with Data

8. Finding objects and related data.
9. Tickets, comments, and chats.
10. Custom fields and the assignment counter.
11. Repeated runs and parallel processing.
12. Bulk operations, transactions, and dry-run preview.

Part IV. Actions and Integrations

13. Notifications, templates, and recipient selection.
14. CSV and Excel reports.
15. External APIs and paginated fetching.
16. Importing users, companies, and articles.

Part V. Operations

17. Diagnostics, performance, and operations.
18. Example catalog and further learning path.

The scripting engine lets you write and run arbitrary Python code to power automations triggered by events, on a schedule, or even manually by a user through a web form.

Administrators can upload scripts and configure their triggers through the Web Form, Event Actions, and Schedule Actions in the Settings > Automation and Routing > Scripts section.

Script plugins let your team build server-side logic for support processes: check conditions, work with tickets and related objects, generate reports, run bulk operations, and exchange data with external systems.

You write a Python file, upload it to Swarmica, and choose how it is launched: by event, by schedule, or through a web form. The same script can serve different scenarios through parameters.

This guide is intended for a developer or administrator with basic Python knowledge. It explains the mechanism and programming techniques based on the examples attached to the knowledge base.

The scenarios and examples below may be incomplete and do not cover every capability of Swarmica script plugins. If you cannot find the method you need, do not understand how to implement your scenario, or cannot get a script to work, contact Swarmica support with your question. Describe what you want to do, specify the system version, and attach your code and the error message, if any. We will help you understand the available mechanisms and implementation approaches.

Versions and verification. This material was prepared against the documentation and files available on September 17, 2026. Internal models and methods may differ between versions. You should test the training examples in a test installation of your Swarmica version before running them in production. The VERSION label inside a file denotes the version of the script itself, not the minimum Swarmica version.

Notational Conventions in This Guide

• Django — the standard framework. Such methods and modules are described in the official Django documentation.
• Swarmica — the product's internal models, helpers, and service objects. Their signatures may change between versions; verify them against your test installation and working examples from the articles.
• Unless stated otherwise, obtain all variables (ticket, group, schedule, channel) before the line containing the example.

Part I. Fundamentals

1. What You Can Program

A custom extension is useful when the desired behavior requires code: additional checks, computations, work with multiple objects, or handling a response from an external system.

For example, a Python script can count ticket assignments and write the result to a field, find unanswered requests, generate a report file, or import data from CSV.

Web widgets are a separate mechanism for extending the interface. Their setup is described in the article “Custom widgets in Swarmica”. This guide focuses on server-side Python code.

2. Execution Environment

The script runs inside the Swarmica installation and can import the product's models and internal functions. Therefore, accessing local data does not require HTTP requests to your own API: the examples use the Django ORM.

This is a map of imports from the files we examined, not a promise of an unchanging API. Descriptions of many objects are available in the article “Objects and parameters for triggers”.

Permissions and Resources

The documentation states that scripts run as root inside the container, have access to the main database and shared file volumes, and share resources with Swarmica. Container isolation does not protect product data from buggy code.

Scripts run inside a container, so they have no access to the host machine or the virtual machine where Swarmica is installed. However, scripts do have access to the shared disk volumes and the main database used by Swarmica, so be careful when performing potentially dangerous operations such as deleting or modifying any data or files.

Additionally, memory and CPU resources are shared with the entire Swarmica installation, so it is strongly recommended to minimize library and module imports to only the minimum set of functions you actually need. For example, instead of importing the whole module with import datetime, import only the function you need: from datetime import timedelta.

Assigning a role for web form access does not turn a direct ORM query into a user query with automatic visibility checks for all objects. If different employees may launch the form, restrictions on accessible data and actions must be enforced in the implementation.

For changes, use a test installation, a limited sample, and a backup. Do not log passwords, tokens, or the entire launch context. These requirements are especially important for imports and bulk operations.

The terms of use for the mechanism are described in the Python plugin documentation. For script plugins, the published examples list the Premium tier or higher; see the pricing page for current terms.

Libraries

The documentation lists, in particular, requests, openpyxl, pandas, and bs4, along with tools for working with JSON, CSV, dates, and regular expressions. Import only what you need. Verify that any given third-party library is available and its version in your own installation.

A number of pre-installed libraries are available for import and use in your scripts. See the list of available methods in the documentation of the respective library:

import re               # For working with regular expressions
import os               # For OS-related things such as file paths, etc.
import bs4              # For parsing HTML, XML, etc. with BeautifulSoup
import csv              # For reading and writing CSV
import json             # For parsing JSON
import time             # For working with the current time, e.g. generating timestamps
import pandas           # For working with datasets
import urllib           # For parsing and writing URL parameters, urlencode, etc.
import dateutil         # For working with dates, e.g. parsing a date from a string
import datetime         # For working with datetime objects, e.g. computing durations, etc.
import openpyxl         # For reading and writing Excel
import requests         # For working with remote APIs and webhooks

Engine Overview

There are several ways to launch a script:

[Event trigger] ----> +context --\
                                  \
[Schedule] ---------> +context----\
                                    >----> Runner.run(kwargs={data: **params, **context})
[CLI] --------------> +params-----/
                                  /
[API/WEB] ----------> +params----/

So, in general, if a runtime script is launched by an Event Trigger or by a Schedule, then the event data and the custom context data configured in the corresponding trigger or schedule are passed to the script's parameters.

Suppose the script is launched by a UserEvent; then the context can be parsed as follows:

class Runner(BaseRunner):
    def run(self, *args, **kwargs):
        data = kwargs.get('data', {})
        if not data or 'event_id' not in data:
            logger.error(f"No event_id passed, exiting")
            return
        event = UserEvent.objects.filter(id=data['event_id']).first()
        if not event:
            logger.error(f"No event with {data['event_id']} found, exiting")
            return

If we configure the Event Trigger to pass additional context into the script — for example, a list of email addresses to notify when a user account changes — that context will also be passed to the script.

Assuming the Event Trigger's custom context looks like this:

{
  "recipients": [
    "user1@domain.tld",
    "user2@domain.tld"
  ]
}

You can now access the variable's value inside the script as follows:

data = kwargs.get('data', None)

if data:
    recipients = data.get('recipients', None)

Accessing Swarmica Objects

You will usually need to manipulate models and querysets for Swarmica objects in your scripts. Since Swarmica uses Django as its framework, most model and queryset methods work the same way here.

See the complete reference for Django model and queryset methods on the official documentation site:

Querysets

Models

Most objects you may need are in the core package, except for User objects — those are in the swarmica_auth package.

You will also typically need to import the shared constants used by Swarmica, like this:

from swarmica import defs

print(defs.USER_ROLES_INTERNAL)

3. A Minimal Script and the Runner Class

Every script is a Python file that declares a Runner class inherited from runtime.runners.BaseRunner and a run method. Everything the script does lives in run or in functions it calls.

A minimal standalone file:

import logging

from runtime.runners import BaseRunner

logger = logging.getLogger(__name__)


class Runner(BaseRunner):
    def run(self, *args, **kwargs):
        data = kwargs.get("data") or {}
        logger.info(f"Script started; parameter keys: {list(data)}")

This script does not read from the database and does not modify data. It simply writes to the log the keys that arrived in data. It is useful as a “Hello, world” and as a first connectivity check: launch it, look at the log, and confirm that parameters are arriving.

The engine calls the run method. Do the main work inside it or in functions it calls. Do not put database queries, network calls, or data modifications at module import level: they may run before a specific launch is processed — including the very first time the file is connected.

What Is Available from Runner

The examples we examined use the following:

The full set of BaseRunner methods is not described here. The presence of the listed attributes is confirmed by the attached files, but this is not a complete contract for the runtime environment.

Handling Errors

Handle expected parameter errors with explicit checks. For unexpected exceptions, use logger.exception(...) to preserve the traceback. Do not hide every error behind a blanket except: pass.

What happens to an exception that leaves run depends on the launch method and version. Do not assume the environment will automatically retry the launch or show the user a meaningful message: verify this in a test installation.

4. Parameters and Context

Three Data Sources

For example, additional context:

{
  "group_uid": "GROUP_UID",
  "hours": 48,
  "recipients": ["support-lead@example.com"],
  "apply": false
}

A Python fragment inside run:

data = kwargs.get("data") or {}
group_uid = data.get("group_uid")
hours = data.get("hours", 48)
recipients = data.get("recipients", [])

This lets you change the group, threshold, and recipients without editing the code. A variable does not become a parameter automatically: the script must explicitly read it and use it.

Web Form with Dates

You can also configure a script so that it has its own web form that can accept, validate, and pass parameters into the script.

Suppose we write a script that generates a custom Excel report and sends it to a list of email recipients. We want users to provide the report's start and end dates and a comma-separated list of recipients.

In the Swarmica web interface, in the script settings, we configure the following web form parameters:

[
  {
    "name": "start_date",
    "type": "datetime",
    "readonly": false,
    "displayName": "Report start date"
  },
  {
    "name": "end_date",
    "type": "datetime",
    "readonly": false,
    "displayName": "Report end date"
  },
  {
    "name": "recipients",
    "type": "text",
    "displayName": "Comma-separated recipients"
  }
]

We can now access these parameters inside the script as follows:

params = kwargs.get('data', {})
if params:
    if 'recipients' in params:
        recipients = params.get('recipients')
        if recipients:
            recipients = [x.strip() for x in recipients.split(',') if '@' in x]
        if not recipients:
            logger.info("Input error: empty/invalid list of 'recipients' specified")
            return
        
        start_date = params.get('start_date', None)
        if isinstance(start_date, str) and start_date.strip():
            report_start_date = parse_datetime(start_date)
            
        end_date = params.get('end_date', None)
        if isinstance(end_date, str) and end_date.strip():
            report_end_date = parse_datetime(end_date)

        if not report_end_date:
            report_end_date = now().replace(hour=23, minute=30, second=0, microsecond=0).replace(tzinfo=default_time_zone)
        if not report_start_date:
            report_start_date = report_end_date - timedelta(days=1)

Distinguish JSON from Python

In the settings, the context is provided as valid JSON: true, false, null, double quotes, no comments. In Python, use True, False, None.

Do not put entries like ['email'], <HOURS>, or # comment into JSON. Replace the GROUP_UID placeholder in the examples with a real UID.

The additional context must contain JSON-compatible values. At the same time, the environment may add a model object event to the launch data: the resulting data dictionary is then no longer required to be JSON-serializable.

Validate Types

A text field passes a string. A date may need conversion. In one of the manual-launch examples, handling is provided for a single value that arrived as a list or tuple. Normalize such forms only for specific scalar parameters; a list of recipients or channels must remain a list.

For a boolean parameter, do not use bool("false"): a non-empty string will give True. The examples use the Swarmica helper common.utils.str2bool. You can also use your own strict function:

def parse_apply(value):
    if value is None:
        return False
    if isinstance(value, bool):
        return value
    if isinstance(value, str):
        normalized = value.strip().lower()
        if normalized in {"true", "1", "yes"}:
            return True
        if normalized in {"false", "0", "no", ""}:
            return False
    raise ValueError("apply must be a boolean value")

An unknown value should never accidentally permit data modification.

Parameter Names

Do not use event and event_id for your own settings: those names are reserved for event data. Choose clear names for your own parameters: group_uid, cf_uid, reminder_hours, recipient.

General explanation of context: article 1351. Python-event specifics: article 1614.

5. Your First Working Script: Loading, Web Form, and Verifying the Result

This is the next step after “Hello, world”: we read data from the database, change nothing, and write the result to the log.

Task. Find a group by UID and show a limited list of new and open tickets.
Trigger. Web form.
Parameters. group_uid, limit (1–100).
Expected result. Entries like ticket_id=… status=… assignee_uid=… in the log and a final line shown=N.

Step 1. Create the Python File

Save the following standalone example as read_group_tickets.py.

import logging

from core.models import Group, Ticket
from runtime.runners import BaseRunner
from swarmica import defs

logger = logging.getLogger(__name__)
VERSION = "1.0.0"


class Runner(BaseRunner):
    def run(self, *args, **kwargs):
        LOG_PREFIX = f"{self.script.name} ({self.script.uid})"
        data = kwargs.get("data") or {}
        if not isinstance(data, dict):
            logger.error(f"{LOG_PREFIX}: data must be a dictionary")
            return

        group_uid = data.get("group_uid")
        if not isinstance(group_uid, str) or not group_uid.strip():
            logger.error(f"{LOG_PREFIX}: specify group_uid")
            return

        raw_limit = data.get("limit", 10)
        if isinstance(raw_limit, bool) or not isinstance(raw_limit, (int, str)):
            logger.error(f"{LOG_PREFIX}: limit must be an integer")
            return
        try:
            limit = int(raw_limit)
        except ValueError:
            logger.error(f"{LOG_PREFIX}: limit must be an integer")
            return
        if not 1 <= limit <= 100:
            logger.error(f"{LOG_PREFIX}: limit must be between 1 and 100")
            return

        group = Group.objects.filter(uid=group_uid.strip()).first()
        if group is None:
            logger.warning(f"{LOG_PREFIX}: group not found")
            return

        tickets = Ticket.objects.filter(
            group=group,
            status__in=[defs.TICKET_STATUSES_NEW, defs.TICKET_STATUSES_OPEN],
        ).select_related("assignee").order_by("id")[:limit]

        shown = 0
        for ticket in tickets:
            assignee_uid = ticket.assignee.uid if ticket.assignee else None
            logger.info(
                f"{LOG_PREFIX}: ticket_id={ticket.id} status={ticket.status} "
                f"assignee_uid={assignee_uid}"
            )
            shown += 1
        logger.info(f"{LOG_PREFIX}: version={VERSION} shown={shown}")

What matters here for later:

• LOG_PREFIX = f"{self.script.name} ({self.script.uid})" — copy this technique into every script. Having the script name on every log line makes searches easier when several extensions are enabled.
• Parameter validation is not paranoia. A form text field always arrives as a string, and int("") or bool can break the logic.
• If there are no tickets, the script will complete normally with shown=0. That is fine, not an error.
• select_related("assignee") eliminates N+1 queries: without it, each ticket.assignee is a separate SQL query.

Step 2. Upload the File

1. Sign in to Swarmica as an administrator.
2. Open Settings → Scripts.
3. Create a script and give it a name.
4. For the first test, allow only administrators to launch it.
5. Upload read_group_tickets.py into the script field.
6. Save the settings.

Step 3. Add a Web Form

In the web form parameters, enter:

[
  {
    "name": "group_uid",
    "type": "string",
    "required": true,
    "readonly": false,
    "displayName": "Group UID"
  },
  {
    "name": "limit",
    "type": "string",
    "required": true,
    "readonly": false,
    "displayName": "How many tickets to show: 1 to 100"
  }
]

name is the name the code reads via data.get(...). displayName is the label shown to the user. In this example, the limit comes from a text field and is converted to a number by the script itself.

The forms we examined also include the types boolean, date, datetime, and dropdown. A dropdown list requires choices variants; the resulting value must match what the code expects. Verify the exact set of types and the format of variants against the interface of your version and the group change example.

Step 4. Launch and Check the Log

In the web form tab, enter a real group UID and a limit of 10, then submit the form.

The script's execution result can be viewed in the celeryworker log:

docker logs swarmica-celeryworker-1 -f --tail 100

The script interface also provides testing tools, whose composition depends on the version. A return statement in the code does not by itself mean that the web form will show the user a result: in this training example, the result is in the log.

Part II. Launch Triggers

6. Event-Driven Execution and Working with event

The Event Object and the Event Code

The word event is used at two levels:

Events of different entities use different models: TicketEvent, UserEvent, ArticleEvent, CustomFieldEvent, KCSEvent. You cannot look up a user event in the ticket events table just because the ID happens to match.

Field names inside events can differ between models. In some events the initiator is responsible, in others it is user or actor. Before accessing a field, verify it against the model of your version (see the _meta.get_fields() technique in section 17).

Getting the Event

The following standalone file logs the fact that the assignee changed. It supports both a passed-in object and loading by ID.

import logging

from core.models import TicketEvent
from runtime.runners import BaseRunner
from swarmica import defs

logger = logging.getLogger(__name__)


class Runner(BaseRunner):
    def run(self, *args, **kwargs):
        LOG_PREFIX = f"{self.script.name} ({self.script.uid})"
        data = kwargs.get("data") or {}
        event = data.get("event")
        if event is not None and not isinstance(event, TicketEvent):
            logger.error(f"{LOG_PREFIX}: expected a TicketEvent object")
            return
        if event is None:
            event_id = data.get("event_id")
            if event_id is None:
                logger.warning(f"{LOG_PREFIX}: neither event nor event_id was passed")
                return
            if isinstance(event_id, bool) or not isinstance(event_id, (int, str)):
                logger.warning(f"{LOG_PREFIX}: invalid event_id")
                return
            try:
                event_id = int(event_id)
            except ValueError:
                logger.warning(f"{LOG_PREFIX}: invalid event_id")
                return
            event = TicketEvent.objects.filter(id=event_id).first()
        if event is None:
            logger.warning(f"{LOG_PREFIX}: event not found")
            return
        if event.event != defs.TICKET_EVENTS_ASSIGNEE_CHANGE:
            logger.info(f"{LOG_PREFIX}: event {event.id} skipped: different type")
            return

        ticket = event.ticket
        logger.info(
            f"{LOG_PREFIX}: assignee change: "
            f"event_id={event.id} ticket_id={ticket.id}"
        )

Configuring the Trigger

1. Create an Event Action.
2. Select the TicketEvent model and a “Script” action type.
3. Specify the connected script.
4. Configure the event “Assignee change”.
5. If needed, set additional conditions and context.
6. Enable the action and test it on a test ticket.

In Python, use the named constant defs.TICKET_EVENTS_ASSIGNEE_CHANGE. The codes for configuration and other events are given in article 913.

Old and New Values

The examples mention old, new, old_value, and new_value. Names available as object attributes or as UI condition fields are not necessarily ORM filter fields.

For example, the assignment counter uses the ORM condition new__isnull=False, while the pending processing example converts a string status into a numeric event value through defs.TICKET_EVENTS_STATUSES. Do not plug a ticket's string status into any event field by analogy.

First check the model description and a working example for your version. With delayed processing, also keep in mind: the event describes a past change, while event.ticket may already hold the ticket's current state. Before acting, re-check the necessary conditions.

7. Scheduled Execution and Business Hours

Configuration

1. Upload the Python file.
2. Create a Schedule Action of type “Script”.
3. Select the file and set the schedule.
4. Provide any additional parameters the code reads.
5. Enable the action after testing on a small sample.

The published reminders example uses * * * * * — a launch every minute. Verify other values and the execution time zone against your own configuration and the crontab reference.

A periodic launch by itself does not pass any specific ticket. The sample must be built inside run. Do not expect event just because the script class matches an event-driven example.

Calendar and Business Time

For a calendar threshold, the fragment looks like this:

from datetime import timedelta
from django.utils import timezone

cutoff = timezone.now() - timedelta(hours=48)

That is 48 regular hours. For business time, you need to use the schedule and exclude non-working intervals. The files we examined call Swarmica methods Schedule.business_time_duration(...), schedule.tz, schedule.is_business_hours(), and methods for building business intervals. These are internal methods; verify them on the target version.

Greeting Example

In trigger_new_chat_greeting.py, a single second from the ticket's creation moment is checked:

from datetime import timedelta

start = ticket.created_at
end = start + timedelta(seconds=1)
is_working_time = schedule.business_time_duration(start=start, end=end).total_seconds() > 0

The variables ticket and schedule must be obtained beforehand. This checks the creation moment, not the current launch time.

Pending Processing Example

The file recurring_autonotify_autosolve.py demonstrates a composite process:

1. Find tickets in PENDING with a schedule on the requester's company.
2. Find the moment of the last transition into PENDING.
3. Check whether the customer replied after that moment.
4. If there is no automatic comment yet — compute the reminder deadline.
5. If there is one comment — compute the deadline for transitioning to SOLVED.
6. Perform the action when the deadline arrives and the current time is within business hours.

The two intervals are set by reminder_hours and close_hours parameters; both default to one hour.

Specifics of the attached file: source_channels is used as a list of UIDs: source__uid__in=source_channels. Pass real channel UIDs. If you need filtering by type, you will have to change the query; the string email must not automatically be treated as a UID.

The main sample excludes tickets without a company schedule, although the helper functions contain variants that work without one. This distinction matters when adapting: the presence of a branch in a function does not mean the main loop will hand it such tickets.

The file also counts all automatic API comments by the service user after the transition to PENDING. Another automation may affect that counter. In your own implementation, use a process-specific marker.

Guide: article 1819. Before running in production, check weekends, holidays, the end-of-business-day transition, the absence of intervals, and a parallel customer reply.

Part III. Working with Data

8. Finding Objects and Related Data

Django models are available inside a script. The fragments below are for insertion into your logic, not standalone files.

Finding a Single Object

from core.models import Group, Ticket

group = Group.objects.filter(uid="GROUP_UID").first()
ticket = Ticket.objects.filter(id=123).first()

If the object is not found, first() returns None. The get(...) method requires exactly one result and may raise an exception — use it only when the object's absence is an error and you are prepared to handle it.

Ticket Sample

from core.models import Ticket
from swarmica import defs

tickets = Ticket.objects.filter(
    status__in=[defs.TICKET_STATUSES_NEW, defs.TICKET_STATUSES_OPEN],
    group__uid="GROUP_UID",
).order_by("id")

The double underscore lets you traverse to a related object or set a condition: group__uid, created_at__gte, body__icontains. Fetch only the sample you need and limit it for early testing.

Ticket Sample Filtered by Dates

Let us print the ticket ID, subject, and assignee name for all new and open tickets created between the specified start and end dates:

from swarmica import defs
from core.models import Ticket

class Runner(BaseRunner):
    def run(self, *args, **kwargs):
        params = kwargs.get('data', {})
        if not params:
            return
            
        start_date = params.get('start_date', None)
        
        if isinstance(start_date, str) and start_date.strip():
            report_start_date = parse_datetime(start_date)
        end_date = params.get('end_date', None)
        
        if isinstance(end_date, str) and end_date.strip():
            report_end_date = parse_datetime(end_date)

        if not report_end_date:
            report_end_date = now().replace(hour=23, minute=30, second=0, microsecond=0).replace(tzinfo=default_time_zone)
        if not report_start_date:
            report_start_date = report_end_date - timedelta(days=1)
            
        tickets = Ticket.objects.filter(created_at__range=(report_start_date,report_end_date), 
                                        status__in=[defs.TICKET_STATUSES_NEW, defs.TICKET_STATUSES_OPEN])
        
        for ticket in ticket:
            print(f"#{ticket.id}: {ticket.subject} ({ticket.assignee.name})")

Working with Users

Let us disable all email notifications for blocked users:

from swarmica import defs
from swarmica_auth.models import User

class Runner(BaseRunner):
    def run(self, *args, **kwargs):
        
        users = User.objects.filter(role=defs.USER_ROLES_BLOCKED)
        
        for user in users:
            user.email_notifications.clear()

Optimizing Relations

• select_related("assignee") — for ForeignKey and OneToOne. Eliminates N+1 because it performs a JOIN.
• prefetch_related("comments") — for ManyToMany and reverse ForeignKey. Performs a separate query and stitches the result together in Python.

Without these calls, accessing a related object inside a loop performs a separate SQL query on every iteration. For large samples, this is the main source of slow scripts.

For very large datasets, use iterator(chunk_size=...) — it reads rows in batches and does not hold the entire result in memory:

for ticket in Ticket.objects.filter(...).iterator(chunk_size=500):
    ...

Checking for a Comment

Fragment after obtaining the ticket variable:

has_comment = ticket.comments.filter(
    public=False,
    body__icontains="requires review",
).exists()

public=False restricts the search to internal comments. icontains performs a case-insensitive search.

Latest Record

Fragment for a ticket's history:

last_event = ticket.events.order_by("-date", "-id").first()

With descending order, first() returns the latest record and last() returns the earliest. The additional ID sort makes the choice more deterministic when dates are equal.

Related Objects May Be Missing

requester = ticket.requester
organization = requester.organization if requester else None
schedule = organization.schedule if organization else None

Do not assume that every ticket has a company, schedule, or assignee. For frequently used relations, the examples use select_related(...); for large samples — iterator(chunk_size=...). General methods are described in the Django QuerySet reference.

9. Tickets, Comments, and Chats

Changing Status

Fragment after obtaining the ticket:

from swarmica import defs

ticket.status = defs.TICKET_STATUSES_SOLVED
ticket.save()

TICKET_STATUSES_SOLVED is the “Solution provided” status, not TICKET_STATUSES_CLOSED. The text of any notification must match the actual status the code sets.

After save(), model handlers, events, and automations may run. Check their impact on your scenario.

About save(update_fields=[...]). If you restrict the write to specific fields (ticket.save(update_fields=["status"])), Django will not call pre_save/post_save for the other fields, and Swarmica automations may not fire the same way as with a full save(). This is not a bug, but the behavior differs from a normal save. Verify it on a test ticket before using update_fields.

A single call to save() does not prove equivalence of any change to all actions of the standard interface.

Creating an Internal Automatic Comment

Fragment with the ticket variable:

from core.models import TicketComment
from swarmica_auth.models import User
from swarmica import defs

author = User.objects.get_swarmica_service()
TicketComment.objects.create(
    ticket=ticket,
    author=author,
    responsible=author,
    body="<p>Additional review completed.</p>",
    public=False,
    is_staff=True,
    is_autocomment=True,
    source=defs.TICKET_COMMENT_SOURCES_API,
)

User.objects.get_swarmica_service() is a Swarmica helper, not Django. It may raise an exception if the Swarmica service user is not configured (by default it is created during Swarmica installation). In a production script, wrap it in try/except and log a clear reason.

The attributes is_staff, is_autocomment, and source appear in the examples we examined. For a public reply, use public=True. Delivery to the customer depends on channel settings and comment processing — creating a record should not be described as a universal guarantee of delivery to every channel.

If the text contains values from a user or an external system, escape them before inserting into HTML. A template with allowed HTML markup and plain text are different types of input.

Full Example: Adding a Note Based on Review Result

Task. When a ticket transitions to SOLVED, check internal comments and add an automatic note.
Trigger. Event “Status changed → Solution provided”.
Parameters. Three strings: what to search for and what to write in each of the two outcomes.
Expected result. An internal non-public comment with a marker of the processed event; re-running the same event does not create a second note.

This training variant, inspired by article 1830, additionally verifies the ticket's state and re-processing of the same event.

Upload add_check_note.py and set the context:

{
  "text": "requires review",
  "comment_if_text": "The ticket contains a request for additional review.",
  "comment_ifnot_text": "No request for additional review was found."
}

import logging
from html import escape

from django.db import transaction
from core.models import Ticket, TicketComment, TicketEvent
from runtime.runners import BaseRunner
from swarmica import defs
from swarmica_auth.models import User

logger = logging.getLogger(__name__)


class Runner(BaseRunner):
    def run(self, *args, **kwargs):
        LOG_PREFIX = f"{self.script.name} ({self.script.uid})"
        data = kwargs.get("data") or {}
        event = data.get("event")
        if not isinstance(event, TicketEvent):
            logger.warning(f"{LOG_PREFIX}: pass a TicketEvent object via an event action")
            return
        if event.event != defs.TICKET_EVENTS_STATUS_CHANGE:
            return

        names = ("text", "comment_if_text", "comment_ifnot_text")
        if not all(isinstance(data.get(k), str) and data[k].strip() for k in names):
            logger.error(f"{LOG_PREFIX}: provide non-empty strings: {', '.join(names)}")
            return
        marker = f"<!-- custom-check-note:event:{event.id} -->"
        author = User.objects.get_swarmica_service()

        with transaction.atomic():
            ticket = Ticket.objects.select_for_update().filter(id=event.ticket_id).first()
            if ticket is None or ticket.status != defs.TICKET_STATUSES_SOLVED:
                logger.info(f"{LOG_PREFIX}: ticket state changed; action skipped")
                return
            if ticket.comments.filter(author=author, public=False, body__contains=marker).exists():
                logger.info(f"{LOG_PREFIX}: event {event.id} has already been processed")
                return

            found = ticket.comments.filter(
                public=False,
                is_autocomment=False,
                body__icontains=data["text"].strip(),
            ).exists()
            message = data["comment_if_text"] if found else data["comment_ifnot_text"]
            body = "<p>" + escape(message.strip()).replace("\n", "<br/>") + "</p>" + marker
            TicketComment.objects.create(
                ticket=ticket, author=author, responsible=author,
                body=body, public=False, is_staff=True, is_autocomment=True,
                source=defs.TICKET_COMMENT_SOURCES_API,
            )
        logger.info(
            f"{LOG_PREFIX}: note added: "
            f"ticket_id={event.ticket_id} event_id={event.id}"
        )

In this variant, the search excludes automatic comments so that your own notes do not affect subsequent checks. The marker protects against repeating the same event when a comment is saved; if you remove it, you need to restore the protection in another way.

Web Chat Message

The greeting example uses a different model and binds to a chat session. Fragment inside Runner, after obtaining the ticket:

from core.models import ChatMessage

session = ticket.chat_sessions.last()
if session is not None:
    ChatMessage.objects.create(
        session=session,
        public=True,
        author=self.service_user,
        body="Hello! Please describe your question.",
    )

To select the latest session, the ordering of the related set must match what your version expects. The original file uses last() without explicit ordering.

To choose a greeting based on schedule, the file trigger_new_chat_greeting.py reads schedule_name, message, and out_of_schedule_message; the texts are dictionaries keyed by language. An existing schedule and a matching requester language are required. Detailed setup: article 1274.

For empty chats, the close_empty_chat.py example first checks for any comment, then creates a public comment and moves the ticket to SOLVED. The presence of even a service comment can stop such a scenario. The channel, status, and time conditions are set by the trigger; the code does not replace their configuration. See article 664.

10. Custom Fields and the Assignment Counter

Reading

After obtaining the ticket object:

value = ticket.custom_fields.get_value("CUSTOM_FIELD_UID")

custom_fields.get_value(...) is a Swarmica helper, not Django ORM. Use the UID of your own installation's field, not a value from someone else's example. Check None separately from zero and False if they have different meanings.

Writing

In the assignment counter example, the value is stored via CustomFieldValue. Fragment with the ticket variable:

from custom_field.models import CustomField, CustomFieldValue
from django.contrib.contenttypes.models import ContentType
from core.models import Ticket

field = CustomField.objects.get(uid="CUSTOM_FIELD_UID")
content_type = ContentType.objects.get_for_model(Ticket)
CustomFieldValue.objects.update_or_create(
    field=field,
    content_type=content_type,
    object_id=ticket.id,
    defaults={"data": {"value": 7}},
)

ContentType and update_or_create are standard Django. CustomField / CustomFieldValue are Swarmica models.

Important note about defaults={"data": {...}}. update_or_create completely overwrites data. If a field stores not only value but also other keys (for example, rich-text metadata or attachment IDs), they will be lost. Before writing, check the actual content of data for your field type and, if needed, merge the dictionaries.

This example applies to a numeric ticket field. For menus, lists, and other types, the same value format is not implied. Confirm the field's format and constraints before writing.

Full Example: Ticket Assignment Count

Task. Recompute the number of assignments of a ticket to a staff member and write it into a custom field.
Trigger. Event “Assignee change”.
Parameters. cf_uid — UID of an integer ticket field.
Expected result. The field contains the current number; reprocessing the same event does not increase the counter.

The original file from article 1333 recomputes assignment events with a non-empty new value. This is more robust than simply incrementing by one on each run: re-processing the same event will not increment the number again.

Metric definition: all assignments to a staff member are counted, including the first one. Unassigning to “Nobody” is not counted. This is not a ready-made handover rate and not a count of unique engineers. If the business needs transfers between staff specifically, a different event sample is needed.

Create an integer ticket field and specify its UID in the additional context:

{
  "cf_uid": "CUSTOM_FIELD_UID"
}

Training standalone file update_assignment_count.py:

import logging

from core.models import Ticket, TicketEvent
from custom_field.models import CustomField, CustomFieldValue
from django.contrib.contenttypes.models import ContentType
from django.db import transaction
from runtime.runners import BaseRunner
from swarmica import defs

logger = logging.getLogger(__name__)


class Runner(BaseRunner):
    def run(self, *args, **kwargs):
        LOG_PREFIX = f"{self.script.name} ({self.script.uid})"
        data = kwargs.get("data") or {}
        event = data.get("event")
        if not isinstance(event, TicketEvent):
            logger.warning(f"{LOG_PREFIX}: pass a TicketEvent")
            return
        if event.event != defs.TICKET_EVENTS_ASSIGNEE_CHANGE:
            return
        cf_uid = data.get("cf_uid")
        if not isinstance(cf_uid, str) or not cf_uid.strip():
            logger.error(f"{LOG_PREFIX}: cf_uid not specified")
            return
        field = CustomField.objects.filter(uid=cf_uid.strip()).first()
        if field is None:
            logger.error(f"{LOG_PREFIX}: field not found")
            return

        content_type = ContentType.objects.get_for_model(Ticket)
        with transaction.atomic():
            ticket = Ticket.objects.select_for_update().filter(id=event.ticket_id).first()
            if ticket is None:
                return
            count = TicketEvent.objects.filter(
                ticket=ticket,
                event=defs.TICKET_EVENTS_ASSIGNEE_CHANGE,
                new__isnull=False,
            ).count()
            CustomFieldValue.objects.update_or_create(
                field=field, content_type=content_type, object_id=ticket.id,
                defaults={"data": {"value": count}},
            )
        logger.info(f"{LOG_PREFIX}: ticket_id={event.ticket_id} assignments_count={count}")

Before running, verify that the field is intended for the ticket and has an integer type.

About locking. select_for_update() here locks the Ticket row against concurrent modification by other launches of the same script. It is important to understand what the lock does not do:

• it does not prevent concurrent insertion of new TicketEvent records;
• it does not protect the field's value from being written by another code path (another script, a manual edit in the interface);

The final value will be correct only as long as all field modifications go through this same script with the same lock. As soon as a second writer appears, agree on a common rule or use a separate coordination mechanism.

Guide and original file: “How to calculate Handover Rate”.

11. Repeated Runs and Parallel Processing

Do not assume that an event will always be processed exactly once, or that scheduled tasks will never overlap. Even a manual re-run can produce duplicates.

The check “there is no comment yet” and the subsequent write must happen consistently, otherwise two processes will see the absence of a comment at the same time. The ticket lock in the training note example helps only if all instances of the process use the same scheme.

Other logic may need a separate state record or a unique constraint. A single transaction.atomic() without a lock and without a unique key does not guarantee the absence of duplicates.

Protecting Against Loops

A script can generate an event that it is itself triggered by, or that triggers another automation. For example, a comment creates a comment event; a field change creates a field event.

Restrict by event type, check the author and the marker, and only change values that differ. The assignment metric should be triggered by assignment, not by any ticket change. Test the chain in a test installation.

12. Bulk Operations, Transactions, and Dry-Run Preview

Required Steps

For an operation over a list of objects:

1. Parse and validate the parameters.
2. Fetch all objects and report the missing IDs.
3. Check allowed statuses and other constraints.
4. Compute what will change.
5. Perform the changes using the chosen method.
6. Log the summary: found, changed, skipped, errors.

The original set_group_for_tickets.py implements this approach: it reads tickets and group_uid, checks that all tickets exist and have status SOLVED / CLOSED, and then changes the group only where it differs.

Transaction

You can combine related changes inside transaction.atomic(). If an exception leaves the block, database changes are rolled back. External email or HTTP requests are not undone by such a rollback. See the Django documentation on transactions.

Fragment with predefined ticket_id and group:

from django.db import transaction
from core.models import Ticket

with transaction.atomic():
    ticket = Ticket.objects.select_for_update().get(id=ticket_id)
    if ticket.group_id != group.id:
        ticket.group = group
        ticket.save()

select_for_update() coordinates changes to a single row between transactions with a compatible locking scheme. Network calls inside a long transaction delay the release of locks, so plan the order of work separately.

save vs. Bulk Writes

save() and bulk_update() differ. The original group change example uses bulk_update: that method does not call each model's save() and does not fire the standard pre_save / post_save signals. Therefore, it is not assumed to create the same events as a manual operation in the interface. See the bulk_update section of the Django docs.

Separately: save(update_fields=[...]) is another case. It saves only the listed fields, and pre_save/post_save for the other fields will not fire. Swarmica automations tied to changes of other fields may not be triggered.

Choose the write method based on the desired behavior, not just speed. Similarly, verify what is required when modifying a custom field directly via CustomFieldValue.

apply and Dry-Run Mode

For administrative actions, an apply parameter defaulting to false is useful. This is a design technique for your own script; the engine does not add a dry-run preview automatically.

The standalone example preview_group_move.py uses only a group change and writes nothing by default:

import logging

from core.models import Group, Ticket
from django.db import transaction
from runtime.runners import BaseRunner
from swarmica import defs

logger = logging.getLogger(__name__)


def parse_apply(value):
    if value is None:
        return False
    if isinstance(value, bool):
        return value
    if isinstance(value, str):
        value = value.strip().lower()
        if value in {"true", "1", "yes"}:
            return True
        if value in {"false", "0", "no", ""}:
            return False
    raise ValueError("Invalid apply")


class Runner(BaseRunner):
    def run(self, *args, **kwargs):
        LOG_PREFIX = f"{self.script.name} ({self.script.uid})"
        data = kwargs.get("data") or {}
        try:
            apply = parse_apply(data.get("apply"))
            raw_ids = data.get("tickets", "")
            if not isinstance(raw_ids, str):
                raise ValueError("tickets must be a string")
            ticket_ids = sorted({int(x.strip()) for x in raw_ids.split(",") if x.strip()})
            if not ticket_ids or len(ticket_ids) > 100 or any(x <= 0 for x in ticket_ids):
                raise ValueError("Specify between 1 and 100 positive IDs")
            group_uid = data.get("group_uid")
            if not isinstance(group_uid, str) or not group_uid.strip():
                raise ValueError("Specify group_uid")
        except ValueError as error:
            logger.error(f"{LOG_PREFIX}: parameter error: {error}")
            return

        group = Group.objects.filter(uid=group_uid.strip()).first()
        if group is None:
            logger.error(f"{LOG_PREFIX}: group not found")
            return

        with transaction.atomic():
            tickets = list(Ticket.objects.select_for_update().filter(id__in=ticket_ids).order_by("id"))
            if {t.id for t in tickets} != set(ticket_ids):
                logger.error(f"{LOG_PREFIX}: some tickets not found; no changes made")
                return
            allowed = {defs.TICKET_STATUSES_SOLVED, defs.TICKET_STATUSES_CLOSED}
            if any(t.status not in allowed for t in tickets):
                logger.error(f"{LOG_PREFIX}: only SOLVED and CLOSED are allowed; no changes made")
                return
            changes = [t for t in tickets if t.group_id != group.id]
            logger.info(
                f"{LOG_PREFIX}: found={len(tickets)} "
                f"to_change={len(changes)} apply={apply}"
            )
            if not apply:
                return
            for ticket in changes:
                ticket.group = group
                ticket.save()
        logger.info(f"{LOG_PREFIX}: operation completed")

Context for a test launch:

{
  "tickets": "101, 102",
  "group_uid": "GROUP_UID",
  "apply": false
}

For a web form, add tickets and group_uid as string types and apply as a boolean type. Setting apply to true permits writes. The code deliberately uses save() instead of the bulk_update() from the original file; the consequences for events and automations must be verified separately.

A dry-run preview does not freeze the state forever: data may change between the preview and the apply. That is why the checks are repeated on each run. Original guide for this operation: article 1887.

Part IV. Actions and Integrations

13. Notifications, Templates, and Recipient Selection

Choosing an Email Channel

The report examples use the standard outgoing channel:

from core.models import EmailChannel

channel = EmailChannel.objects.filter(default=True, is_deleted=False).first()

EmailChannel is a Swarmica model. If there is no channel, or send_from is not set, the script must exit with a clear message. For a custom SMTP connection, use channel.get_smtp_connection(); for the system relay, the examples pass connection=None.

Sending an Email

Fragment after checking the channel and recipients:

from core.tasks.notifications import send_email

send_email(
    from_email=channel.send_from,
    to=["support-lead@example.com"],
    subject="Additional review result",
    text="The review is complete. Details are available in the ticket.",
    connection=None if channel.use_system_smtp_relay else channel.get_smtp_connection(),
)

send_email is a Swarmica helper from core.tasks.notifications. This is the signature from the attached examples, not a complete reference. A log line “sent” in the script does not prove delivery to the mailbox: verify the mail-sending result and the channel settings.

Jinja HTML Template

The template engine in the files is obtained via engines["jinja2"]. Fragment with the ticket variable:

from django.template import engines
from core.models import Instance

template = engines["jinja2"].from_string(
    "<p>Ticket #{{ ticket.id }}. Status: {{ ticket.status }}</p>"
)
html_body = template.render(context={
    "instance": Instance.get_solo(),
    "ticket": ticket,
})

engines is Django. Instance.get_solo() is a Swarmica helper.

In the email, HTML can be passed via html=html_body together with a plain-text version. Verify the escaping of dynamic values and the behavior of the chosen template. The existing base template notifications/base_email.html is also used in the source files.

Recipients Depend on the Event Reason

For CSAT and other classified events, it is convenient to provide a dictionary of “reason → recipients” in the context. An anonymized example of the settings:

{
  "reason_map": {
    "productbug": ["product-owner@example.com"],
    "documentation": ["docs-owner@example.com", "support-lead@example.com"]
  },
  "survey_score_threshold": 79
}

Training function for selecting recipients:

def recipients_for_reasons(selected_reasons, reason_map):
    if not isinstance(reason_map, dict):
        raise ValueError("reason_map must be a dictionary")
    if not isinstance(selected_reasons, (list, tuple)):
        raise ValueError("Reasons must be a list")
    recipients = set()
    for reason in selected_reasons:
        addresses = reason_map.get(reason, [])
        if not isinstance(addresses, list):
            raise ValueError("Recipients for each reason must be a list")
        for address in addresses:
            if not isinstance(address, str) or not address.strip():
                raise ValueError("Each recipient must be a non-empty string")
            recipients.add(address.strip())
    return sorted(recipients)

The function only selects addresses: it does not validate emails, does not find the survey, and does not send the email. Those steps must be added separately.

The CSAT example shows how to link a TicketEvent, CSATSurvey, custom fields, and an email template. The score in the analyzed file is compared against a threshold on a 0–100 scale; do not transfer the number from a five-point display directly. The survey must correspond to the specific event being processed — selecting the latest record for the ticket and author may be insufficient for delayed processing.

Notifications About Unanswered Requests

The file recurring_ticket_group_notify_noanswer.py demonstrates group selection, retrieval of its members' emails, analysis of the ticket's latest events, and sending an email. In it, LIMIT_HOURS, CHECK_GROUP, and MAIL_FROM are hard-coded rather than read from the context.

When adapting, move the needed values into parameters and define what “unanswered” means: time since creation, since the engineer's last reply, or since the customer's last message. A ticket's age alone does not define this metric. Guide: article 1449.

14. CSV and Excel Reports

Choose the Meaning of the Report

First, define the period, time zone, object filter, and the unit of a row. For example, an event report must distinguish the ticket's current group from the group at the moment of the event: the filter ticket__group=group refers to the current group.

The file from article 1906 selects events for a period and produces a CSV; the file from article 1434 groups created tickets by hour and weekday and produces an XLSX.

Full Training Example: List of Active Tickets in CSV

Task. Export up to 1,000 new and open tickets of a single group to CSV and send the file by email.
Trigger. Web form.
Parameters. group_uid, recipient.
Expected result. A CSV with columns Ticket ID;Status attached to an email from the specified address.

This standalone file send_active_tickets_csv.py uses familiar models and file sending from the analyzed examples.

Web form parameters:

[
  {
    "name": "group_uid",
    "type": "string",
    "required": true,
    "displayName": "Group UID"
  },
  {
    "name": "recipient",
    "type": "string",
    "required": true,
    "displayName": "Recipient email"
  }
]

For the first launch, allow the form to administrators only. If you expand access, restrict the allowed groups and recipients according to your company's rules. For sending reports, the recipient should not automatically receive any data merely because they know the group UID.

import csv
import logging
import os
import tempfile

from django.core.validators import validate_email
from django.core.exceptions import ValidationError
from core.models import EmailChannel, Group, Ticket
from core.tasks.notifications import send_email
from runtime.runners import BaseRunner
from swarmica import defs

logger = logging.getLogger(__name__)
MAX_ROWS = 1000


class Runner(BaseRunner):
    def run(self, *args, **kwargs):
        LOG_PREFIX = f"{self.script.name} ({self.script.uid})"
        data = kwargs.get("data") or {}
        group_uid = data.get("group_uid")
        recipient = data.get("recipient")
        if not isinstance(group_uid, str) or not isinstance(recipient, str):
            logger.error(f"{LOG_PREFIX}: specify group_uid and recipient as strings")
            return
        recipient = recipient.strip()
        try:
            validate_email(recipient)
        except ValidationError:
            logger.error(f"{LOG_PREFIX}: invalid email")
            return
        group = Group.objects.filter(uid=group_uid.strip()).first()
        channel = EmailChannel.objects.filter(default=True, is_deleted=False).first()
        if group is None or channel is None or not channel.send_from:
            logger.error(f"{LOG_PREFIX}: group or outgoing email channel not configured")
            return

        rows = list(Ticket.objects.filter(
            group=group,
            status__in=[defs.TICKET_STATUSES_NEW, defs.TICKET_STATUSES_OPEN],
        ).order_by("id").values_list("id", "status")[:MAX_ROWS + 1])
        if len(rows) > MAX_ROWS:
            logger.error(f"{LOG_PREFIX}: sample exceeds {MAX_ROWS} rows; narrow the report")
            return

        path = None
        try:
            with tempfile.NamedTemporaryFile(
                mode="w", encoding="utf-8-sig", newline="",
                suffix=".csv", delete=False,
            ) as output:
                path = output.name
                writer = csv.writer(output, delimiter=";")
                writer.writerow(["Ticket ID", "Status"])
                writer.writerows(rows)

            send_email(
                from_email=channel.send_from,
                to=[recipient],
                subject="Active tickets of the group",
                text=f"The attachment contains {len(rows)} records.",
                attachment_files=[path],
                connection=None if channel.use_system_smtp_relay else channel.get_smtp_connection(),
            )
            logger.info(f"{LOG_PREFIX}: report queued for sending; rows={len(rows)}")
        finally:
            if path is not None and os.path.exists(path):
                os.remove(path)

An empty sample will produce a CSV with just a header. If the limit is exceeded, the script does not send a truncated report. The temporary file is removed in finally.

This cleanup order matches a direct call to send_email in the analyzed files. If you switch the implementation to an asynchronous queue, the file must exist until the handler reads it; immediate deletion may break sending.

XLSX

For an Excel file, openpyxl is used. A training function that accepts a list of rows or another iterable:

from openpyxl import Workbook


def save_xlsx(rows, path):
    workbook = Workbook(write_only=True)
    sheet = workbook.create_sheet("Tickets")
    sheet.append(["Ticket ID", "Status"])
    for ticket_id, status in rows:
        sheet.append([ticket_id, status])
    workbook.save(path)

For free text from external sources, consider the possibility that a value will be interpreted as a formula by a spreadsheet editor. In the example above, only IDs and a fixed status are exported.

The hourly report uses ExtractHour, ExtractWeekDay, Count, Case, and When. Grouping is done in the query; then rows are turned into an Excel sheet. The time zone is passed into the time-extraction functions. Guide: article 1434.

Dates and Time Zones

For date comparison, use timezone-aware values. For a report covering whole days, it is convenient to define a half-open interval: from the start of the first day inclusive to the start of the day after the last one exclusive.

Standalone helper function for YYYY-MM-DD strings:

from datetime import date, datetime, time, timedelta
from zoneinfo import ZoneInfo


def report_period(date_from, date_to, timezone_name="Europe/Moscow"):
    start_day = date.fromisoformat(date_from)
    end_day = date.fromisoformat(date_to)
    if end_day < start_day:
        raise ValueError("End of period is earlier than the start")
    tz = ZoneInfo(timezone_name)
    start = datetime.combine(start_day, time.min, tzinfo=tz)
    end_exclusive = datetime.combine(end_day + timedelta(days=1), time.min, tzinfo=tz)
    utc = ZoneInfo("UTC")
    return start.astimezone(utc), end_exclusive.astimezone(utc)

In the query, use date__gte=start and date__lt=end_exclusive, or equivalent conditions for created_at. For special historical time-zone transitions, define ambiguous-time rules separately. General reference: Python zoneinfo.

Guide for the original status-transition report: article 1906. When adapting, additionally restrict the sample to the “Status change” event type and verify the ORM fields for the old and new values in your version.

15. External APIs and Paginated Fetching

Request and Response Handling

The migration example uses requests. In your implementation, set a timeout, check the HTTP status, and check the response format. The following fragment illustrates the call; the url, headers, and payload variables are defined by your scenario:

import requests

response = requests.post(url, json=payload, headers=headers, timeout=(5, 30))
response.raise_for_status()
result = response.json()

json=payload encodes the dictionary as JSON. timeout limits the wait for connection and reading. The response may not be JSON even with a successful HTTP status — handle that separately. See the official Requests documentation.

For network errors, provide a clear outcome: retry later, leave a record for manual review, or stop the scenario. Recreating an external object after a timeout can produce a duplicate if the first attempt actually succeeded. Use an idempotency key when the external system supports one.

Paginated Fetching

The article migrator reads results and follows next. Below is a training function for an API with that format. It limits the number of pages and does not forward authorization headers to another host.

from urllib.parse import urljoin, urlsplit
import requests


def iter_api_results(start_url, headers, max_pages=100):
    origin = urlsplit(start_url)
    if origin.scheme != "https" or not origin.netloc or origin.username or origin.password:
        raise ValueError("This example requires an HTTPS URL without credentials")
    url = start_url
    visited = set()
    page_count = 0

    with requests.Session() as session:
        while url:
            parsed = urlsplit(url)
            if (parsed.scheme, parsed.netloc) != (origin.scheme, origin.netloc):
                raise ValueError("next points to a different host")
            if url in visited or page_count >= max_pages:
                raise ValueError("Pagination loop or page limit exceeded")
            visited.add(url)
            page_count += 1
            response = session.get(
                url, headers=headers, timeout=(5, 30), allow_redirects=False,
            )
            if 300 <= response.status_code < 400:
                raise ValueError("Redirect requires separate review")
            response.raise_for_status()
            try:
                page = response.json()
            except ValueError as error:
                raise ValueError(
                    f"Response is not JSON (Content-Type={response.headers.get('Content-Type')!r})"
                ) from error
            if not isinstance(page, dict) or not isinstance(page.get("results"), list):
                raise ValueError("Expected an object with a results array")
            for item in page["results"]:
                yield item
            next_url = page.get("next")
            if next_url is not None and not isinstance(next_url, str):
                raise ValueError("next must be a string or null")
            url = urljoin(url, next_url) if next_url else None

In the same file, inside run, after obtaining verified host and token:

headers = {"Authorization": f"Token {token}"}
for category in iter_api_results(f"{host.rstrip('/')}/api/categories/", headers):
    logger.info(f"Fetched category id={category.get('id')}")

The Token scheme matches the API-token usage in the analyzed Swarmica migrator. For another system, the authorization method may differ. Here HTTPS is a constraint of the training function, not a description of all Swarmica installation variants.

Do not put real secrets into published code, client widgets, or logs. Agree on how the server-side secret is stored and issued with the installation administrator.

Article Migration: Why Two Passes

The file onetime_hc_migration.py does the work in two stages:

1. Fetch categories and create or find the corresponding local categories.
2. Create stub articles with external identifiers.
3. Build a mapping of source and local article IDs.
4. Upload attachments and translations, and replace links with local addresses.

This approach is needed when articles link to each other and the new IDs are not known in advance. refresh_migrated allows updating an article migrated earlier; in the original file, its translations and attachments are removed before the update. This is a replacement operation that must be tested on a backup.

The main load in the original file is limited to visibility=ALL. It is not a universal migration of all segmentation and access variants. Guide: article 306.

When adapting, keep the mapping inside a specific launch. State in module-level globals may persist if the environment reuses the module between launches.

16. Importing Users, Companies, and Articles

Import typically combines reading the file, checking columns, normalizing values, finding an existing object, creating or updating, and producing a summary.

Files and Containers

CSV examples use settings.UPLOADS_ROOT. The Markdown import expects the directory /swarmica/swarmica/old_articles, mounted into the container.

The file must be accessible to the exact process that runs the script. Copying into the django container helps only if the filesystem is shared with the relevant handler, or if the script runs in that container. Check your installation's volumes, not just the directory name.

Do not accept an arbitrary absolute path from a form without restriction. A training function for a file directly in the allowed directory:

from pathlib import Path


def resolve_csv_file(upload_root, filename):
    if not isinstance(filename, str) or not filename.strip():
        raise ValueError("Specify a file name")
    root = Path(upload_root).resolve()
    candidate = (root / filename.strip()).resolve()
    if candidate.parent != root:
        raise ValueError("The file must be directly in the uploads directory")
    if candidate.suffix.lower() != ".csv" or not candidate.is_file():
        raise ValueError("CSV file not found")
    return candidate

Reading CSV

Fragment with the obtained path:

import csv

with open(path, encoding="utf-8-sig", newline="") as source:
    reader = csv.DictReader(source, delimiter=";")
    required = {"Employee", "E-Mail"}
    if not required.issubset(set(reader.fieldnames or [])):
        raise ValueError("Columns Employee and E-Mail are required")
    for line_number, row in enumerate(reader, start=2):
        name = row["Employee"].strip()
        email = row["E-Mail"].strip().lower()
        # Row validation, object lookup, and processing happen here.

utf-8-sig accounts for a possible BOM. The delimiter and column names must match your file. The fragment deliberately does not create users.

Users with the “Employee” Role

File import_users.py:

• Reads the CSV name from the filename parameter.
• Expects columns Employee and E-Mail, separated by ;.
• Checks for an existing user by email and email identifier.
• Creates a user with defs.USER_ROLES_INTERNAL_USER.
• Calls set_unusable_password(); signing in requires a separate password setup.
• Counts created and skipped records.

The “Employee” role in this example does not automatically equal the “Agent” role. Do not change the role based on a file name without checking your access model. Guide: article 1741.

Companies and Customers

The file import_organizations_and_clients.py expects client;email;organization, supports a number of column synonyms, and uses Swarmica's internal data_import services with data schemas.

This is a useful example of extracting import into a separate layer. But its rules for merging objects are specific: the organization is looked up by name and external ID, and an existing user from a different organization is skipped. For your task, decide in advance how to handle matching names, differing external IDs, and company changes.

In the version we read, an existing user without a company does not get a company from the CSV: the branch treats the record as already processed. If you need that linking, add a separate, tested logic. Guide: article 1753.

Articles from Markdown

File article_import.py:

• Reads .md directly from the specified directory.
• Uses the file name without the extension as the article's subject.
• Finds or creates a category by name and language.
• Creates Article and ArticleTranslation.
• Uses Russian and status UNAPPROVED by default.

It does not look up already-imported articles: re-running creates new articles. For repeatable imports, add a stable source key and an update rule. Importing the text also does not mean that all local images and related files are transferred automatically. Guide: article 1661.

Disabling Signals in Migration Examples

Some imports use factory.django.mute_signals(...). This is a special technique for bulk migration that disables some model handlers during the run. It can affect event registration and related actions; in a normal process, signals may be important for other tasks as well.

Do not carry such a decorator into ordinary automation just for speed or to protect against repeats. For your own implementation, first determine which standard handlers you need, and verify the consequences of the altered write method.

Part V. Operations

17. Diagnostics, Performance, and Operations

What to Log

• Script name, UID, and version.
• Event and object IDs.
• The reason for a skip or stop.
• The number of found, changed, and failed records.
• The traceback of an unexpected exception.

Example fragment for handling a single object:

try:
    process_ticket(ticket)
except Exception:
    logger.exception(f"error processing ticket_id={ticket.id}")

Here, process_ticket is your own function. For independent objects, you can continue processing after one of them fails. For an all-or-nothing operation, the exception should roll back, not be hidden inside the transaction.

Common Problems

How to Check Model Fields on Your Version

If an example raises FieldError, you can temporarily print the model's field names in a diagnostic test script. Fragment to place inside run after importing the model:

from core.models import TicketEvent

field_names = sorted(field.name for field in TicketEvent._meta.get_fields())
logger.info(f"TicketEvent fields: {field_names}")

These are the names of ORM fields and relations, not all properties and methods of the object. Having an attribute in the context description does not make it a filterable field. For models or methods not covered by the documentation, send support the specific operation, your version, and the error; do not guess names blindly.

Performance and Launch Frequency

Start with a limited sample. Move repeated queries out of the loop, prefetch the necessary relations, and aggregate data in the database when that fits the task. Do not hold a large dataset in memory without need.

The schedule frequency must match the processing duration. If a task can run longer than the interval, provide coordination between launches. Limit network waits with timeouts; clean up temporary files.

Adding a file log handler on every module import can lead to duplicate records. For the first script, the standard logging.getLogger(__name__) is sufficient.

What to Check Before Running in Production

1. Valid, missing, and incorrectly typed parameters.
2. An empty sample and missing related objects.
3. Repeating the same event.
4. Parallel execution and concurrent user modification of the object.
5. Failure of an external system or SMTP.
6. Business hours, holidays, and time zone for time-based logic.
7. Events and other automations after data is written.
8. Form access and allowed data / actions.
9. Processing volume and run duration.

Keep the source in a version control system, record the verified Swarmica version and the maintenance owner. Before updating, verify extensions in a test installation. For diagnostics, prepare the script name and version, the launch method, anonymized parameters, the test object ID, and the error traceback.

18. Example Catalog and Further Learning Path

Below are publicly available articles with attached Python files that you can use to study individual techniques. A file from an article is a source example to verify and adapt, not a ready-made extension for any process.

Script Template

Download the sample.py template

Recommended path:

1. Run the minimal script from section 3 and confirm that parameters arrive.
2. Run your first working script from section 5 on a test group.
3. Set up a script that only logs an event (section 6).
4. On a test ticket, try a custom field or an internal note (sections 9–10).
5. Study the scheduled example and define the re-run rules (sections 7, 11).
6. Add a report or integration for your specific process (sections 13–16).
7. Agree on maintenance, access restrictions, and pre-update verification (sections 17–18).

Other guides are in the automation category. For launching via API or the command line, use the scheme of your version and an agreed-upon procedure: this guide does not define a universal CLI command or endpoint for all versions.