Coverage for src/backend/InvenTree/plugin/plugin.py: 72%

1"""Base Class for InvenTree plugins."""

3import inspect

4import json

5import re

6import warnings

7from datetime import datetime

8from importlib.metadata import PackageNotFoundError, metadata

9from pathlib import Path

10from typing import Optional

12from django.conf import settings

13from django.contrib.staticfiles.storage import StaticFilesStorage

14from django.utils.text import slugify

15from django.utils.translation import gettext_lazy as _

17import structlog

19import InvenTree.helpers

20from generic.enums import StringEnum

21from plugin.helpers import get_git_log

23logger = structlog.get_logger('inventree')

26def is_method_like(method) -> bool:

27 """Check if a method is callable and not a property."""

28 return any([

29 callable(method),

30 isinstance(method, classmethod),

31 isinstance(method, staticmethod),

32 isinstance(method, property),

33 ])

36def mark_final(method):

37 """Decorator to mark a method as 'final'.

39 This prevents subclasses from overriding this method.

40 """

41 if not is_method_like(method): 41 ↛ 42line 41 didn't jump to line 42 because the condition on line 41 was never true

42 raise TypeError('mark_final can only be applied to functions')

44 method.__final__ = True

45 return method

48def get_final_methods(cls):

49 """Find all methods of a class marked with the @mark_final decorator."""

50 return [

51 name

52 for name, method in inspect.getmembers(cls)

53 if getattr(method, '__final__', False) and is_method_like(method)

54 ]

57class PluginMixinEnum(StringEnum):

58 """Enumeration of the available plugin mixin types."""

60 BASE = 'base'

62 ACTION = 'action'

63 API_CALL = 'api_call'

64 APP = 'app'

65 BARCODE = 'barcode'

66 CURRENCY_EXCHANGE = 'currencyexchange'

67 EVENTS = 'events'

68 EXPORTER = 'exporter'

69 ICON_PACK = 'icon_pack'

70 LABELS = 'labels'

71 LOCATE = 'locate'

72 MACHINE = 'machine'

73 MAIL = 'mail'

74 NAVIGATION = 'navigation'

75 NOTIFICATION = 'notification'

76 REPORT = 'report'

77 SCHEDULE = 'schedule'

78 SETTINGS = 'settings'

79 SETTINGS_CONTENT = 'settingscontent'

80 SUPPLIER = 'supplier'

81 STATE_TRANSITION = 'statetransition'

82 SUPPLIER_BARCODE = 'supplier-barcode'

83 URLS = 'urls'

84 USER_INTERFACE = 'ui'

85 VALIDATION = 'validation'

88class MetaBase:

89 """Base class for a plugins metadata."""

91 # Override the plugin name for each concrete plugin instance

92 NAME = ''

93 SLUG = None

94 TITLE = None

96 @mark_final

97 def get_meta_value(self, key: str, old_key: Optional[str] = None, default=None):

98 """Reference a meta item with a key.

100 Args:

101 key (str): key for the value

102 old_key (str, optional): deprecated key - will throw warning

103 default (optional): Value if nothing with key can be found. Defaults to None.

104

105 Returns:

106 Value referenced with key, old_key or default if set and not value found

107 """

108 value = getattr(self, key, None)

109

110 # The key was not used

111 if old_key and value is None:

112 value = getattr(self, old_key, None)

113

114 # Sound of a warning if old_key worked

115 if value: 115 ↛ 116line 115 didn't jump to line 116 because the condition on line 115 was never true

116 warnings.warn(

117 f'Usage of {old_key} was depreciated in 0.7.0 in favour of {key}',

118 DeprecationWarning,

119 stacklevel=2,

120 )

121

122 # Use default if still nothing set

123 if value is None and default is not None: 123 ↛ 124line 123 didn't jump to line 124 because the condition on line 123 was never true

124 return default

125 return value

126

127 @mark_final

128 def plugin_name(self):

129 """Name of plugin."""

130 return self.get_meta_value('NAME', 'PLUGIN_NAME')

131

132 @property

133 @mark_final

134 def name(self):

135 """Name of plugin."""

136 return self.plugin_name()

137

138 @mark_final

139 def plugin_slug(self):

140 """Slug of plugin.

141

142 If not set plugin name slugified

143 """

144 slug = self.get_meta_value('SLUG', 'PLUGIN_SLUG', None)

145 if not slug:

146 slug = self.plugin_name()

147

148 return slugify(slug.lower())

149

150 @property

151 @mark_final

152 def slug(self):

153 """Slug of plugin."""

154 return self.plugin_slug()

155

156 @mark_final

157 def plugin_title(self):

158 """Title of plugin."""

159 title = self.get_meta_value('TITLE', 'PLUGIN_TITLE', None)

160 if title: 160 ↛ 162line 160 didn't jump to line 162 because the condition on line 160 was always true

161 return title

162 return self.plugin_name()

163

164 @property

165 @mark_final

166 def human_name(self):

167 """Human readable name of plugin."""

168 return self.plugin_title()

169

170 @mark_final

171 def plugin_config(self):

172 """Return the PluginConfig object associated with this plugin."""

173 from plugin.registry import registry

174

175 return registry.get_plugin_config(self.plugin_slug())

176

177 @mark_final

178 def is_active(self) -> bool:

179 """Return True if this plugin is currently active."""

180 # Mandatory plugins are always considered "active"

181 if self.is_mandatory(): 181 ↛ 184line 181 didn't jump to line 184 because the condition on line 181 was always true

182 return True

183

184 config = self.plugin_config()

185

186 if config:

187 return config.is_active()

188

189 return False

190

191

192class MixinBase:

193 """Base set of mixin functions and mechanisms."""

194

195 def __init__(self, *args, **kwargs) -> None:

196 """Init sup-parts.

197

198 Adds state dicts.

199 """

200 self._mixinreg = {}

201 self._mixins = {}

202 super().__init__(*args, **kwargs)

203

204 @mark_final

205 def mixin(self, key: str) -> bool:

206 """Check if mixin is registered."""

207 key = str(key).lower()

208 return key in self._mixins

209

210 @mark_final

211 def mixin_enabled(self, key: str) -> bool:

212 """Check if mixin is registered, enabled and ready."""

213 key = str(key).lower()

214

215 if self.mixin(key):

216 fnc_name = self._mixins.get(key)

217

218 # Allow for simple case where the mixin is "always" ready

219 if fnc_name is True:

220 return True

221

222 attr = getattr(self, fnc_name, True)

223

224 if callable(attr): 224 ↛ 225line 224 didn't jump to line 225 because the condition on line 224 was never true

225 return attr()

226 else:

227 return attr

228

229 return False

230

231 @mark_final

232 def add_mixin(self, key: str, fnc_enabled=True, cls=None):

233 """Add a mixin to the plugins registry."""

234 key = str(key).lower()

235

236 self._mixins[key] = fnc_enabled

237 self.setup_mixin(key, cls=cls)

238

239 @mark_final

240 def setup_mixin(self, key, cls=None):

241 """Define mixin details for the current mixin -> provides meta details for all active mixins."""

242 # get human name

243 human_name = (

244 getattr(cls.MixinMeta, 'MIXIN_NAME', key)

245 if cls and hasattr(cls, 'MixinMeta')

246 else key

247 )

248

249 # register

250 self._mixinreg[key] = {'key': key, 'human_name': human_name, 'cls': cls}

251

252 @mark_final

253 def get_registered_mixins(self, with_base: bool = False, with_cls: bool = True):

254 """Get all registered mixins for the plugin."""

255 mixins = getattr(self, '_mixinreg', None)

256 if not mixins:

257 return {}

258

259 mixins = mixins.copy()

260 # filter out base

261 if not with_base and 'base' in mixins:

262 del mixins['base']

263

264 # Do not return the mixin class if flas is set

265 if not with_cls:

266 return {

267 key: {k: v for k, v in mixin.items() if k != 'cls'}

268 for key, mixin in mixins.items()

269 }

270 return mixins

271

272 @property

273 @mark_final

274 def registered_mixins(self, with_base: bool = False):

275 """Get all registered mixins for the plugin."""

276 return self.get_registered_mixins(with_base=with_base)

277

278

279class VersionMixin:

280 """Mixin to enable version checking."""

281

282 MIN_VERSION = None

283 MAX_VERSION = None

284

285 @mark_final

286 def check_version(self, latest=None) -> bool:

287 """Check if plugin functions for the current InvenTree version."""

288 from InvenTree import version

289

290 latest = latest if latest else version.inventreeVersionTuple()

291 min_v = version.inventreeVersionTuple(self.MIN_VERSION)

292 max_v = version.inventreeVersionTuple(self.MAX_VERSION)

293

294 return bool(min_v <= latest <= max_v)

295

296

297class InvenTreePlugin(VersionMixin, MixinBase, MetaBase):

298 """The InvenTreePlugin class is used to integrate with 3rd party software.

299

300 DO NOT USE THIS DIRECTLY, USE plugin.InvenTreePlugin

301 """

302

303 AUTHOR = None

304 DESCRIPTION = None

305 PUBLISH_DATE = None

306 VERSION = None

307 WEBSITE = None

308 LICENSE = None

309

310 # Optional path to a JavaScript file which will be loaded in the admin panel

311 # This file must provide a function called renderPluginSettings

312 ADMIN_SOURCE = None

313

314 def __init__(self):

315 """Init a plugin.

316

317 Set paths and load metadata.

318 """

319 super().__init__()

320 self.add_mixin(PluginMixinEnum.BASE)

321

322 self.define_package()

323

324 def __init_subclass__(cls):

325 """Custom code to initialize a subclass of InvenTreePlugin.

326

327 This is a security measure to prevent plugins from overriding methods

328 which are decorated with @mark_final.

329 """

330 final_methods = get_final_methods(InvenTreePlugin)

331

332 child_methods = [

333 name for name, method in cls.__dict__.items() if is_method_like(method)

334 ]

335

336 for name in child_methods:

337 if name in final_methods: 337 ↛ 338line 337 didn't jump to line 338 because the condition on line 337 was never true

338 raise TypeError(

339 'INVE-E11: '

340 + f"Plugin '{cls.__name__}' cannot override final method '{name}' from InvenTreePlugin."

341 )

342

343 return super().__init_subclass__()

344

345 @mark_final

346 @classmethod

347 def file(cls) -> Path:

348 """File that contains plugin definition."""

349 return Path(inspect.getfile(cls))

350

351 @mark_final

352 @classmethod

353 def path(cls) -> Path:

354 """Path to plugins base folder."""

355 return cls.file().parent

356

357 def _get_value(self, meta_name: str, package_name: str) -> str:

358 """Extract values from class meta or package info.

359

360 Args:

361 meta_name (str): Name of the class meta to use.

362 package_name (str): Name of the package data to use.

363

364 Returns:

365 str: Extracted value, None if nothing found.

366 """

367 val = getattr(self, meta_name, None)

368 if not val:

369 val = self.package.get(package_name, None)

370 return val

371

372 # region properties

373 @property

374 @mark_final

375 def description(self):

376 """Description of plugin."""

377 description = self._get_value('DESCRIPTION', 'description')

378 if not description: 378 ↛ 379line 378 didn't jump to line 379 because the condition on line 378 was never true

379 description = self.plugin_name()

380 return description

381

382 @property

383 @mark_final

384 def author(self):

385 """Author of plugin - either from plugin settings or git."""

386 author = self._get_value('AUTHOR', 'author')

387 if not author: 387 ↛ 388line 387 didn't jump to line 388 because the condition on line 387 was never true

388 author = _('No author found') # pragma: no cover

389 return author

390

391 @property

392 @mark_final

393 def pub_date(self):

394 """Publishing date of plugin - either from plugin settings or git."""

395 pub_date = getattr(self, 'PUBLISH_DATE', None)

396 if not pub_date: 396 ↛ 399line 396 didn't jump to line 399 because the condition on line 396 was always true

397 pub_date = self.package.get('date')

398 else:

399 pub_date = datetime.fromisoformat(str(pub_date))

400

401 return pub_date

402

403 @property

404 @mark_final

405 def version(self):

406 """Version of plugin."""

407 return self._get_value('VERSION', 'version')

408

409 @property

410 @mark_final

411 def website(self):

412 """Website of plugin - if set else None."""

413 return self._get_value('WEBSITE', 'website')

414

415 @property

416 @mark_final

417 def license(self):

418 """License of plugin."""

419 return self._get_value('LICENSE', 'license')

420

421 # endregion

422

423 @classmethod

424 @mark_final

425 def check_is_package(cls):

426 """Is the plugin delivered as a package."""

427 return getattr(cls, 'is_package', False)

428

429 @property

430 @mark_final

431 def _is_package(self):

432 """Is the plugin delivered as a package."""

433 return getattr(self, 'is_package', False)

434

435 @classmethod

436 @mark_final

437 def check_is_sample(cls) -> bool:

438 """Is this plugin part of the samples?"""

439 return str(cls.check_package_path()).startswith('plugin/samples/')

440

441 @property

442 @mark_final

443 def is_sample(self) -> bool:

444 """Is this plugin part of the samples?"""

445 return self.check_is_sample()

446

447 @classmethod

448 @mark_final

449 def check_is_builtin(cls) -> bool:

450 """Determine if a particular plugin class is a 'builtin' plugin."""

451 return str(cls.check_package_path()).startswith('plugin/builtin')

452

453 @mark_final

454 def is_builtin(self) -> bool:

455 """Is this plugin is builtin."""

456 return self.check_is_builtin()

457

458 @mark_final

459 def is_mandatory(self) -> bool:

460 """Is this plugin mandatory (always forced to be active)."""

461 config = self.plugin_config()

462 if config: 462 ↛ 466line 462 didn't jump to line 466 because the condition on line 462 was always true

463 # If the plugin is configured, check if it is marked as mandatory

464 return config.is_mandatory()

465

466 return False # pragma: no cover

467

468 @classmethod

469 @mark_final

470 def check_package_path(cls):

471 """Path to the plugin."""

472 if cls.check_is_package(): 472 ↛ 473line 472 didn't jump to line 473 because the condition on line 472 was never true

473 return cls.__module__ # pragma: no cover

474

475 try:

476 return cls.file().relative_to(settings.BASE_DIR)

477 except ValueError:

478 return cls.file()

479

480 @property

481 @mark_final

482 def package_path(self):

483 """Path to the plugin."""

484 return self.check_package_path()

485

486 @classmethod

487 @mark_final

488 def check_package_install_name(cls) -> str | None:

489 """Installable package name of the plugin.

490

491 e.g. if this plugin was installed via 'pip install <x>',

492 then this function should return '<x>'

493

494 Returns:

495 str: Install name of the package, else None

496 """

497 return getattr(cls, 'package_name', None)

498

499 @property

500 @mark_final

501 def package_install_name(self) -> str | None:

502 """Installable package name of the plugin.

503

504 e.g. if this plugin was installed via 'pip install <x>',

505 then this function should return '<x>'

506

507 Returns:

508 str: Install name of the package, else None

509 """

510 return self.check_package_install_name()

511

512 @property

513 @mark_final

514 def settings_url(self) -> str:

515 """URL to the settings panel for this plugin."""

516 if config := self.db: 516 ↛ 518line 516 didn't jump to line 518 because the condition on line 516 was always true

517 return InvenTree.helpers.pui_url(f'/settings/admin/plugin/{config.pk}/')

518 return InvenTree.helpers.pui_url('/settings/admin/plugin/')

519

520 # region package info

521 @mark_final

522 def _get_package_commit(self):

523 """Get last git commit for the plugin."""

524 return get_git_log(str(self.file()))

525

526 @classmethod

527 @mark_final

528 def is_editable(cls):

529 """Returns if the current part is editable."""

530 from distutils.sysconfig import get_python_lib

531

532 pkg_name = cls.__name__.split('.')[0]

533 dist_info = list(Path(get_python_lib()).glob(f'{pkg_name}-*.dist-info'))

534 return bool(len(dist_info) == 1)

535

536 @classmethod

537 @mark_final

538 def _get_package_metadata(cls):

539 """Get package metadata for plugin."""

540 # Try simple metadata lookup

541 try:

542 meta = metadata(cls.__name__)

543 # Simple lookup did not work - get data from module

544 except PackageNotFoundError:

545 try:

546 meta = metadata(cls.__module__.split('.')[0])

547 except PackageNotFoundError:

548 # Not much information we can extract at this point

549 return {}

550

551 try:

552 website = meta['Project-URL'].split(', ')[1]

553 except Exception:

554 website = meta.get('Project-URL')

555

556 return {

557 'author': meta.get('Author-email'),

558 'description': meta.get('Summary'),

559 'version': meta.get('Version'),

560 'website': website,

561 'license': meta.get('License'),

562 }

563

564 def define_package(self):

565 """Add package info of the plugin into plugins context."""

566 try:

567 package = (

568 self._get_package_metadata()

569 if self._is_package

570 else self._get_package_commit()

571 )

572 except TypeError:

573 package = {}

574

575 # process date

576 date = package.get('date')

577 if date: 577 ↛ 578line 577 didn't jump to line 578 because the condition on line 577 was never true

578 package['date'] = datetime.fromisoformat(date)

579

580 # set variables

581 self.package = package

582

583 # endregion

584

585 def get_static_path(self) -> list[str]:

586 """Return the path components to the plugin's static files."""

587 return ['plugins', self.slug]

588

589 def hashed_file_lookup(self, *args) -> str | None:

590 """Find a hashed version of the given file, if it exists.

591

592 This is used to support cache busting for static files.

593

594 Arguments:

595 *args: Path components to the static file (e.g. 'js', 'admin.js')

596

597 Returns:

598 str: Path to the hashed version of the file if it exists, else None

599 """

600 storage = StaticFilesStorage()

601

602 # First, try to find a manifest file which maps original filenames to hashed filenames

603 # We only support vite manifest files - as generated by the plugin creator framework

604 manifest_file = str(Path(*self.get_static_path(), '.vite', 'manifest.json'))

605

606 if not storage.exists(manifest_file):

607 return None

608

609 # Read the contents of the manifest file

610 try:

611 with storage.open(manifest_file) as f:

612 manifest = json.load(f)

613 except json.JSONDecodeError:

614 logger.error(f"Failed to parse manifest file for plugin '{self.SLUG}'")

615 return None

616

617 # Find the entry associated with the requested file

618 # Remove the file extension, as the manifest may contain hashed files with different extensions (e.g. .js, .css)

619 filename = str(args[-1] or '').split('.')[0]

620 pattern = re.compile(rf'{re.escape(filename)}\.(js|jsx|tsx)')

621

622 for key, value in manifest.items():

623 if re.search(pattern, key):

624 return value.get('file')

625

626 return None

627

628 @mark_final

629 def plugin_static_file(

630 self, *args, check_exists: bool = True, check_hash: bool = True

631 ) -> str:

632 """Construct a path to a static file within the plugin directory.

633

634 Arguments:

635 *args: Path components to the static file (e.g. 'js', 'admin.js')

636 check_exists: If True, will check if the file actually exists on disk

637 check_hash: If True, will fallback to checking if the file has a hash in its name (for cache busting)

638

639 - This will return a URL can be used to access the static file

640 - The path is constructed using the STATIC_URL setting and the plugin slug

641 - Note: If the plugin is selected for "development" mode, the path will point to a vite server URL

642

643 Hash Checking:

644

645 - Plugins may distribute static files with a hash in the filename for cache busting purposes (e.g. 'file-abc123.js').

646 - If available, this file is priorities, and the non-hashed version is ignored.

647 - If no hashed file is available, the non-hashed version will be used (if it exists).

648 - If check_hash is False, the non-hashed version of the file will be used (even if a hashed version exists).

649

650 """

651 from django.conf import settings

652 from django.contrib.staticfiles.storage import StaticFilesStorage

653

654 # If the plugin is selected for development mode, use the development host

655 # This allows the plugin developer to run a local vite server and have the plugin load files directly from that server

656 if (

657 settings.DEBUG

658 and settings.PLUGIN_DEV_HOST

659 and settings.PLUGIN_DEV_SLUG

660 and self.SLUG == settings.PLUGIN_DEV_SLUG

661 ):

662 pathname = '/'.join(list(args))

663 url = f'{settings.PLUGIN_DEV_HOST}/src/{pathname}'

664 url = url.replace('.js', '.tsx')

665 return url

666

667 storage = StaticFilesStorage()

668

669 file_name = args[-1] or ''

670

671 # The file may be specified with a function, e.g. 'file.js:renderFunction'

672 if ':' in file_name:

673 file_name, function_name = file_name.split(':')[:2]

674 else:

675 function_name = ''

676

677 # Determine the preceding path to the file (if any)

678 file_path = args[:-1]

679

680 # If enabled, check for a hashed version of the file

681 if check_hash:

682 file_name = self.hashed_file_lookup(*file_path, file_name) or file_name

683

684 full_path = str(Path(*self.get_static_path(), *file_path, file_name))

685

686 if check_exists:

687 if not storage.exists(full_path):

688 logger.error(

689 f"Static file not found for plugin '{self.SLUG}': {full_path}"

690 )

691

692 # Resolve the URL to the static file

693 url = storage.url(full_path)

694

695 # Re-append the function name (if provided)

696 if function_name:

697 url += f':{function_name}'

698

699 return url

700

701 def get_admin_source(self) -> str | None:

702 """Return a path to a JavaScript file which contains custom UI settings.

703

704 The frontend code expects that this file provides a function named 'renderPluginSettings'.

705 """

706 if not self.ADMIN_SOURCE: 706 ↛ 709line 706 didn't jump to line 709 because the condition on line 706 was always true

707 return None

708

709 return self.plugin_static_file(self.ADMIN_SOURCE)

710

711 def get_admin_context(self) -> dict | None:

712 """Return a context dictionary for the admin panel settings.

713

714 This is an optional method which can be overridden by the plugin.

715 """

716 return None