Skip to content

Supplier Mixin

SupplierMixin

The SupplierMixin class enables plugins to integrate with external suppliers, enabling seamless creation of parts, supplier parts, and manufacturer parts with just a few clicks from the supplier. The import process is split into multiple phases:

  • Search supplier
  • Select InvenTree category
  • Match Part Parameters
  • Create initial Stock

Import Methods

A plugin can connect to multiple suppliers. The get_suppliers method should return a list of available supplier connections (e.g. using different credentials). When a user initiates a search through the UI, the get_search_results function is called with the search term, supplier slug returned previously, and the search results are returned. These contain a part_id which is then passed to get_import_data along with the supplier_slug, if a user decides to import that specific part. This function should return a bunch of data that is needed for the import process. This data may be cached in the future for the same part_id. Then depending if the user only wants to import the supplier and manufacturer part or the whole part, the import_part, import_manufacturer_part and import_supplier_part methods are called automatically. If the user has imported the complete part, the get_parameters method is used to get a list of parameters which then can be match to inventree part parameter templates with some provided guidance. Additionally the get_pricing_data method is used to extract price breaks which are automatically considered when creating initial stock through the UI in the part import wizard.

For that to work, a few methods need to be overridden:

Mixin which provides integration to specific suppliers.

Source code in src/backend/InvenTree/plugin/base/supplier/mixins.py 
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
class SupplierMixin(SettingsMixin, Generic[PartData]):
    """Mixin which provides integration to specific suppliers."""

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

        MIXIN_NAME = 'Supplier'

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

        self.SETTINGS['SUPPLIER'] = {
            'name': 'Supplier',
            'description': 'The Supplier which this plugin integrates with.',
            'model': 'company.company',
            'model_filters': {'is_supplier': True},
            'required': True,
        }

    @property
    def supplier_company(self):
        """Return the supplier company object."""
        pk = self.get_setting('SUPPLIER', cache=True)
        if not pk:
            raise supplier.PartImportError('Supplier setting is missing.')

        return company.models.Company.objects.get(pk=pk)

    # --- Methods to be overridden by plugins ---
    def get_suppliers(self) -> list[supplier.Supplier]:
        """Return a list of available suppliers."""
        raise NotImplementedError('This method needs to be overridden.')

    def get_search_results(
        self, supplier_slug: str, term: str
    ) -> list[supplier.SearchResult]:
        """Return a list of search results for the given search term."""
        raise NotImplementedError('This method needs to be overridden.')

    def get_import_data(self, supplier_slug: str, part_id: str) -> PartData:
        """Return the import data for the given part ID."""
        raise NotImplementedError('This method needs to be overridden.')

    def get_pricing_data(self, data: PartData) -> dict[int, tuple[float, str]]:
        """Return a dictionary of pricing data for the given part data."""
        raise NotImplementedError('This method needs to be overridden.')

    def get_parameters(self, data: PartData) -> list[supplier.ImportParameter]:
        """Return a list of parameters for the given part data."""
        raise NotImplementedError('This method needs to be overridden.')

    def import_part(
        self,
        data: PartData,
        *,
        category: Optional[part_models.PartCategory],
        creation_user: Optional[django.contrib.auth.models.User],
    ) -> part_models.Part:
        """Import a part using the provided data.

        This may include:
          - Creating a new part
          - Add an image to the part
          - if this part has several variants, (create) a template part and assign it to the part
          - create related parts
          - add attachments to the part
        """
        raise NotImplementedError('This method needs to be overridden.')

    def import_manufacturer_part(
        self, data: PartData, *, part: part_models.Part
    ) -> company.models.ManufacturerPart:
        """Import a manufacturer part using the provided data.

        This may include:
          - Creating a new manufacturer
          - Creating a new manufacturer part
          - Assigning the part to the manufacturer part
          - Setting the default supplier for the part
          - Adding parameters to the manufacturer part
          - Adding attachments to the manufacturer part
        """
        raise NotImplementedError('This method needs to be overridden.')

    def import_supplier_part(
        self,
        data: PartData,
        *,
        part: part_models.Part,
        manufacturer_part: company.models.ManufacturerPart,
    ) -> part_models.SupplierPart:
        """Import a supplier part using the provided data.

        This may include:
          - Creating a new supplier part
          - Creating supplier price breaks
        """
        raise NotImplementedError('This method needs to be overridden.')

    # --- Helper methods for importing parts ---
    def download_image(self, img_url: str):
        """Download an image from the given URL and return it as a ContentFile."""
        img_r = download_image_from_url(img_url)
        fmt = img_r.format or 'PNG'
        buffer = io.BytesIO()
        img_r.save(buffer, format=fmt)

        return ContentFile(buffer.getvalue()), fmt

    def get_template_part(
        self, other_variants: list[part_models.Part], template_kwargs: dict[str, Any]
    ) -> part_models.Part:
        """Helper function to handle variant parts.

        This helper function identifies all roots for the provided 'other_variants' list
            - for no root => root part will be created using the 'template_kwargs'
            - for one root
                - root is a template => return it
                - root is no template, create a new template like if there is no root
                  and assign it to only root that was found and return it
            - for multiple roots => error raised
        """
        root_set = {v.get_root() for v in other_variants}

        # check how much roots for the variant parts exist to identify the parent_part
        parent_part = None  # part that should be used as parent_part
        root_part = None  # part that was discovered as root part in root_set
        if len(root_set) == 1:
            root_part = next(iter(root_set))
            if root_part.is_template:
                parent_part = root_part

        if len(root_set) == 0 or (root_part and not root_part.is_template):
            parent_part = part_models.Part.objects.create(**template_kwargs)

        if not parent_part:
            raise supplier.PartImportError(
                f'A few variant parts from the supplier are already imported, but have different InvenTree variant root parts, try to merge them to the same root variant template part (parts: {", ".join(str(p.pk) for p in other_variants)}).'
            )

        # assign parent_part to root_part if root_part has no variant of already
        if root_part and not root_part.is_template and not root_part.variant_of:
            root_part.variant_of = parent_part  # type: ignore
            root_part.save()

        return parent_part

    def create_related_parts(
        self, part: part_models.Part, related_parts: list[part_models.Part]
    ):
        """Create relationships between the given part and related parts."""
        for p in related_parts:
            try:
                part_models.PartRelated.objects.create(part_1=part, part_2=p)
            except ValidationError:
                pass  # pass, duplicate relationship detected

get_search_results(supplier_slug, term)

Return a list of search results for the given search term.

Source code in src/backend/InvenTree/plugin/base/supplier/mixins.py 
55
56
57
58
59
def get_search_results(
    self, supplier_slug: str, term: str
) -> list[supplier.SearchResult]:
    """Return a list of search results for the given search term."""
    raise NotImplementedError('This method needs to be overridden.')

get_import_data(supplier_slug, part_id)

Return the import data for the given part ID.

Source code in src/backend/InvenTree/plugin/base/supplier/mixins.py 
61
62
63
def get_import_data(self, supplier_slug: str, part_id: str) -> PartData:
    """Return the import data for the given part ID."""
    raise NotImplementedError('This method needs to be overridden.')

get_pricing_data(data)

Return a dictionary of pricing data for the given part data.

Source code in src/backend/InvenTree/plugin/base/supplier/mixins.py 
65
66
67
def get_pricing_data(self, data: PartData) -> dict[int, tuple[float, str]]:
    """Return a dictionary of pricing data for the given part data."""
    raise NotImplementedError('This method needs to be overridden.')

get_parameters(data)

Return a list of parameters for the given part data.

Source code in src/backend/InvenTree/plugin/base/supplier/mixins.py 
69
70
71
def get_parameters(self, data: PartData) -> list[supplier.ImportParameter]:
    """Return a list of parameters for the given part data."""
    raise NotImplementedError('This method needs to be overridden.')

import_part(data, *, category, creation_user)

Import a part using the provided data.

This may include
  • Creating a new part
  • Add an image to the part
  • if this part has several variants, (create) a template part and assign it to the part
  • create related parts
  • add attachments to the part
Source code in src/backend/InvenTree/plugin/base/supplier/mixins.py 
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
def import_part(
    self,
    data: PartData,
    *,
    category: Optional[part_models.PartCategory],
    creation_user: Optional[django.contrib.auth.models.User],
) -> part_models.Part:
    """Import a part using the provided data.

    This may include:
      - Creating a new part
      - Add an image to the part
      - if this part has several variants, (create) a template part and assign it to the part
      - create related parts
      - add attachments to the part
    """
    raise NotImplementedError('This method needs to be overridden.')

import_manufacturer_part(data, *, part)

Import a manufacturer part using the provided data.

This may include
  • Creating a new manufacturer
  • Creating a new manufacturer part
  • Assigning the part to the manufacturer part
  • Setting the default supplier for the part
  • Adding parameters to the manufacturer part
  • Adding attachments to the manufacturer part
Source code in src/backend/InvenTree/plugin/base/supplier/mixins.py 
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
def import_manufacturer_part(
    self, data: PartData, *, part: part_models.Part
) -> company.models.ManufacturerPart:
    """Import a manufacturer part using the provided data.

    This may include:
      - Creating a new manufacturer
      - Creating a new manufacturer part
      - Assigning the part to the manufacturer part
      - Setting the default supplier for the part
      - Adding parameters to the manufacturer part
      - Adding attachments to the manufacturer part
    """
    raise NotImplementedError('This method needs to be overridden.')

import_supplier_part(data, *, part, manufacturer_part)

Import a supplier part using the provided data.

This may include
  • Creating a new supplier part
  • Creating supplier price breaks
Source code in src/backend/InvenTree/plugin/base/supplier/mixins.py 
106
107
108
109
110
111
112
113
114
115
116
117
118
119
def import_supplier_part(
    self,
    data: PartData,
    *,
    part: part_models.Part,
    manufacturer_part: company.models.ManufacturerPart,
) -> part_models.SupplierPart:
    """Import a supplier part using the provided data.

    This may include:
      - Creating a new supplier part
      - Creating supplier price breaks
    """
    raise NotImplementedError('This method needs to be overridden.')

Sample Plugin

A simple example is provided in the InvenTree code base. Note that this uses some static data, but this can be extended in a real world plugin to e.g. call the supplier's API:

Example plugin to integrate with a dummy supplier.

Source code in src/backend/InvenTree/plugin/samples/supplier/supplier_sample.py 
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
class SampleSupplierPlugin(SupplierMixin, InvenTreePlugin):
    """Example plugin to integrate with a dummy supplier."""

    NAME = 'SampleSupplierPlugin'
    SLUG = 'samplesupplier'
    TITLE = 'My sample supplier plugin'

    VERSION = '0.0.1'

    def __init__(self):
        """Initialize the sample supplier plugin."""
        super().__init__()

        self.sample_data = []
        for material in ['Steel', 'Aluminium', 'Brass']:
            for size in ['M1', 'M2', 'M3', 'M4', 'M5']:
                for length in range(5, 30, 5):
                    self.sample_data.append({
                        'material': material,
                        'thread': size,
                        'length': length,
                        'sku': f'BOLT-{material}-{size}-{length}',
                        'name': f'Bolt {size}x{length}mm {material}',
                        'description': f'This is a sample part description demonstration purposes for the {size}x{length} {material} bolt.',
                        'price': {
                            1: [1.0, 'EUR'],
                            10: [0.9, 'EUR'],
                            100: [0.8, 'EUR'],
                            5000: [0.5, 'EUR'],
                        },
                        'link': f'https://example.com/sample-part-{size}-{length}-{material}',
                        'image_url': r'https://github.com/inventree/demo-dataset/blob/main/media/part_images/flat-head.png?raw=true',
                        'brand': 'Bolt Manufacturer',
                    })

    def get_suppliers(self) -> list[supplier.Supplier]:
        """Return a list of available suppliers."""
        return [supplier.Supplier(slug='sample-fasteners', name='Sample Fasteners')]

    def get_search_results(
        self, supplier_slug: str, term: str
    ) -> list[supplier.SearchResult]:
        """Return a list of search results based on the search term."""
        return [
            supplier.SearchResult(
                sku=p['sku'],
                name=p['name'],
                description=p['description'],
                exact=p['sku'] == term,
                price=f'{p["price"][1][0]:.2f}€',
                link=p['link'],
                image_url=p['image_url'],
                existing_part=getattr(
                    SupplierPart.objects.filter(SKU=p['sku']).first(), 'part', None
                ),
            )
            for p in self.sample_data
            if all(t.lower() in p['name'].lower() for t in term.split())
        ]

    def get_import_data(self, supplier_slug: str, part_id: str):
        """Return import data for a specific part ID."""
        for p in self.sample_data:
            if p['sku'] == part_id:
                p = p.copy()
                p['variants'] = [
                    x['sku']
                    for x in self.sample_data
                    if x['thread'] == p['thread'] and x['length'] == p['length']
                ]
                return p

        raise supplier.PartNotFoundError()

    def get_pricing_data(self, data) -> dict[int, tuple[float, str]]:
        """Return pricing data for the given part data."""
        return data['price']

    def get_parameters(self, data) -> list[supplier.ImportParameter]:
        """Return a list of parameters for the given part data."""
        return [
            supplier.ImportParameter(name='Thread', value=data['thread'][1:]),
            supplier.ImportParameter(name='Length', value=f'{data["length"]}mm'),
            supplier.ImportParameter(name='Material', value=data['material']),
            supplier.ImportParameter(name='Head', value='Flat Head'),
        ]

    def import_part(self, data, **kwargs) -> Part:
        """Import a part based on the provided data."""
        part, created = Part.objects.get_or_create(
            name__iexact=data['sku'],
            purchaseable=True,
            defaults={
                'name': data['sku'],
                'description': data['description'],
                'link': data['link'],
                **kwargs,
            },
        )

        # If the part was created, set additional fields
        if created:
            if data['image_url']:
                file, fmt = self.download_image(data['image_url'])
                filename = f'part_{part.pk}_image.{fmt.lower()}'
                part.image.save(filename, file)

            # link other variants if they exist in our inventree database
            if len(data['variants']):
                # search for other parts that may already have a template part associated
                variant_parts = [
                    x.part
                    for x in SupplierPart.objects.filter(SKU__in=data['variants'])
                ]
                parent_part = self.get_template_part(
                    variant_parts,
                    {
                        # we cannot extract a real name for the root part, but we can try to guess a unique name
                        'name': data['sku'].replace(data['material'] + '-', ''),
                        'description': data['name'].replace(' ' + data['material'], ''),
                        'link': data['link'],
                        'image': part.image.name,
                        'is_template': True,
                        **kwargs,
                    },
                )

                # after the template part was created, we need to refresh the part from the db because its tree id may have changed
                # which results in an error if saved directly
                part.refresh_from_db()
                part.variant_of = parent_part  # type: ignore
                part.save()

        return part

    def import_manufacturer_part(self, data, **kwargs) -> ManufacturerPart:
        """Import a manufacturer part based on the provided data."""
        mft, _ = Company.objects.get_or_create(
            name__iexact=data['brand'],
            defaults={
                'is_manufacturer': True,
                'is_supplier': False,
                'name': data['brand'],
            },
        )

        mft_part, created = ManufacturerPart.objects.get_or_create(
            MPN=f'MAN-{data["sku"]}', manufacturer=mft, **kwargs
        )

        if created:
            # Attachments, notes, parameters and more can be added here
            pass

        return mft_part

    def import_supplier_part(self, data, **kwargs) -> SupplierPart:
        """Import a supplier part based on the provided data."""
        spp, _ = SupplierPart.objects.get_or_create(
            SKU=data['sku'],
            supplier=self.supplier_company,
            **kwargs,
            defaults={'link': data['link']},
        )

        SupplierPriceBreak.objects.filter(part=spp).delete()
        SupplierPriceBreak.objects.bulk_create([
            SupplierPriceBreak(
                part=spp, quantity=quantity, price=price, price_currency=currency
            )
            for quantity, (price, currency) in data['price'].items()
        ])

        return spp