Coverage for src/backend/InvenTree/InvenTree/api.py: 70%

1"""Main JSON interface views."""

3import collections

4import json

5from pathlib import Path

7from django.conf import settings

8from django.contrib.contenttypes.models import ContentType

9from django.core.exceptions import ObjectDoesNotExist

10from django.db import transaction

11from django.http import JsonResponse

12from django.urls import path, reverse

13from django.utils.translation import gettext_lazy as _

14from django.views.generic.base import RedirectView

16import structlog

17from django_q.models import OrmQ

18from drf_spectacular.utils import OpenApiParameter, OpenApiResponse, extend_schema

19from rest_framework import permissions, serializers, viewsets

20from rest_framework.generics import GenericAPIView

21from rest_framework.request import clone_request

22from rest_framework.response import Response

23from rest_framework.serializers import ValidationError

24from rest_framework.views import APIView

26import InvenTree.config

27import InvenTree.filters

28import InvenTree.permissions

29import InvenTree.version

30from common.settings import get_global_setting

31from InvenTree import helpers, ready

32from InvenTree.auth_overrides import registration_enabled

33from InvenTree.mixins import ListCreateAPI

34from plugin.serializers import MetadataSerializer

35from users.models import ApiToken

36from users.permissions import check_user_permission, prefetch_rule_sets

38from .helpers import plugins_info, str2bool

39from .helpers_email import is_email_configured

40from .mixins import ListAPI, RetrieveUpdateAPI

41from .status import check_system_health, is_worker_running

42from .version import inventreeApiText

44logger = structlog.get_logger('inventree')

47def read_license_file(path: Path) -> list:

48 """Extract license information from the provided file.

50 Arguments:

51 path: Path to the license file

53 Returns: A list of items containing the license information

54 """

55 # Check if the file exists

56 if not path.exists():

57 logger.error("License file not found at '%s'", path)

58 return []

60 try:

61 data = json.loads(path.read_text(encoding='utf-8'))

62 except Exception as e:

63 logger.exception("Failed to parse license file '%s': %s", path, e)

64 return []

66 output = []

67 names = set()

69 # Ensure we do not have any duplicate 'name' values in the list

70 for entry in data:

71 name = None

72 for key in entry: 72 ↛ 77line 72 didn't jump to line 77 because the loop on line 72 didn't complete

73 if key.lower() == 'name':

74 name = entry[key]

75 break

77 if name is None or name in names:

78 continue

80 names.add(name)

81 output.append({key.lower(): value for key, value in entry.items()})

83 return sorted(output, key=lambda x: x.get('name', '').lower())

86class LicenseViewSerializer(serializers.Serializer):

87 """Serializer for license information."""

89 backend = serializers.ListField(help_text='Backend licenses texts', read_only=True)

90 frontend = serializers.ListField(

91 help_text='Frontend licenses texts', read_only=True

92 )

95class LicenseView(APIView):

96 """Simple JSON endpoint for InvenTree license information."""

98 permission_classes = [InvenTree.permissions.IsAuthenticatedOrReadScope]

100 @extend_schema(responses={200: OpenApiResponse(response=LicenseViewSerializer)})

101 def get(self, request, *args, **kwargs):

102 """Return information about the InvenTree server."""

103 backend = InvenTree.config.get_base_dir().joinpath('InvenTree', 'licenses.txt')

104 frontend = InvenTree.config.get_base_dir().joinpath(

105 'web/static/web/.vite/dependencies.json'

106 )

107 return JsonResponse({

108 'backend': read_license_file(backend),

109 'frontend': read_license_file(frontend),

110 })

111

112

113class VersionViewSerializer(serializers.Serializer):

114 """Serializer for a single version."""

115

116 class VersionSerializer(serializers.Serializer):

117 """Serializer for server version."""

118

119 server = serializers.CharField()

120 api = serializers.IntegerField()

121 commit_hash = serializers.CharField()

122 commit_date = serializers.CharField()

123 commit_branch = serializers.CharField(allow_null=True)

124 python = serializers.CharField()

125 django = serializers.CharField()

126

127 class LinkSerializer(serializers.Serializer):

128 """Serializer for all possible links."""

129

130 doc = serializers.URLField()

131 code = serializers.URLField()

132 app = serializers.URLField()

133 bug = serializers.URLField()

134

135 dev = serializers.BooleanField()

136 up_to_date = serializers.BooleanField()

137 version = VersionSerializer()

138 links = LinkSerializer()

139

140

141class VersionView(APIView):

142 """Simple JSON endpoint for InvenTree version information."""

143

144 permission_classes = [InvenTree.permissions.IsAdminOrAdminScope]

145

146 @extend_schema(responses={200: OpenApiResponse(response=VersionViewSerializer)})

147 def get(self, request, *args, **kwargs):

148 """Return information about the InvenTree server."""

149 return JsonResponse({

150 'dev': InvenTree.version.isInvenTreeDevelopmentVersion(),

151 'up_to_date': InvenTree.version.isInvenTreeUpToDate(),

152 'version': {

153 'server': InvenTree.version.inventreeVersion(),

154 'api': InvenTree.version.inventreeApiVersion(),

155 'commit_hash': InvenTree.version.inventreeCommitHash(),

156 'commit_date': InvenTree.version.inventreeCommitDate(),

157 'commit_branch': InvenTree.version.inventreeBranch(),

158 'python': InvenTree.version.inventreePythonVersion(),

159 'django': InvenTree.version.inventreeDjangoVersion(),

160 },

161 'links': {

162 'doc': InvenTree.version.inventreeDocUrl(),

163 'code': InvenTree.version.inventreeGithubUrl(),

164 'app': InvenTree.version.inventreeAppUrl(),

165 'bug': f'{InvenTree.version.inventreeGithubUrl()}issues',

166 },

167 })

168

169

170class VersionInformationSerializer(serializers.Serializer):

171 """Serializer for a single version."""

172

173 version = serializers.CharField()

174 date = serializers.DateField()

175 gh = serializers.CharField(allow_null=True)

176 text = serializers.ListField(child=serializers.CharField())

177 latest = serializers.BooleanField()

178

179 class Meta:

180 """Meta class for VersionInformationSerializer."""

181

182 fields = '__all__'

183

184

185@extend_schema(

186 parameters=[

187 OpenApiParameter(

188 name='versions',

189 type=int,

190 description='Number of versions to return.',

191 default=10,

192 ),

193 OpenApiParameter(

194 name='start_version',

195 type=int,

196 description='First version to report. Defaults to return the latest {versions} versions.',

197 ),

198 ]

199)

200class VersionTextView(ListAPI):

201 """Simple JSON endpoint for InvenTree version text."""

202

203 serializer_class = VersionInformationSerializer

204

205 permission_classes = [InvenTree.permissions.IsAdminOrAdminScope]

206

207 # Specifically disable pagination for this view

208 pagination_class = None

209

210 def list(self, request, *args, **kwargs):

211 """Return information about the InvenTree server."""

212 versions = request.query_params.get('versions')

213 start_version = request.query_params.get('start_version')

214

215 api_kwargs = {}

216 if versions is not None:

217 api_kwargs['versions'] = int(versions)

218 if start_version is not None:

219 api_kwargs['start_version'] = int(start_version)

220

221 version_data = inventreeApiText(**api_kwargs)

222 return JsonResponse(list(version_data.values()), safe=False)

223

224

225class InfoApiSerializer(serializers.Serializer):

226 """InvenTree server information - some information might be blanked if called without elevated credentials."""

227

228 class SettingsSerializer(serializers.Serializer):

229 """Serializer for InfoApiSerializer."""

230

231 sso_registration = serializers.BooleanField()

232 registration_enabled = serializers.BooleanField()

233 password_forgotten_enabled = serializers.BooleanField()

234

235 class CustomizeSerializer(serializers.Serializer):

236 """Serializer for customize field."""

237

238 logo = serializers.CharField()

239 splash = serializers.CharField()

240 login_message = serializers.CharField(allow_null=True)

241 navbar_message = serializers.CharField(allow_null=True)

242 disable_theme_storage = serializers.BooleanField(default=False)

243

244 server = serializers.CharField(read_only=True)

245 id = serializers.CharField(read_only=True, allow_null=True)

246 version = serializers.CharField(read_only=True)

247 instance = serializers.CharField(read_only=True)

248 apiVersion = serializers.IntegerField(read_only=True) # noqa: N815

249 worker_running = serializers.BooleanField(read_only=True)

250 worker_count = serializers.IntegerField(read_only=True)

251 worker_pending_tasks = serializers.IntegerField(read_only=True)

252 plugins_enabled = serializers.BooleanField(read_only=True)

253 plugins_install_disabled = serializers.BooleanField(read_only=True)

254 active_plugins = serializers.JSONField(read_only=True)

255 email_configured = serializers.BooleanField(read_only=True)

256 debug_mode = serializers.BooleanField(read_only=True)

257 docker_mode = serializers.BooleanField(read_only=True)

258 default_locale = serializers.CharField(read_only=True)

259 customize = CustomizeSerializer(read_only=True)

260 system_health = serializers.BooleanField(read_only=True)

261 database = serializers.CharField(read_only=True)

262 platform = serializers.CharField(read_only=True)

263 installer = serializers.CharField(read_only=True)

264 target = serializers.CharField(read_only=True, allow_null=True)

265 django_admin = serializers.CharField(read_only=True)

266 settings = SettingsSerializer(read_only=True, many=False)

267

268

269class InfoView(APIView):

270 """JSON endpoint for InvenTree server information.

271

272 Use to confirm that the server is running, etc.

273 """

274

275 permission_classes = [InvenTree.permissions.AllowAnyOrReadScope]

276

277 def worker_pending_tasks(self):

278 """Return the current number of outstanding background tasks."""

279 return OrmQ.objects.count()

280

281 @extend_schema(

282 responses={

283 200: OpenApiResponse(

284 response=InfoApiSerializer, description='InvenTree server information'

285 )

286 }

287 )

288 def get(self, request, *args, **kwargs):

289 """Serve current server information."""

290 is_staff = request.user.is_staff

291 if not is_staff and request.user.is_anonymous:

292 # Might be Token auth - check if so

293 is_staff = self.check_auth_header(request)

294

295 data = {

296 'server': 'InvenTree',

297 'id': InvenTree.version.inventree_identifier(),

298 'version': InvenTree.version.inventreeVersion(),

299 'instance': InvenTree.version.inventreeInstanceName(),

300 'apiVersion': InvenTree.version.inventreeApiVersion(),

301 'worker_running': is_worker_running(),

302 'worker_count': settings.Q_CLUSTER['workers'],

303 'worker_pending_tasks': self.worker_pending_tasks(),

304 'plugins_enabled': settings.PLUGINS_ENABLED,

305 'plugins_install_disabled': settings.PLUGINS_INSTALL_DISABLED,

306 'email_configured': is_email_configured(),

307 'debug_mode': settings.DEBUG,

308 'docker_mode': settings.DOCKER,

309 'default_locale': settings.LANGUAGE_CODE,

310 'customize': {

311 'logo': helpers.getLogoImage(),

312 'splash': helpers.getSplashScreen(),

313 'login_message': helpers.getCustomOption('login_message'),

314 'navbar_message': helpers.getCustomOption('navbar_message'),

315 'disable_theme_storage': str2bool(

316 helpers.getCustomOption('disable_theme_storage')

317 ),

318 },

319 'active_plugins': plugins_info(),

320 # Following fields are only available to staff users

321 'system_health': check_system_health() if is_staff else None,

322 'database': InvenTree.version.inventreeDatabase() if is_staff else None,

323 'platform': InvenTree.version.inventreePlatform() if is_staff else None,

324 'installer': InvenTree.config.inventreeInstaller() if is_staff else None,

325 'target': InvenTree.version.inventreeTarget() if is_staff else None,

326 'django_admin': settings.INVENTREE_ADMIN_URL

327 if (is_staff and settings.INVENTREE_ADMIN_ENABLED)

328 else None,

329 'settings': {

330 'sso_registration': registration_enabled('LOGIN_ENABLE_SSO_REG'),

331 'registration_enabled': registration_enabled('LOGIN_ENABLE_REG'),

332 'password_forgotten_enabled': get_global_setting(

333 'LOGIN_ENABLE_PWD_FORGOT'

334 ),

335 },

336 }

337

338 return JsonResponse(data)

339

340 def check_auth_header(self, request):

341 """Check if user is authenticated via a token in the header."""

342 from InvenTree.middleware import get_token_from_request

343

344 if token := get_token_from_request(request): 344 ↛ 346line 344 didn't jump to line 346 because the condition on line 344 was never true

345 # Does the provided token match a valid user?

346 try:

347 token = ApiToken.objects.get(key=token)

348

349 # Check if the token is active and the user is a staff member

350 if token.active and token.user and token.user.is_staff:

351 return True

352 except ApiToken.DoesNotExist:

353 pass

354

355 return False

356

357

358class NotFoundView(APIView):

359 """Simple JSON view when accessing an invalid API view."""

360

361 permission_classes = [

362 permissions.IsAuthenticated,

363 InvenTree.permissions.AllowAnyOrReadScope,

364 ]

365

366 def not_found(self, request):

367 """Return a 404 error."""

368 return JsonResponse(

369 {

370 'detail': _('API endpoint not found'),

371 'url': request.build_absolute_uri(),

372 },

373 status=404,

374 )

375

376 def options(self, request, *args, **kwargs):

377 """Return 404."""

378 return self.not_found(request)

379

380 @extend_schema(exclude=True)

381 def get(self, request, *args, **kwargs):

382 """Return 404."""

383 return self.not_found(request)

384

385 @extend_schema(exclude=True)

386 def post(self, request, *args, **kwargs):

387 """Return 404."""

388 return self.not_found(request)

389

390 @extend_schema(exclude=True)

391 def patch(self, request, *args, **kwargs):

392 """Return 404."""

393 return self.not_found(request)

394

395 @extend_schema(exclude=True)

396 def put(self, request, *args, **kwargs):

397 """Return 404."""

398 return self.not_found(request)

399

400 @extend_schema(exclude=True)

401 def delete(self, request, *args, **kwargs):

402 """Return 404."""

403 return self.not_found(request)

404

405

406class BulkRequestSerializer(serializers.Serializer):

407 """Parameters for selecting items for bulk operations."""

408

409 items = serializers.ListField(

410 label='A list of primary key values',

411 child=serializers.IntegerField(),

412 required=False,

413 )

414

415 filters = serializers.DictField(

416 label='A dictionary of filter values', required=False

417 )

418

419

420class BulkOperationMixin:

421 """Mixin class for handling bulk data operations.

422

423 Bulk operations are implemented for two major reasons:

424 - Speed (single API call vs multiple API calls)

425 - Atomicity (guaranteed that either *all* items are updated, or *none*)

426 """

427

428 def get_bulk_queryset(self, request):

429 """Return a queryset based on the selection made in the request.

430

431 Selection can be made by providing a list of primary key values,

432 which will be used to filter the queryset.

433 """

434 items = request.data.pop('items', None)

435 all_filter = request.GET.get('all', None)

436

437 # Return the base queryset for this model

438 queryset = self.get_queryset()

439

440 if not items and all_filter is None:

441 raise ValidationError({

442 'non_field_errors': _(

443 'List of items must be provided for bulk operation'

444 )

445 })

446

447 if items: 447 ↛ 461line 447 didn't jump to line 461 because the condition on line 447 was always true

448 if type(items) is not list:

449 raise ValidationError({

450 'non_field_errors': _('Items must be provided as a list')

451 })

452

453 # Filter by primary key

454 try:

455 queryset = queryset.filter(pk__in=items)

456 except Exception:

457 raise ValidationError({

458 'non_field_errors': _('Invalid items list provided')

459 })

460

461 if all_filter and not helpers.str2bool(all_filter): 461 ↛ 462line 461 didn't jump to line 462 because the condition on line 461 was never true

462 raise ValidationError({

463 'non_field_errors': _('All filter must only be used with true')

464 })

465

466 if queryset.count() == 0: 466 ↛ 471line 466 didn't jump to line 471 because the condition on line 466 was always true

467 raise ValidationError({

468 'non_field_errors': _('No items match the provided criteria')

469 })

470

471 return queryset

472

473

474class BulkCreateMixin:

475 """Mixin class for enabling 'bulk create' operations for various models.

476

477 Bulk create allows for multiple items to be created in a single API query,

478 rather than using multiple API calls to same endpoint.

479 """

480

481 def create(self, request, *args, **kwargs):

482 """Perform a POST operation against this list endpoint."""

483 data = request.data

484

485 if isinstance(data, list): 485 ↛ 486line 485 didn't jump to line 486 because the condition on line 485 was never true

486 created_items = []

487 errors = []

488 has_errors = False

489

490 # If data is a list, we assume it is a bulk create request

491 if len(data) == 0:

492 raise ValidationError({'non_field_errors': _('No data provided')})

493

494 # validate unique together fields

495 if unique_create_fields := getattr(self, 'unique_create_fields', None):

496 existing = collections.defaultdict(list)

497 for idx, item in enumerate(data):

498 key = tuple(item[v] for v in list(unique_create_fields)) # ty: ignore[not-subscriptable]

499 existing[key].append(idx)

500

501 unique_errors = [[] for _ in range(len(data))]

502 has_unique_errors = False

503 for item in existing.values():

504 if len(item) > 1:

505 has_unique_errors = True

506 error = {}

507 for field_name in unique_create_fields:

508 error[field_name] = [_('This field must be unique.')]

509 for idx in item:

510 unique_errors[idx] = error

511 if has_unique_errors:

512 raise ValidationError(unique_errors)

513

514 with transaction.atomic():

515 for item in data:

516 serializer = self.get_serializer(data=item)

517 if serializer.is_valid():

518 self.perform_create(serializer)

519 created_items.append(serializer.data)

520 errors.append([])

521 else:

522 errors.append(serializer.errors)

523 has_errors = True

524

525 if has_errors:

526 raise ValidationError(errors)

527

528 return Response(created_items, status=201)

529

530 return super().create(request, *args, **kwargs)

531

532

533class BulkUpdateMixin(BulkOperationMixin):

534 """Mixin class for enabling 'bulk update' operations for various models.

535

536 Bulk update allows for multiple items to be updated in a single API query,

537 rather than using multiple API calls to the various detail endpoints.

538

539 Each instance is validated and saved individually, so that any custom save methods are triggered.

540 """

541

542 BULK_ID_FIELD: str = 'pk'

543

544 def validate_update(self, queryset, request) -> None:

545 """Perform validation right before updating.

546

547 Arguments:

548 queryset: The queryset to be updated

549 request: The request object

550

551 Returns:

552 None

553

554 Raises:

555 ValidationError: If the update should not proceed

556 """

557 # Default implementation does nothing

558

559 def filter_update_queryset(self, queryset, request):

560 """Provide custom filtering for the queryset *before* it is updated.

561

562 The default implementation does nothing, just returns the queryset.

563 """

564 return queryset

565

566 def put(self, request, *args, **kwargs):

567 """Perform a PUT operation against this list endpoint.

568

569 Simply redirects to the PATCH method.

570 """

571 return self.patch(request, *args, **kwargs)

572

573 def patch(self, request, *args, **kwargs):

574 """Perform a PATCH operation against this list endpoint.

575

576 Note that the typical DRF list endpoint does not support PATCH,

577 so this method is provided as a custom implementation.

578 """

579 queryset = self.get_bulk_queryset(request)

580 queryset = self.filter_update_queryset(queryset, request)

581

582 self.validate_update(queryset, request)

583

584 # Perform the update operation

585 data = request.data

586

587 # Extract the primary key values up-front:

588 # Each instance is re-fetched from the database immediately before it is

589 # updated, as saving one instance may alter database state which other

590 # instances in the queryset depend on (e.g. MPTT tree structure fields).

591 # Saving a stale instance can result in database corruption (and it must

592 # be the *instance* that is fresh - refresh_from_db is not sufficient here,

593 # as MPTT caches original field values when the instance is loaded).

594 pk_values = sorted(queryset.values_list(self.BULK_ID_FIELD, flat=True))

595

596 instance_data = []

597

598 with transaction.atomic():

599 # Perform object update

600 # Note that we do not perform a bulk-update operation here,

601 # as we want to trigger any custom post_save methods on the model

602

603 # Run validation first

604 for pk in pk_values:

605 try:

606 instance = queryset.select_for_update(of=('self',)).get(**{

607 self.BULK_ID_FIELD: pk

608 })

609 except ObjectDoesNotExist:

610 raise ValidationError({

611 'non_field_errors': _(

612 'Item no longer matches the provided criteria'

613 )

614 })

615

616 serializer = self.get_serializer(instance, data=data, partial=True)

617 serializer.is_valid(raise_exception=True)

618 serializer.save()

619

620 instance_data.append(serializer.data)

621

622 return Response(

623 {'success': 'Updated multiple items', 'items': instance_data}, status=200

624 )

625

626

627class ParameterListMixin:

628 """Mixin class which supports filtering against parametric fields."""

629

630 def filter_queryset(self, queryset):

631 """Perform filtering against parametric fields."""

632 import common.filters

633

634 queryset = super().filter_queryset(queryset)

635

636 # Filter by parametric data

637 queryset = common.filters.filter_parametric_data(

638 queryset, self.request.query_params

639 )

640

641 serializer_class = (

642 getattr(self, 'serializer_class', None) or self.get_serializer_class()

643 )

644

645 model_class = serializer_class.Meta.model

646

647 # Apply ordering based on query parameter

648 queryset = common.filters.order_by_parameter(

649 queryset, model_class, self.request.query_params.get('ordering', None)

650 )

651

652 return queryset

653

654

655class CommonBulkDeleteMixin(BulkOperationMixin):

656 """Helper for creating bulk delete operation on classic cbv and viewsets."""

657

658 def validate_delete(self, queryset, request) -> None:

659 """Perform validation right before deletion.

660

661 Arguments:

662 queryset: The queryset to be deleted

663 request: The request object

664

665 Returns:

666 None

667

668 Raises:

669 ValidationError: If the deletion should not proceed

670 """

671 # Default implementation does nothing

672

673 def filter_delete_queryset(self, queryset, request):

674 """Provide custom filtering for the queryset *before* it is deleted.

675

676 The default implementation does nothing, just returns the queryset.

677 """

678 return queryset

679

680 @extend_schema(request=BulkRequestSerializer)

681 def _delete(self, request, *args, **kwargs):

682 """Perform a DELETE operation against this list endpoint.

683

684 Note that the typical DRF list endpoint does not support DELETE,

685 so this method is provided as a custom implementation.

686 """

687 queryset = self.get_bulk_queryset(request)

688 queryset = self.filter_delete_queryset(queryset, request)

689

690 self.validate_delete(queryset, request)

691

692 # Keep track of how many items we deleted

693 n_deleted = queryset.count()

694

695 with transaction.atomic():

696 # Perform object deletion

697 # Note that we do not perform a bulk-delete operation here,

698 # as we want to trigger any custom post_delete methods on the model

699 for item in queryset:

700 item.delete()

701

702 return Response({'success': f'Deleted {n_deleted} items'}, status=200)

703

704

705class BulkDeleteMixin(CommonBulkDeleteMixin):

706 """Mixin class for enabling 'bulk delete' operations for various models.

707

708 Bulk delete allows for multiple items to be deleted in a single API query,

709 rather than using multiple API calls to the various detail endpoints.

710 """

711

712 @extend_schema(request=BulkRequestSerializer)

713 def delete(self, request, *args, **kwargs):

714 """Perform a DELETE operation against this list endpoint.

715

716 Note that the typical DRF list endpoint does not support DELETE,

717 so this method is provided as a custom implementation.

718 """

719 return self._delete(request, *args, **kwargs)

720

721

722class BulkDeleteViewsetMixin(CommonBulkDeleteMixin, viewsets.GenericViewSet):

723 """Mixin class for enabling 'bulk delete' operations for viewsets."""

724

725 @extend_schema(request=BulkRequestSerializer)

726 def bulk_delete(self, request, *args, **kwargs):

727 """Perform a bulk delete operation.

728

729 Provide either a list of ids (via `items`) or a filter (via `filters`) to select the items to be deleted.

730

731 This action is performed attomically, so either all items will be deleted, or none will be deleted.

732 """

733 return self._delete(request, *args, **kwargs)

734

735

736class ListCreateDestroyAPIView(BulkDeleteMixin, ListCreateAPI):

737 """Custom API endpoint which provides BulkDelete functionality in addition to List and Create."""

738

739

740class APISearchViewSerializer(serializers.Serializer):

741 """Serializer for the APISearchView."""

742

743 search = serializers.CharField()

744 search_regex = serializers.BooleanField(default=False, required=False)

745 search_whole = serializers.BooleanField(default=False, required=False)

746 search_notes = serializers.BooleanField(default=False, required=False)

747 limit = serializers.IntegerField(default=1, required=False)

748 offset = serializers.IntegerField(default=0, required=False)

749

750

751class APISearchView(GenericAPIView):

752 """A general-purpose 'search' API endpoint.

753

754 Returns hits against a number of different models simultaneously,

755 to consolidate multiple API requests into a single query.

756

757 Is much more efficient and simplifies code!

758 """

759

760 permission_classes = [InvenTree.permissions.IsAuthenticatedOrReadScope]

761 serializer_class = APISearchViewSerializer

762

763 def get_result_types(self):

764 """Construct a list of search types we can return."""

765 import build.api

766 import company.api

767 import order.api

768 import part.api

769 import stock.api

770

771 return {

772 'build': build.api.BuildList,

773 'company': company.api.CompanyList,

774 'supplier': company.api.CompanyList,

775 'manufacturer': company.api.CompanyList,

776 'customer': company.api.CompanyList,

777 'manufacturerpart': company.api.ManufacturerPartList,

778 'supplierpart': company.api.SupplierPartList,

779 'part': part.api.PartList,

780 'partcategory': part.api.CategoryList,

781 'purchaseorder': order.api.PurchaseOrderList,

782 'returnorder': order.api.ReturnOrderList,

783 'salesorder': order.api.SalesOrderList,

784 'salesordershipment': order.api.SalesOrderShipmentList,

785 'stockitem': stock.api.StockList,

786 'stocklocation': stock.api.StockLocationList,

787 }

788

789 def get_result_filters(self):

790 """Provide extra filtering options for particular search groups."""

791 return {

792 'supplier': {'is_supplier': True},

793 'manufacturer': {'is_manufacturer': True},

794 'customer': {'is_customer': True},

795 }

796

797 def post(self, request, *args, **kwargs):

798 """Perform search query against available models."""

799 data = request.data

800

801 results = {}

802

803 # These parameters are passed through to the individual queries, with optional default values

804 pass_through_params = {

805 'search': '',

806 'search_regex': False,

807 'search_whole': False,

808 'search_notes': False,

809 'limit': 1,

810 'offset': 0,

811 }

812

813 if 'search' not in data:

814 raise ValidationError({'search': 'Search term must be provided'})

815

816 search_filters = self.get_result_filters()

817

818 # Create a clone of the request object to modify

819 # Use GET method for the individual list views

820 cloned_request = clone_request(request, 'GET')

821

822 # Fetch and cache all groups associated with the current user

823 groups = prefetch_rule_sets(request.user)

824

825 for key, cls in self.get_result_types().items():

826 # Only return results which are specifically requested

827 if key in data: 827 ↛ 828line 827 didn't jump to line 828 because the condition on line 827 was never true

828 params = data[key]

829

830 for k, v in pass_through_params.items():

831 params[k] = request.data.get(k, v)

832

833 # Add in any extra filters for this particular search type

834 if key in search_filters:

835 for k, v in search_filters[key].items():

836 params[k] = v

837

838 # Enforce json encoding

839 params['format'] = 'json'

840

841 # Ignore if the params are wrong

842 if type(params) is not dict:

843 continue

844

845 view = cls()

846

847 # Override regular query params with specific ones for this search request

848 cloned_request._request.GET = params

849 view.request = cloned_request

850 view.format_kwarg = 'format'

851

852 # Check permissions and update results dict with particular query

853 model = view.serializer_class.Meta.model

854

855 if not check_user_permission(

856 request.user, model, 'view', groups=groups

857 ):

858 results[key] = {

859 'error': _('User does not have permission to view this model')

860 }

861 continue

862

863 try:

864 results[key] = view.list(request, *args, **kwargs).data

865 except Exception as exc:

866 results[key] = {'error': str(exc)}

867

868 return Response(results)

869

870

871class GenericMetadataView(RetrieveUpdateAPI):

872 """Metadata for specific instance; see https://docs.inventree.org/en/stable/plugins/metadata/ for more detail on how metadata works. Most core models support metadata."""

873

874 model = None # Placeholder for the model class

875 serializer_class = MetadataSerializer

876 permission_classes = [InvenTree.permissions.ContentTypePermission]

877

878 def get_permission_model(self):

879 """Return the 'permission' model associated with this view."""

880 model_name = self.kwargs.get('model', None)

881

882 if model_name is None:

883 raise ValidationError(

884 "GenericMetadataView called without 'model' URL parameter"

885 ) # pragma: no cover

886

887 model = ContentType.objects.filter(model=model_name).first()

888

889 if model is None:

890 raise ValidationError(

891 f"GenericMetadataView called with invalid model '{model_name}'"

892 ) # pragma: no cover

893

894 return model.model_class()

895

896 def get_queryset(self):

897 """Return the queryset for this endpoint."""

898 model = self.get_permission_model()

899 return model.objects.all()

900

901 def get_serializer(self, *args, **kwargs):

902 """Return MetadataSerializer instance."""

903 is_gen = ready.isGeneratingSchema()

904 # Detect if we are currently generating the OpenAPI schema

905 if self.model is None and not is_gen: 905 ↛ 907line 905 didn't jump to line 907 because the condition on line 905 was always true

906 self.model = self.get_permission_model()

907 if self.model is None and is_gen: 907 ↛ 909line 907 didn't jump to line 909 because the condition on line 907 was never true

908 # Provide a default model for schema generation

909 import users.models

910

911 self.model = users.models.User

912 return MetadataSerializer(self.model, *args, **kwargs)

913

914 def dispatch(self, request, *args, **kwargs):

915 """Override dispatch to set lookup field dynamically."""

916 self.lookup_field = self.kwargs.get('lookup_field', 'pk')

917 self.lookup_url_kwarg = (

918 'lookup_value' if 'lookup_field' in self.kwargs else 'pk'

919 )

920 return super().dispatch(request, *args, **kwargs)

921

922

923class SimpleGenericMetadataView(GenericMetadataView):

924 """Simplified version of GenericMetadataView which always uses 'pk' as the lookup field."""

925

926 def dispatch(self, request, *args, **kwargs):

927 """Override dispatch to set lookup field to 'pk'."""

928 self.lookup_field = 'pk'

929 self.lookup_url_kwarg = None

930 return super().dispatch(request, *args, **kwargs)

931

932 @extend_schema(operation_id='metadata_pk_retrieve')

933 def get(self, request, *args, **kwargs):

934 """Perform a GET request to retrieve metadata for the given object."""

935 return super().get(request, *args, **kwargs)

936

937 @extend_schema(operation_id='metadata_pk_update')

938 def put(self, request, *args, **kwargs):

939 """Perform a PUT request to update metadata for the given object."""

940 return super().put(request, *args, **kwargs)

941

942 @extend_schema(operation_id='metadata_pk_partial_update')

943 def patch(self, request, *args, **kwargs):

944 """Perform a PATCH request to partially update metadata for the given object."""

945 return super().patch(request, *args, **kwargs)

946

947

948class MetadataRedirectView(RedirectView):

949 """Redirect to the generic metadata view for a given model."""

950

951 model_name = None # Placeholder for the model class

952 lookup_field = 'pk'

953 lookup_field_ref = 'pk'

954 permanent = True

955

956 def get_redirect_url(self, *args, **kwargs) -> str | None:

957 """Return the redirect URL for this view."""

958 _kwargs = {

959 'model': self.model_name,

960 'lookup_value': self.kwargs.get(self.lookup_field_ref, None),

961 'lookup_field': self.lookup_field,

962 }

963 return reverse('api-generic-metadata', args=args, kwargs=_kwargs)

964

965

966def meta_path(model, lookup_field: str = 'pk', lookup_field_ref: str = 'pk'):

967 """Helper function for constructing metadata path for a given model.

968

969 Arguments:

970 model: The model class to use

971 lookup_field: The lookup field to use (if not 'pk')

972 lookup_field_ref: The reference name for the lookup field in the request(if not 'pk')

973

974 Returns:

975 A path to the generic metadata view for the given model

976 """

977 if model is None: 977 ↛ 978line 977 didn't jump to line 978 because the condition on line 977 was never true

978 raise ValidationError(

979 "redirect_metadata_view called without 'model' arg"

980 ) # pragma: no cover

981

982 return path(

983 'metadata/',

984 MetadataRedirectView.as_view(

985 model_name=model._meta.model_name,

986 lookup_field=lookup_field,

987 lookup_field_ref=lookup_field_ref,

988 ),

989 )

990

991

992class TreeMixin:

993 """A mixin class for supporting tree-structured data in the API."""

994

995 # Any API view which inherits from this mixin must define a 'model_class' attribute

996 model_class = None

997

998 filter_backends = InvenTree.filters.SEARCH_ORDER_FILTER

999 search_fields = ['name', 'description']

1000 ordering_fields = ['level', 'name', 'subcategories']

1001 ordering_field_aliases = {'level': ['level', 'name'], 'name': ['name', 'level']}

1002 ordering = ['level']

1003

1004 def filter_queryset(self, queryset):

1005 """Filter the queryset, and provide extra support for tree-structured data."""

1006 queryset = super().filter_queryset(queryset)

1007

1008 # If a search term is provided, include all ancestors of matched items in the results

1009 if self.request.query_params.get('search', '').strip():

1010 ancestors = self.model_class.objects.get_queryset_ancestors(

1011 queryset, include_self=True

1012 )

1013 queryset = queryset | ancestors

1014

1015 # If a specific ID is provided to "expand_to", include all ancestors and siblings

1016 if expand_to := self.request.query_params.get('expand_to'): 1016 ↛ 1017line 1016 didn't jump to line 1017 because the condition on line 1016 was never true

1017 try:

1018 target = self.model_class.objects.get(pk=int(expand_to))

1019 target_ancestors = target.get_ancestors(include_self=True)

1020 queryset = queryset | target_ancestors

1021

1022 # We also want to include the "sibling" nodes of the expanded item

1023 siblings = target.get_siblings(include_self=True)

1024 queryset = queryset | siblings

1025

1026 except (self.model_class.DoesNotExist, ValueError):

1027 pass

1028

1029 return queryset.distinct()