Data Export Mixin

DataExportMixin

The DataExportMixin class provides a plugin with the ability to customize the data export process. The InvenTree API provides an integrated method to export a dataset to a tabulated file. The default export process is generic, and simply exports the data presented via the API in a tabulated file format.

Custom data export plugins allow this process to be adjusted:

• Data columns can be added or removed
• Rows can be removed or added
• Custom calculations or annotations can be performed.

Supported Export Types

Each plugin can dictate which datasets are supported using the supports_export method. This allows a plugin to dynamically specify whether it can be selected by the user for a given export session.

Return True if this plugin supports exporting data for the given model.

Parameters:

Name	Type	Description	Default
model_class	type	The model class to check	required
user	User	The user requesting the export	required
serializer_class	Optional[Serializer]	The serializer class to use for exporting the data	None
view_class	Optional[APIView]	The view class to use for exporting the data	None

Returns:

Type	Description
bool	True if the plugin supports exporting data for the given model

Source code in src/backend/InvenTree/plugin/base/integration/DataExport.py

def supports_export(
    self,
    model_class: type,
    user: User,
    serializer_class: Optional[serializers.Serializer] = None,
    view_class: Optional[views.APIView] = None,
    *args,
    **kwargs,
) -> bool:
    """Return True if this plugin supports exporting data for the given model.

    Args:
        model_class: The model class to check
        user: The user requesting the export
        serializer_class: The serializer class to use for exporting the data
        view_class: The view class to use for exporting the data

    Returns:
        True if the plugin supports exporting data for the given model
    """
    # By default, plugins support all models
    return True

The default implementation returns True for all data types.

Filename Generation

The generate_filename method constructs a filename for the exported file.

Generate a filename for the exported data.

Source code in src/backend/InvenTree/plugin/base/integration/DataExport.py

def generate_filename(self, model_class, export_format: str) -> str:
    """Generate a filename for the exported data."""
    model = model_class.__name__
    date = current_date().isoformat()

    return f'InvenTree_{model}_{date}.{export_format}'

Adjust Columns

The update_headers method allows the plugin to adjust the columns selected to be exported to the file.

Update the headers for the data export.

Allows for optional modification of the headers for the data export.

Parameters:

Name	Type	Description	Default
headers	OrderedDict	The current headers for the export	required
context	dict	The context for the export (provided by the plugin serializer)	required

Source code in src/backend/InvenTree/plugin/base/integration/DataExport.py

def update_headers(
    self, headers: OrderedDict, context: dict, **kwargs
) -> OrderedDict:
    """Update the headers for the data export.

    Allows for optional modification of the headers for the data export.

    Arguments:
        headers: The current headers for the export
        context: The context for the export (provided by the plugin serializer)

    Returns: The updated headers
    """
    # The default implementation does nothing
    return headers

Queryset Filtering

The filter_queryset method allows the plugin to provide custom filtering to the database query, before it is exported.

Filter the queryset before exporting data.

Source code in src/backend/InvenTree/plugin/base/integration/DataExport.py

def filter_queryset(self, queryset: QuerySet) -> QuerySet:
    """Filter the queryset before exporting data."""
    # The default implementation returns the queryset unchanged
    return queryset

Export Data

The export_data method performs the step of transforming a Django QuerySet into a dataset which can be processed by the tablib library.

Export data from the queryset.

This method should be implemented by the plugin to provide the actual data export functionality.

Parameters:

Name	Type	Description	Default
queryset	QuerySet	The queryset to export	required
serializer_class	Serializer	The serializer class to use for exporting the data	required
headers	OrderedDict	The headers for the export	required
context	dict	Any custom context for the export (provided by the plugin serializer)	required
output	DataOutput	The DataOutput object for the export	required

Source code in src/backend/InvenTree/plugin/base/integration/DataExport.py

def export_data(
    self,
    queryset: QuerySet,
    serializer_class: serializers.Serializer,
    headers: OrderedDict,
    context: dict,
    output: DataOutput,
    **kwargs,
) -> list:
    """Export data from the queryset.

    This method should be implemented by the plugin to provide
    the actual data export functionality.

    Arguments:
        queryset: The queryset to export
        serializer_class: The serializer class to use for exporting the data
        headers: The headers for the export
        context: Any custom context for the export (provided by the plugin serializer)
        output: The DataOutput object for the export

    Returns: The exported data (a list of dict objects)
    """
    # The default implementation simply serializes the queryset
    return serializer_class(queryset, many=True, exporting=True).data

Note that the default implementation simply uses the builtin tabulation functionality of the provided serializer class. In most cases, this will be sufficient.

Custom Export Options

To provide the user with custom options to control the behavior of the export process at the time of export, the plugin can define a custom serializer class.

To enable this feature, define an ExportOptionsSerializer attribute on the plugin class which points to a DRF serializer class. Refer to the examples below for more information.

Builtin Exporter Classes

InvenTree provides the following builtin data exporter classes.

InvenTreeExporter

A generic exporter class which simply serializes the API output into a data file.

Generic exporter plugin for InvenTree.

Source code in src/backend/InvenTree/plugin/builtin/exporter/inventree_exporter.py

class InvenTreeExporter(DataExportMixin, InvenTreePlugin):
    """Generic exporter plugin for InvenTree."""

    NAME = 'InvenTree Exporter'
    SLUG = 'inventree-exporter'
    TITLE = _('InvenTree Generic Exporter')
    DESCRIPTION = _('Provides support for exporting data from InvenTree')
    VERSION = '1.0.0'
    AUTHOR = _('InvenTree contributors')

    def supports_export(self, model_class: type, user, *args, **kwargs) -> bool:
        """This exporter supports all model classes."""
        return True

BOM Exporter

A custom exporter which only supports bill of materials exporting.

Builtin plugin for performing multi-level BOM exports.

Source code in src/backend/InvenTree/plugin/builtin/exporter/bom_exporter.py

class BomExporterPlugin(DataExportMixin, InvenTreePlugin):
    """Builtin plugin for performing multi-level BOM exports."""

    NAME = 'BOM Exporter'
    SLUG = 'bom-exporter'
    TITLE = _('Multi-Level BOM Exporter')
    DESCRIPTION = _('Provides support for exporting multi-level BOMs')
    VERSION = '1.1.0'
    AUTHOR = _('InvenTree contributors')

    ExportOptionsSerializer = BomExporterOptionsSerializer

    def supports_export(self, model_class: type, user, *args, **kwargs) -> bool:
        """This exported only supports the BomItem model."""
        return (
            model_class == BomItem
            and kwargs.get('serializer_class') == BomItemSerializer
        )

    def update_headers(self, headers, context, **kwargs):
        """Update headers for the BOM export."""
        export_total_quantity = context.get('export_total_quantity', True)

        if not self.export_stock_data:
            # Remove stock data from the headers
            for field in [
                'available_stock',
                'available_substitute_stock',
                'available_variant_stock',
                'external_stock',
                'on_order',
                'building',
                'can_build',
            ]:
                headers.pop(field, None)

        if not self.export_pricing_data:
            # Remove pricing data from the headers
            for field in [
                'pricing_min',
                'pricing_max',
                'pricing_min_total',
                'pricing_max_total',
                'pricing_updated',
            ]:
                headers.pop(field, None)

        # Append a "BOM Level" field
        headers['level'] = _('BOM Level')

        if export_total_quantity:
            # Append a 'total quantity' field
            headers['total_quantity'] = _('Total Quantity')

        # Append variant part columns
        if self.export_substitute_data and self.n_substitute_cols > 0:
            for idx in range(self.n_substitute_cols):
                n = idx + 1
                headers[f'substitute_{idx}'] = _(f'Substitute {n}')

        # Append supplier part columns
        if self.export_supplier_data and self.n_supplier_cols > 0:
            for idx in range(self.n_supplier_cols):
                n = idx + 1
                headers[f'supplier_name_{idx}'] = _(f'Supplier {n}')
                headers[f'supplier_sku_{idx}'] = _(f'Supplier {n} SKU')
                headers[f'supplier_mpn_{idx}'] = _(f'Supplier {n} MPN')

        # Append manufacturer part columns
        if self.export_manufacturer_data and self.n_manufacturer_cols > 0:
            for idx in range(self.n_manufacturer_cols):
                n = idx + 1
                headers[f'manufacturer_name_{idx}'] = _(f'Manufacturer {n}')
                headers[f'manufacturer_mpn_{idx}'] = _(f'Manufacturer {n} MPN')

        # Append part parameter columns
        if self.export_parameter_data and len(self.parameters) > 0:
            for key, value in self.parameters.items():
                headers[f'parameter_{key}'] = value

        return headers

    def prefetch_queryset(self, queryset):
        """Perform pre-fetch on the provided queryset."""
        queryset = queryset.prefetch_related('sub_part')

        if self.export_substitute_data:
            queryset = queryset.prefetch_related('substitutes')

        if self.export_supplier_data:
            queryset = queryset.prefetch_related(
                'sub_part__supplier_parts',
                'sub_part__supplier_parts__supplier',
                'sub_part__supplier_parts__manufacturer_part',
                'sub_part__supplier_parts__manufacturer_part__manufacturer',
            )

        if self.export_manufacturer_data:
            queryset = queryset.prefetch_related(
                'sub_part__manufacturer_parts',
                'sub_part__manufacturer_parts__manufacturer',
            )

        if self.export_parameter_data:
            queryset = queryset.prefetch_related(
                'sub_part__parameters_list', 'sub_part__parameters_list__template'
            )

        return queryset

    def export_data(
        self, queryset, serializer_class, headers, context, output, **kwargs
    ):
        """Export BOM data from the queryset."""
        self.serializer_class = serializer_class

        # Track how many extra columns we need
        self.n_substitute_cols = 0
        self.n_supplier_cols = 0
        self.n_manufacturer_cols = 0

        # A dict of "Parameter ID" -> "Parameter Name"
        self.parameters = {}

        # Extract the export options from the context (and cache for later)
        self.export_levels = context.get('export_levels', 1)
        self.export_stock_data = context.get('export_stock_data', True)
        self.export_pricing_data = context.get('export_pricing_data', True)
        self.export_supplier_data = context.get('export_supplier_data', True)
        self.export_manufacturer_data = context.get('export_manufacturer_data', True)
        self.export_substitute_data = context.get('export_substitute_data', True)
        self.export_parameter_data = context.get('export_parameter_data', True)
        self.export_total_quantity = context.get('export_total_quantity', True)

        # Pre-fetch related data to reduce database queries
        queryset = self.prefetch_queryset(queryset)

        self.bom_data = []

        # Run through each item in the queryset
        for bom_item in queryset:
            self.process_bom_row(bom_item, 1, **kwargs)

        return self.bom_data

    def process_bom_row(
        self, bom_item, level: int = 1, multiplier: Optional[Decimal] = None, **kwargs
    ) -> list:
        """Process a single BOM row.

        Arguments:
            bom_item: The BomItem object to process
            level: The current level of export
            multiplier: The multiplier for the quantity (used for recursive calls)
        """
        # Add this row to the output dataset
        row = self.serializer_class(bom_item, exporting=True).data
        row['level'] = level

        if multiplier is None:
            multiplier = Decimal(1)

        # Extend with additional data

        if self.export_substitute_data:
            row.update(self.get_substitute_data(bom_item))

        if self.export_supplier_data:
            row.update(self.get_supplier_data(bom_item))

        if self.export_manufacturer_data:
            row.update(self.get_manufacturer_data(bom_item))

        if self.export_parameter_data:
            row.update(self.get_parameter_data(bom_item))

        if self.export_total_quantity:
            # Calculate the total quantity for this BOM item
            total_quantity = Decimal(bom_item.quantity) * multiplier
            row['total_quantity'] = normalize(total_quantity)

        self.bom_data.append(row)

        # If we have reached the maximum export level, return just this bom item
        if bom_item.sub_part.assembly and (
            self.export_levels <= 0 or level < self.export_levels
        ):
            sub_items = bom_item.sub_part.get_bom_items()
            sub_items = self.prefetch_queryset(sub_items)
            sub_items = BomItemSerializer.annotate_queryset(sub_items)

            for item in sub_items.all():
                self.process_bom_row(
                    item,
                    level=level + 1,
                    multiplier=multiplier * bom_item.quantity,
                    **kwargs,
                )

    def get_substitute_data(self, bom_item: BomItem) -> dict:
        """Return substitute part data for a BomItem."""
        substitute_part_data = {}

        idx = 0

        for substitute in bom_item.substitutes.all():
            substitute_part_data.update({f'substitute_{idx}': substitute.part.name})

            idx += 1

        self.n_substitute_cols = max(self.n_substitute_cols, idx)

        return substitute_part_data

    def get_supplier_data(self, bom_item: BomItem) -> dict:
        """Return supplier and manufacturer data for a BomItem."""
        supplier_part_data = {}

        idx = 0

        for supplier_part in bom_item.sub_part.supplier_parts.all():
            manufacturer_part = supplier_part.manufacturer_part
            supplier_part_data.update({
                f'supplier_name_{idx}': supplier_part.supplier.name
                if supplier_part.supplier
                else '',
                f'supplier_sku_{idx}': supplier_part.SKU,
                f'supplier_mpn_{idx}': manufacturer_part.MPN
                if manufacturer_part
                else '',
            })

            idx += 1

        self.n_supplier_cols = max(self.n_supplier_cols, idx)

        return supplier_part_data

    def get_manufacturer_data(self, bom_item: BomItem) -> dict:
        """Return manufacturer data for a BomItem."""
        manufacturer_part_data = {}

        idx = 0

        for manufacturer_part in bom_item.sub_part.manufacturer_parts.all():
            manufacturer_part_data.update({
                f'manufacturer_name_{idx}': manufacturer_part.manufacturer.name
                if manufacturer_part.manufacturer
                else '',
                f'manufacturer_mpn_{idx}': manufacturer_part.MPN,
            })

            idx += 1

        self.n_manufacturer_cols = max(self.n_manufacturer_cols, idx)

        return manufacturer_part_data

    def get_parameter_data(self, bom_item: BomItem) -> dict:
        """Return parameter data for a BomItem."""
        parameter_data = {}

        for parameter in bom_item.sub_part.parameters.all():
            template = parameter.template
            if template.pk not in self.parameters:
                self.parameters[template.pk] = template.name

            parameter_data.update({f'parameter_{template.pk}': parameter.data})

        return parameter_data

Source Code

The full source code of the DataExportMixin class:

DataExportMixin

"""Plugin class for custom data exporting."""

from collections import OrderedDict
from typing import Optional

from django.contrib.auth.models import User
from django.db.models import QuerySet

from rest_framework import serializers, views

from common.models import DataOutput
from InvenTree.helpers import current_date
from plugin import PluginMixinEnum


class DataExportMixin:
    """Mixin which provides ability to customize data exports.

    When exporting data from the API, this mixin can be used to provide
    custom data export functionality.
    """

    ExportOptionsSerializer = None

    class MixinMeta:
        """Meta options for this mixin."""

        MIXIN_NAME = 'DataExport'

    def __init__(self):
        """Register mixin."""
        super().__init__()
        self.add_mixin(PluginMixinEnum.EXPORTER, True, __class__)

    def supports_export(
        self,
        model_class: type,
        user: User,
        serializer_class: Optional[serializers.Serializer] = None,
        view_class: Optional[views.APIView] = None,
        *args,
        **kwargs,
    ) -> bool:
        """Return True if this plugin supports exporting data for the given model.

        Args:
            model_class: The model class to check
            user: The user requesting the export
            serializer_class: The serializer class to use for exporting the data
            view_class: The view class to use for exporting the data

        Returns:
            True if the plugin supports exporting data for the given model
        """
        # By default, plugins support all models
        return True

    def generate_filename(self, model_class, export_format: str) -> str:
        """Generate a filename for the exported data."""
        model = model_class.__name__
        date = current_date().isoformat()

        return f'InvenTree_{model}_{date}.{export_format}'

    def update_headers(
        self, headers: OrderedDict, context: dict, **kwargs
    ) -> OrderedDict:
        """Update the headers for the data export.

        Allows for optional modification of the headers for the data export.

        Arguments:
            headers: The current headers for the export
            context: The context for the export (provided by the plugin serializer)

        Returns: The updated headers
        """
        # The default implementation does nothing
        return headers

    def filter_queryset(self, queryset: QuerySet) -> QuerySet:
        """Filter the queryset before exporting data."""
        # The default implementation returns the queryset unchanged
        return queryset

    def export_data(
        self,
        queryset: QuerySet,
        serializer_class: serializers.Serializer,
        headers: OrderedDict,
        context: dict,
        output: DataOutput,
        **kwargs,
    ) -> list:
        """Export data from the queryset.

        This method should be implemented by the plugin to provide
        the actual data export functionality.

        Arguments:
            queryset: The queryset to export
            serializer_class: The serializer class to use for exporting the data
            headers: The headers for the export
            context: Any custom context for the export (provided by the plugin serializer)
            output: The DataOutput object for the export

        Returns: The exported data (a list of dict objects)
        """
        # The default implementation simply serializes the queryset
        return serializer_class(queryset, many=True, exporting=True).data

    def get_export_options_serializer(self, **kwargs) -> serializers.Serializer | None:
        """Return a serializer class with dynamic export options for this plugin.

        Returns:
            A class instance of a DRF serializer class, by default this an instance of
            self.ExportOptionsSerializer using the *args, **kwargs if existing for this plugin
        """
        # By default, look for a class level attribute
        serializer = getattr(self, 'ExportOptionsSerializer', None)

        if serializer:
            return serializer(**kwargs)