Coverage for src/backend/InvenTree/InvenTree/metadata.py: 82%

1"""Custom metadata for DRF."""

3from django.core.exceptions import PermissionDenied

4from django.http import Http404

5from django.urls import reverse

7import structlog

8from rest_framework import exceptions, permissions, serializers

9from rest_framework.fields import empty

10from rest_framework.metadata import SimpleMetadata

11from rest_framework.request import clone_request

12from rest_framework.utils import model_meta

14import common.models

15import InvenTree.permissions

16from InvenTree.helpers import str2bool

17from InvenTree.serializers import DependentField

18from users.permissions import check_user_permission, check_user_role

20logger = structlog.get_logger('inventree')

23class InvenTreeMetadata(SimpleMetadata):

24 """Custom metadata class for the DRF API.

26 This custom metadata class limits the available "actions",

27 based on the user's role permissions.

29 Thus when a client send an OPTIONS request to an API endpoint,

30 it will only receive a list of actions which it is allowed to perform!

32 Additionally, we include some extra information about database models,

33 so we can perform lookup for ForeignKey related fields.

34 """

36 def determine_actions(self, request, view):

37 """Determine the 'actions' available to the user for the given view.

39 Note that this differs from the standard DRF implementation,

40 in that we also allow annotation for the 'GET' method.

42 This allows the client to determine what fields are available,

43 even if they are only for a read (GET) operation.

45 See SimpleMetadata.determine_actions for more information.

46 """

47 from InvenTree.api import BulkUpdateMixin

49 actions = {}

51 for method in {'PUT', 'POST', 'GET'} & set(view.allowed_methods):

52 view.request = clone_request(request, method)

54 # Mark this request, to prevent expensive prefetching

55 view.request._metadata_requested = True

57 try:

58 # Test global permissions

59 if hasattr(view, 'check_permissions'): 59 ↛ 62line 59 didn't jump to line 62 because the condition on line 59 was always true

60 view.check_permissions(view.request)

61 # Test object permissions

62 if method == 'PUT' and hasattr(view, 'get_object'):

63 if not issubclass(view.__class__, BulkUpdateMixin):

64 # Bypass the get_object method for the BulkUpdateMixin

65 view.get_object()

66 except (exceptions.APIException, PermissionDenied, Http404):

67 pass

68 else:

69 # If user has appropriate permissions for the view, include

70 # appropriate metadata about the fields that should be supplied.

71 serializer = view.get_serializer()

72 actions[method] = self.get_serializer_info(serializer)

73 finally:

74 view.request = request

76 return actions

78 def determine_metadata(self, request, view):

79 """Overwrite the metadata to adapt to the request user."""

80 self.request = request

81 self.view = view

83 metadata = super().determine_metadata(request, view)

85 """

86 Custom context information to pass through to the OPTIONS endpoint,

87 if the "context=True" is supplied to the OPTIONS request

89 Serializer class can supply context data by defining a get_context_data() method (no arguments)

90 """

92 context = {}

94 if str2bool(request.query_params.get('context', False)): 94 ↛ 95line 94 didn't jump to line 95 because the condition on line 94 was never true

95 if hasattr(self, 'serializer') and hasattr(

96 self.serializer, 'get_context_data'

97 ):

98 context = self.serializer.get_context_data()

100 metadata['context'] = context

101

102 user = request.user

103

104 if user is None: 104 ↛ 106line 104 didn't jump to line 106 because the condition on line 104 was never true

105 # No actions for you!

106 metadata['actions'] = {}

107 return metadata

108

109 try:

110 # Extract the model name associated with the view

111 self.model = InvenTree.permissions.get_model_for_view(view)

112

113 # Construct the 'table name' from the model

114 tbl_label = self.model._meta.model_name

115 metadata['model'] = tbl_label

116

117 actions = metadata.get('actions', None)

118

119 if actions is None: 119 ↛ 120line 119 didn't jump to line 120 because the condition on line 119 was never true

120 actions = {}

121

122 # Map the request method to a permission type

123 rolemap = {**InvenTree.permissions.ACTION_MAP, 'OPTIONS': 'view'}

124

125 # let the view define a custom rolemap

126 if hasattr(view, 'rolemap'): 126 ↛ 127line 126 didn't jump to line 127 because the condition on line 126 was never true

127 rolemap.update(view.rolemap)

128

129 # The view may define a custom role requirement

130 role_required = getattr(view, 'role_required', None)

131

132 # Remove any HTTP methods that the user does not have permission for

133 for method, permission in rolemap.items():

134 # general model / role permission

135 result = check_user_permission(user, self.model, permission) or (

136 role_required and check_user_role(user, role_required, permission)

137 )

138

139 # check if simple IsAuthenticated permission class is used

140 if not result: 140 ↛ 141line 140 didn't jump to line 141 because the condition on line 140 was never true

141 result = (

142 view.permission_classes

143 and len(view.permission_classes) == 1

144 and any(

145 perm

146 in [

147 permissions.IsAuthenticated,

148 InvenTree.permissions.IsAuthenticatedOrReadScope,

149 ]

150 for perm in view.permission_classes

151 )

152 )

153

154 if method in actions and not result: 154 ↛ 155line 154 didn't jump to line 155 because the condition on line 154 was never true

155 del actions[method]

156

157 # Add a 'DELETE' action if we are allowed to delete

158 if 'DELETE' in view.allowed_methods:

159 if check_user_permission(user, self.model, 'delete') or ( 159 ↛ 164line 159 didn't jump to line 164 because the condition on line 159 was always true

160 role_required and check_user_role(user, role_required, 'delete')

161 ):

162 actions['DELETE'] = {}

163

164 metadata['actions'] = actions

165

166 except AttributeError:

167 # We will assume that if the serializer class does *not* have a Meta

168 # then we don't need a permission

169 pass

170

171 return metadata

172

173 def override_value(self, field_name: str, field_key: str, field_value, model_value):

174 """Override a value on the serializer with a matching value for the model.

175

176 Often, the serializer field will point to an underlying model field,

177 which contains extra information (which is translated already).

178

179 Rather than duplicating this information in the serializer, we can extract it from the model.

180

181 This is used to override the serializer values with model values,

182 if (and *only* if) the model value should take precedence.

183

184 The values are overridden under the following conditions:

185 - field_value is None

186 - model_value is callable, and field_value is not (this indicates that the model value is translated)

187 - model_value is not a string, and field_value is a string (this indicates that the model value is translated)

188

189 Args:

190 field_name (str): The name of the field.

191 field_key (str): The property key to override.

192 field_value: The value of the field (if available).

193 model_value: The equivalent value of the model (if available).

194 """

195 if field_value is None and model_value is not None:

196 return model_value

197

198 if model_value is None and field_value is not None: 198 ↛ 199line 198 didn't jump to line 199 because the condition on line 198 was never true

199 return field_value

200

201 # Callable values will be evaluated later

202 if callable(model_value) and not callable(field_value): 202 ↛ 203line 202 didn't jump to line 203 because the condition on line 202 was never true

203 return model_value

204

205 if callable(field_value) and not callable(model_value): 205 ↛ 206line 205 didn't jump to line 206 because the condition on line 205 was never true

206 return field_value

207

208 # Prioritize translated text over raw string values

209 if type(field_value) is str and type(model_value) is not str:

210 return model_value

211

212 return field_value

213

214 def get_serializer_info(self, serializer):

215 """Override get_serializer_info so that we can add 'default' values to any fields whose Meta.model specifies a default value."""

216 self.serializer = serializer

217

218 request = getattr(self, 'request', None)

219

220 serializer_info = super().get_serializer_info(serializer)

221

222 # Look for any dynamic fields which were not available when the serializer was instantiated

223 if hasattr(serializer, 'Meta'):

224 for field_name in serializer.Meta.fields:

225 if field_name in serializer_info:

226 # Already know about this one

227 continue

228

229 if field := getattr(serializer, field_name, None):

230 serializer_info[field_name] = self.get_field_info(field)

231

232 model_class = None

233

234 # Extract read_only_fields and write_only_fields from the Meta class (if available)

235 if meta := getattr(serializer, 'Meta', None):

236 read_only_fields = getattr(meta, 'read_only_fields', [])

237 write_only_fields = getattr(meta, 'write_only_fields', [])

238 else:

239 read_only_fields = []

240 write_only_fields = []

241

242 # Attributes to copy extra attributes from the model to the field (if they don't exist)

243 # Note that the attributes may be named differently on the underlying model!

244 extra_attributes = {

245 'help_text': 'help_text',

246 'max_length': 'max_length',

247 'label': 'verbose_name',

248 }

249

250 try:

251 model_class = serializer.Meta.model

252

253 model_fields = model_meta.get_field_info(model_class)

254

255 if model_default_func := getattr(model_class, 'api_defaults', None):

256 model_default_values = model_default_func(request=request) or {}

257 else:

258 model_default_values = {}

259

260 # Iterate through simple fields

261 for name, field in model_fields.fields.items():

262 if name in serializer_info:

263 if name in read_only_fields:

264 serializer_info[name]['read_only'] = True

265

266 if name in write_only_fields: 266 ↛ 267line 266 didn't jump to line 267 because the condition on line 266 was never true

267 serializer_info[name]['write_only'] = True

268

269 if field.has_default():

270 default = field.default

271

272 if callable(default):

273 try:

274 default = default()

275 except Exception:

276 continue

277

278 serializer_info[name]['default'] = default

279

280 elif name in model_default_values: 280 ↛ 281line 280 didn't jump to line 281 because the condition on line 280 was never true

281 serializer_info[name]['default'] = model_default_values[name]

282

283 for field_key, model_key in extra_attributes.items():

284 field_value = getattr(serializer.fields[name], field_key, None)

285 model_value = getattr(field, model_key, None)

286

287 if value := self.override_value(

288 name, field_key, field_value, model_value

289 ):

290 serializer_info[name][field_key] = value

291

292 # Iterate through relations

293 for name, relation in model_fields.relations.items():

294 if name not in serializer_info:

295 # Skip relation not defined in serializer

296 continue

297

298 if relation.reverse:

299 # Ignore reverse relations

300 continue

301

302 if name in read_only_fields:

303 serializer_info[name]['read_only'] = True

304

305 if name in write_only_fields: 305 ↛ 306line 305 didn't jump to line 306 because the condition on line 305 was never true

306 serializer_info[name]['write_only'] = True

307

308 # Extract and provide the "limit_choices_to" filters

309 # This is used to automatically filter AJAX requests

310 serializer_info[name]['filters'] = (

311 relation.model_field.get_limit_choices_to()

312 )

313

314 for field_key, model_key in extra_attributes.items():

315 field_value = getattr(serializer.fields[name], field_key, None)

316 model_value = getattr(relation.model_field, model_key, None)

317

318 if value := self.override_value(

319 name, field_key, field_value, model_value

320 ):

321 serializer_info[name][field_key] = value

322

323 if name in model_default_values:

324 serializer_info[name]['default'] = model_default_values[name]

325

326 except AttributeError:

327 pass

328

329 # Try to extract 'instance' information

330 instance = None

331

332 # Extract extra information if an instance is available

333 if hasattr(serializer, 'instance'): 333 ↛ 336line 333 didn't jump to line 336 because the condition on line 333 was always true

334 instance = serializer.instance

335

336 if instance is None and model_class is not None:

337 # Attempt to find the instance based on kwargs lookup

338 view = getattr(self, 'view', None)

339 kwargs = getattr(view, 'kwargs', None) if view else None

340

341 if kwargs:

342 pk = None

343

344 for field in ['pk', 'id', 'PK', 'ID']:

345 if field in kwargs:

346 pk = kwargs[field]

347 break

348

349 if issubclass(model_class, common.models.BaseInvenTreeSetting):

350 instance = model_class.get_setting_object(**kwargs, create=False)

351

352 elif pk is not None:

353 try:

354 instance = model_class.objects.get(pk=pk)

355 except (ValueError, model_class.DoesNotExist):

356 pass

357

358 if instance is not None: 358 ↛ 359line 358 didn't jump to line 359 because the condition on line 358 was never true

359 """If there is an instance associated with this API View, introspect that instance to find any specific API info."""

360

361 if hasattr(instance, 'api_instance_filters'):

362 instance_filters = instance.api_instance_filters()

363

364 for field_name, field_filters in instance_filters.items():

365 if field_name not in serializer_info:

366 # The field might be missing, but is added later on

367 # This function seems to get called multiple times?

368 continue

369

370 if 'instance_filters' not in serializer_info[field_name]:

371 serializer_info[field_name]['instance_filters'] = {}

372

373 for key, value in field_filters.items():

374 serializer_info[field_name]['instance_filters'][key] = value

375

376 return serializer_info

377

378 def get_field_info(self, field):

379 """Given an instance of a serializer field, return a dictionary of metadata about it.

380

381 We take the regular DRF metadata and add our own unique flavor

382 """

383 from InvenTree.serializers import OptionalField

384

385 if isinstance(field, OptionalField) or issubclass(

386 field.__class__, OptionalField

387 ):

388 # Rehydrate the OptionalField for proper introspection

389 rehydrated_field = field.serializer_class(**(field.serializer_kwargs or {}))

390 return self.get_field_info(rehydrated_field)

391

392 # Try to add the child property to the dependent field to be used by the super call

393 if self.label_lookup[field] == 'dependent field': 393 ↛ 394line 393 didn't jump to line 394 because the condition on line 393 was never true

394 field.get_child(raise_exception=True)

395

396 # Run super method first

397 field_info = super().get_field_info(field)

398

399 # If a default value is specified for the serializer field, add it!

400 if 'default' not in field_info and field.default != empty:

401 field_info['default'] = field.get_default()

402

403 # Force non-nullable fields to read as "required"

404 # (even if there is a default value!)

405 if ( 405 ↛ 410line 405 didn't jump to line 410 because the condition on line 405 was never true

406 'required' not in field_info

407 and not field.allow_null

408 and not (hasattr(field, 'allow_blank') and field.allow_blank)

409 ):

410 field_info['required'] = True

411

412 # Introspect writable related fields

413 if field_info['type'] == 'field' and not field_info['read_only']:

414 # If the field is a PrimaryKeyRelatedField, we can extract the model from the queryset

415 if isinstance(field, serializers.PrimaryKeyRelatedField) or issubclass(

416 field.__class__, serializers.PrimaryKeyRelatedField

417 ):

418 model = field.queryset.model

419 else:

420 logger.debug(

421 'Could not extract model for: %s -> %s',

422 field_info.get('label'),

423 field,

424 )

425 model = None

426

427 if model:

428 # Mark this field as "related", and point to the URL where we can get the data!

429 field_info['type'] = 'related field'

430 field_info['model'] = model._meta.model_name

431

432 # Special case for special models

433 if field_info['model'] == 'user':

434 field_info['api_url'] = reverse('api-user-list')

435 elif field_info['model'] == 'group':

436 field_info['api_url'] = reverse('api-group-list')

437 elif field_info['model'] == 'contenttype':

438 field_info['api_url'] = reverse('api-contenttype-list')

439 elif hasattr(model, 'get_api_url'): 439 ↛ 442line 439 didn't jump to line 442 because the condition on line 439 was always true

440 field_info['api_url'] = model.get_api_url()

441 else:

442 logger.warning("'get_api_url' method not defined for %s", model)

443 field_info['api_url'] = getattr(model, 'api_url', None)

444

445 # Handle custom 'primary key' field

446 field_info['pk_field'] = getattr(field, 'pk_field', 'pk') or 'pk'

447

448 # Add more metadata about dependent fields

449 if field_info['type'] == 'dependent field': 449 ↛ 450line 449 didn't jump to line 450 because the condition on line 449 was never true

450 field_info['depends_on'] = field.depends_on

451

452 # Extends with extra attributes from the serializer

453 extra_field_attributes = ['allow_blank', 'allow_null']

454

455 for attr in extra_field_attributes:

456 if hasattr(field, attr):

457 field_info[attr] = getattr(field, attr)

458

459 # Extend field info if the field has a get_field_info method

460 if (

461 not field_info.get('read_only')

462 and hasattr(field, 'get_field_info')

463 and callable(field.get_field_info)

464 ):

465 field_info = field.get_field_info(field, field_info)

466

467 return field_info

468

469

470InvenTreeMetadata.label_lookup[DependentField] = 'dependent field'