diff --git a/docs/changelog.rst b/docs/changelog.rst index a8465bf..3c6a006 100644 --- a/docs/changelog.rst +++ b/docs/changelog.rst @@ -15,6 +15,7 @@ Changelog applied. Uninstalling from the same panel removes the PAS plugin and the per-user JWT signing secrets. +- #97 Extract create/update/delete helpers to api/mutation - #96 Add typed exception subclasses for the JSON API error envelope - #95 Extract serialization helpers to api/serialization - #94 Extract registry and settings helpers to api/settings diff --git a/src/senaite/jsonapi/api/__init__.py b/src/senaite/jsonapi/api/__init__.py index 67a268b..5750c13 100644 --- a/src/senaite/jsonapi/api/__init__.py +++ b/src/senaite/jsonapi/api/__init__.py @@ -18,15 +18,11 @@ # Copyright 2017-2025 by it's authors. # Some rights reserved, see README and LICENSE. -import copy import datetime import json -import transaction -from AccessControl import Unauthorized from Acquisition import ImplicitAcquisitionWrapper from bika.lims import api -from bika.lims.utils.analysisrequest import create_analysisrequest as create_ar from DateTime import DateTime from plone import api as ploneapi from plone.behavior.interfaces import IBehaviorAssignable @@ -41,14 +37,8 @@ from senaite.jsonapi.interfaces import IBatch from senaite.jsonapi.interfaces import ICatalog from senaite.jsonapi.interfaces import ICatalogQuery -from senaite.jsonapi.interfaces import ICreate from senaite.jsonapi.interfaces import IDataManager -from senaite.jsonapi.interfaces import IInfo -from senaite.jsonapi.interfaces import IUpdate -from zope.component import getAdapters from zope.component import getMultiAdapter -from zope.component import queryAdapter -from zope.deprecation import deprecate from zope.schema import getFields _marker = object() @@ -103,164 +93,9 @@ def get_batched(portal_type=None, uid=None, endpoint=None, **kw): complete=complete) -# CREATE -def create_items(portal_type=None, uid=None, endpoint=None, **kw): - """ create items - - 1. If the uid is given, get the object and create the content in there - (assumed that it is folderish) - 2. If the uid is 0, the target folder is assumed the portal. - 3. If there is no uid given, the payload is checked for either a key - - `parent_uid` specifies the *uid* of the target folder - - `parent_path` specifies the *physical path* of the target folder - """ - # disable CSRF - req.disable_csrf_protection() - - # destination where to create the content - container = uid and get_object_by_uid(uid) or None - - # extract the data from the request - records = req.get_request_data() - - results = [] - for record in records: - - # get the portal_type - if portal_type is None: - # try to fetch the portal type out of the request data - portal_type = record.pop("portal_type", None) - - if container is None: - # find the container for content creation - container = find_target_container(record) - - # Check if we have a container and a portal_type - if not all([container, portal_type]): - fail(400, "Please provide a container path/uid and portal_type") - - # check if it is allowed to create the portal_type - if not is_creation_allowed(portal_type, container): - fail(401, "Creation of '{}' in '{}' is not allowed".format( - portal_type, api.get_path(container))) - - # create the object and pass in the record data - sp = transaction.savepoint() - try: - obj = create_object(container, portal_type, **record) - results.append(obj) - except Exception as e: - # rollback the subtransaction if an error occurred - # => this ensures that the new generated ID is also rolled back - sp.rollback() - logger.exception("Error while creating object: %s", e) - - if not results: - fail(400, "No Objects could be created") - - return make_items_for(results, endpoint=endpoint) - - -# PATCH (alias for update_items) -def patch_items(portal_type=None, uid=None, endpoint=None, **kw): - return update_items(portal_type=portal_type, uid=uid, endpoint=endpoint, **kw) - - -# PUT (alias for update_items) -def put_items(portal_type=None, uid=None, endpoint=None, **kw): - return update_items(portal_type=portal_type, uid=uid, endpoint=endpoint, **kw) - - -# UPDATE -def update_items(portal_type=None, uid=None, endpoint=None, **kw): - """ update items - - 1. If the uid is given, the user wants to update the object with the data - given in request body - 2. If no uid is given, the user wants to update a bunch of objects. - -> each record contains either an UID, path or parent_path + id - """ - - # disable CSRF - req.disable_csrf_protection() - - # the data to update - records = req.get_request_data() - - # we have an uid -> try to get an object for it - obj = get_object_by_uid(uid) - if obj: - record = records[0] # ignore other records if we got an uid - - # Can this object be updated? - if not is_update_allowed(obj): - fail(401, "Update of {} is not allowed".format(api.get_path(obj))) - - obj = update_object_with_data(obj, record) - return make_items_for([obj], endpoint=endpoint) - - # no uid -> go through the record items - results = [] - for record in records: - obj = get_object_by_record(record) - - # no object found for this record - if obj is None: - continue - - # Can this object be updated? - if not is_update_allowed(obj): - fail(401, "Update of {} is not allowed".format(api.get_path(obj))) - - # update the object with the given record data - obj = update_object_with_data(obj, record) - results.append(obj) - - if not results: - fail(400, "No Objects could be updated") - - return make_items_for(results, endpoint=endpoint) - - -# DELETE -def delete_items(portal_type=None, uid=None, endpoint=None, **kw): - """ delete items - - 1. If the uid is given, we can ignore the request body and delete the - object with the given uid (if the uid was valid). - 2. If no uid is given, the user wants to delete more than one item. - => go through each item and extract the uid. Delete it afterwards. - // we should do this kind of transaction base. So if we can not get an - // object for an uid, no item will be deleted. - 3. we could check if the portal_type matches, just to be sure the user - wants to delete the right content. - """ - - # disable CSRF - req.disable_csrf_protection() - - # try to find the requested objects - objects = find_objects(uid=uid) - - # We don't want to delete the portal object - if filter(lambda o: is_root(o), objects): - fail(400, "Can not delete the portal object") - - results = [] - for obj in objects: - # We deactivate only! - deactivate_object(obj) - - # Extract the data with proper adapters - info = {} - for name, adapter in getAdapters((obj,), IInfo): - info.update(adapter.to_dict()) - results.append(info) - - if not results: - fail(404, "No Objects could be found") - - return results +# Route orchestrators (create_items, patch_items, put_items, +# update_items, delete_items) live in senaite.jsonapi.api.mutation and +# are re-exported at the bottom of this module. def make_items_for(brains_or_objects, endpoint=None, complete=False): @@ -837,72 +672,9 @@ def resource_to_portal_type(resource): return portal_type -def is_creation_allowed(portal_type, container): - """Checks if it is allowed to create the portal type - - :param portal_type: The portal type requested - :type portal_type: string - :container container: The parent of the object to be created - :returns: True if it is allowed to create this object - :rtype: bool - """ - # Do not allow the creation of objects directly inside portal root - if container == api.get_portal(): - return False - - # Do not allow the creation of objects directly inside setup folder - if container == api.get_setup(): - return False - - # Do not allow the update of objects that belong to senaite_setup folder - if container == api.get_senaite_setup(): - return False - - # Check if the portal_type is allowed in the container - container_info = container.getTypeInfo() - if container_info.filter_content_types: - if portal_type not in container_info.allowed_content_types: - return False - - # Look for a create-specific adapter for this portal type and container - adapter = queryAdapter(container, ICreate, name=portal_type) - if adapter: - return adapter.is_creation_allowed() - - return True - - -def is_update_allowed(obj): - """Returns whether the update of the object passed in is supported - - :param obj: The object to be updated - :type obj: ATContentType/DexterityContentType - :returns: True if it is allowed to update this object - :rtype: bool - """ - # Do not allow to update the site itself - if api.is_portal(obj): - return False - - # Do not allow the update of objects that belong to site root folder - parent = api.get_parent(obj) - if api.is_portal(parent): - return False - - # Do not allow the update of objects that belong to setup folder - if parent == api.get_setup(): - return False - - # Do not allow the update of objects that belong to senaite_setup folder - if parent == api.get_senaite_setup(): - return False - - # Look for an update-specific adapter for this object - adapter = queryAdapter(obj, IUpdate) - if adapter: - return adapter.is_update_allowed() - - return True +# is_creation_allowed and is_update_allowed live in +# senaite.jsonapi.api.mutation and are re-exported at the bottom of +# this module. def url_for(endpoint, default=DEFAULT_ENDPOINT, **values): @@ -1071,190 +843,9 @@ def find_objects(uid=None): return objects -def find_target_container(record): - """Locates a target container for the given portal_type and record - - :param record: The dictionary representation of a content object - :type record: dict - :returns: folder which contains the object - :rtype: object - """ - parent_uid = record.pop("parent_uid", None) - parent_path = record.pop("parent_path", None) - - # Try to find the target object - target = None - if parent_uid: - target = get_object_by_uid(parent_uid) - elif parent_path: - target = get_object_by_path(parent_path) - - if not target: - fail(404, "No target container found") - - return target - - -def create_object(container, portal_type, **data): - """Creates an object slug - - :returns: The new created content object - :rtype: object - """ - - if "id" in data: - # always omit the id as senaite LIMS generates a proper one - id = data.pop("id") - logger.warn("Passed in ID '{}' omitted! Senaite LIMS " - "generates a proper ID for you" .format(id)) - - try: - # Is there any adapter registered to handle the creation of this type? - adapter = queryAdapter(container, ICreate, name=portal_type) - if adapter and adapter.is_creation_delegated(): - logger.info("Delegating 'create' operation of '{}' in '{}'".format( - portal_type, api.get_path(container) - )) - return adapter.create_object(**data) - - # Special case for ARs - # => return immediately w/o update - if portal_type == "AnalysisRequest": - # convert physical paths to objects - # NOTE: for all other objects we handle this already in the - # fieldmanager - data = convert_physical_paths_to_objects(data) - obj = create_analysisrequest(container, **data) - # Omit values which are already set through the helper - data = u.omit(data, "SampleType", "Analyses") - # Set the container as the client, as the AR lives in it - data["Client"] = container - return obj - # Standard content creation - else: - # we want just a minimun viable object and set the data later - obj = api.create(container, portal_type) - # obj = api.create(container, portal_type, **data) - except Unauthorized: - fail(401, "You are not allowed to create this content") - - # Update the object with the given data, but omit the id - update_object_with_data(obj, data) - - return obj - - -def create_analysisrequest(container, **data): - """Create a minimun viable AnalysisRequest - - :param container: A single folderish catalog brain or content object - :type container: ATContentType/DexterityContentType/CatalogBrain - """ - container = get_object(container) - request = req.get_request() - return create_ar(container, request, data) - - -def update_object_with_data(content, record): - """Update the content with the record data - - :param content: A single folderish catalog brain or content object - :type content: ATContentType/DexterityContentType/CatalogBrain - :param record: The data to update - :type record: dict - :returns: The updated content object - :rtype: object - :raises: - APIError, - :class:`~plone.jsonapi.routes.exceptions.APIError` - """ - - # ensure we have a full content object - content = get_object(content) - - # Look for an update-specific adapter for this object - adapter = queryAdapter(content, IUpdate) - if adapter: - # Use the adapter to update the object - logger.info("Delegating 'update' operation of '{}'".format( - api.get_path(content) - )) - adapter.update_object(**record) - - else: - # Fall-back to default update machinery - # get the proper data manager - dm = IDataManager(content) - - if dm is None: - fail(400, "Update for this object is not allowed") - - # Bail-out non-update-able fields - purged_records = copy.deepcopy(record) - map(lambda key: purged_records.pop(key, None), SKIP_UPDATE_FIELDS) - - # Iterate through record items - for k, v in purged_records.items(): - try: - success = dm.set(k, v, **record) - except Unauthorized: - fail(401, "Not allowed to set the field '%s'" % k) - except ValueError, exc: - fail(400, str(exc)) - - if success is False: - logger.warning("update_object_with_data::skipping key=%r", k) - continue - - logger.debug("update_object_with_data::field %r updated", k) - - # Validate the entire content object - invalid = api.validate(content) - if invalid: - fail(400, u.to_json(invalid)) - - # do a wf transition - if record.get("transition", None): - t = record.get("transition") - logger.debug(">>> Do Transition '%s' for Object %s", t, content.getId()) - do_transition_for(content, t) - - # reindex the object - content.reindexObject() - return content - - -@deprecate("Use senaite.core.api.validate instead") -def validate_object(brain_or_object, data): - """Validate the entire object - - :param brain_or_object: A single catalog brain or content object - :type brain_or_object: ATContentType/DexterityContentType/CatalogBrain - :param data: The sharing dictionary as returned from the API - :type data: dict - :returns: invalidity status - :rtype: dict - """ - obj = get_object(brain_or_object) - return api.validate(obj) - - -def deactivate_object(brain_or_object): - """Deactivate the given object - - :param brain_or_object: A single catalog brain or content object - :type brain_or_object: ATContentType/DexterityContentType/CatalogBrain - :returns: Nothing - :rtype: None - """ - obj = get_object(brain_or_object) - # we do not want to delete the site root! - if is_root(obj): - fail(401, "Deactivating the Portal is not allowed") - try: - do_transition_for(brain_or_object, "deactivate") - except Unauthorized: - fail(401, "Not allowed to deactivate object '%s'" % obj.getId()) +# find_target_container, create_object, create_analysisrequest, +# update_object_with_data, validate_object and deactivate_object live +# in senaite.jsonapi.api.mutation and are re-exported at the bottom. def is_relationship_object(brain_or_object): @@ -1316,3 +907,16 @@ def make_batch(sequence, size=25, start=0): from senaite.jsonapi.api.serialization import get_children_info # noqa: E402,F401 from senaite.jsonapi.api.serialization import get_file_info # noqa: E402,F401 from senaite.jsonapi.api.serialization import get_workflow_info # noqa: E402,F401 +from senaite.jsonapi.api.mutation import create_items # noqa: E402,F401 +from senaite.jsonapi.api.mutation import patch_items # noqa: E402,F401 +from senaite.jsonapi.api.mutation import put_items # noqa: E402,F401 +from senaite.jsonapi.api.mutation import update_items # noqa: E402,F401 +from senaite.jsonapi.api.mutation import delete_items # noqa: E402,F401 +from senaite.jsonapi.api.mutation import find_target_container # noqa: E402,F401 +from senaite.jsonapi.api.mutation import create_object # noqa: E402,F401 +from senaite.jsonapi.api.mutation import create_analysisrequest # noqa: E402,F401 +from senaite.jsonapi.api.mutation import update_object_with_data # noqa: E402,F401 +from senaite.jsonapi.api.mutation import validate_object # noqa: E402,F401 +from senaite.jsonapi.api.mutation import deactivate_object # noqa: E402,F401 +from senaite.jsonapi.api.mutation import is_creation_allowed # noqa: E402,F401 +from senaite.jsonapi.api.mutation import is_update_allowed # noqa: E402,F401 diff --git a/src/senaite/jsonapi/api/mutation.py b/src/senaite/jsonapi/api/mutation.py new file mode 100644 index 0000000..93813f3 --- /dev/null +++ b/src/senaite/jsonapi/api/mutation.py @@ -0,0 +1,408 @@ +# -*- coding: utf-8 -*- +# +# This file is part of SENAITE.JSONAPI. +# +# SENAITE.JSONAPI is free software: you can redistribute it and/or modify it +# under the terms of the GNU General Public License as published by the Free +# Software Foundation, version 2. +# +# This program is distributed in the hope that it will be useful, but WITHOUT +# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS +# FOR A PARTICULAR PURPOSE. See the GNU General Public License for more +# details. +# +# You should have received a copy of the GNU General Public License along with +# this program; if not, write to the Free Software Foundation, Inc., 51 +# Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA. +# +# Copyright 2017-2026 by it's authors. +# Some rights reserved, see README and LICENSE. + +"""Create / update / delete helpers for the JSON API. + +The three top-level route orchestrators (`create_items`, `update_items`, +`delete_items`) plus the low-level building blocks +(`create_object`, `update_object_with_data`, `deactivate_object`, +`create_analysisrequest`, `find_target_container`, `validate_object`) +and the permission checks (`is_creation_allowed`, `is_update_allowed`). + +Backward compatibility: every name is re-exported from +`senaite.jsonapi.api` so route code (`v1/routes/content.action` looks +up `api.create_items` / `api.update_items` / `api.delete_items` via +getattr) keeps working unchanged. +""" + +import copy + +import transaction +from AccessControl import Unauthorized +from bika.lims import api as bika_api +from bika.lims.utils.analysisrequest import ( + create_analysisrequest as _create_ar, +) + +from senaite.jsonapi import logger +from senaite.jsonapi import request as req +from senaite.jsonapi import underscore as u +from senaite.jsonapi.api import convert_physical_paths_to_objects +from senaite.jsonapi.api import do_transition_for +from senaite.jsonapi.api import find_objects +from senaite.jsonapi.api import get_object +from senaite.jsonapi.api import get_object_by_path +from senaite.jsonapi.api import get_object_by_record +from senaite.jsonapi.api import get_object_by_uid +from senaite.jsonapi.api import is_root +from senaite.jsonapi.api import make_items_for +from senaite.jsonapi.api import SKIP_UPDATE_FIELDS +from senaite.jsonapi.exceptions import BadRequestError +from senaite.jsonapi.exceptions import ForbiddenError +from senaite.jsonapi.exceptions import NotFoundError +from senaite.jsonapi.interfaces import ICreate +from senaite.jsonapi.interfaces import IDataManager +from senaite.jsonapi.interfaces import IInfo +from senaite.jsonapi.interfaces import IUpdate +from zope.component import getAdapters +from zope.component import queryAdapter +from zope.deprecation import deprecate + + +# --------------------------------------------------------------------------- +# Route orchestrators (called by v1/routes/content.action via getattr) +# --------------------------------------------------------------------------- + +def create_items(portal_type=None, uid=None, endpoint=None, **kw): + """Create one or more objects from the request payload. + + Container resolution: + + 1. If `uid` is given and non-zero, treat it as the target folder's + UID. + 2. If `uid` is 0, the portal itself is the target folder. + 3. Otherwise the payload must provide `parent_uid` or `parent_path` + (handled by `find_target_container`). + """ + req.disable_csrf_protection() + + container = uid and get_object_by_uid(uid) or None + records = req.get_request_data() + + results = [] + for record in records: + if portal_type is None: + portal_type = record.pop("portal_type", None) + + if container is None: + container = find_target_container(record) + + if not all([container, portal_type]): + raise BadRequestError( + "Please provide a container path/uid and portal_type") + + if not is_creation_allowed(portal_type, container): + raise ForbiddenError("Creation of '{}' in '{}' is not allowed".format( + portal_type, bika_api.get_path(container))) + + # Wrap each creation in a savepoint so a failure does not leak + # the generated ID (SENAITE increments a counter on create). + sp = transaction.savepoint() + try: + obj = create_object(container, portal_type, **record) + results.append(obj) + except Exception as e: + sp.rollback() + logger.exception("Error while creating object: %s", e) + + if not results: + raise BadRequestError("No Objects could be created") + + return make_items_for(results, endpoint=endpoint) + + +def patch_items(portal_type=None, uid=None, endpoint=None, **kw): + """PATCH is an alias for update_items.""" + return update_items( + portal_type=portal_type, uid=uid, endpoint=endpoint, **kw) + + +def put_items(portal_type=None, uid=None, endpoint=None, **kw): + """PUT is an alias for update_items.""" + return update_items( + portal_type=portal_type, uid=uid, endpoint=endpoint, **kw) + + +def update_items(portal_type=None, uid=None, endpoint=None, **kw): + """Update one or more objects from the request payload. + + 1. If `uid` is given, update that object with the first record in + the body (later records are ignored). + 2. Otherwise, each record must carry a `uid`, `path`, or + `parent_path` + `id` so the object can be resolved. + """ + req.disable_csrf_protection() + + records = req.get_request_data() + + obj = get_object_by_uid(uid) + if obj: + record = records[0] + if not is_update_allowed(obj): + raise ForbiddenError( + "Update of {} is not allowed".format(bika_api.get_path(obj))) + obj = update_object_with_data(obj, record) + return make_items_for([obj], endpoint=endpoint) + + results = [] + for record in records: + obj = get_object_by_record(record) + if obj is None: + continue + if not is_update_allowed(obj): + raise ForbiddenError( + "Update of {} is not allowed".format(bika_api.get_path(obj))) + obj = update_object_with_data(obj, record) + results.append(obj) + + if not results: + raise BadRequestError("No Objects could be updated") + + return make_items_for(results, endpoint=endpoint) + + +def delete_items(portal_type=None, uid=None, endpoint=None, **kw): + """Deactivate one or more objects (SENAITE never hard-deletes). + + 1. If `uid` is given, that single object is the target. + 2. Otherwise the payload identifies each target + (`find_objects` handles it). + + The portal object itself is refused; other roots (setup folders) + fall through to `deactivate_object`, which prevents them from + transitioning via workflow guards. + """ + req.disable_csrf_protection() + + objects = find_objects(uid=uid) + + if filter(lambda o: is_root(o), objects): + raise BadRequestError("Can not delete the portal object") + + results = [] + for obj in objects: + deactivate_object(obj) + + info = {} + for name, adapter in getAdapters((obj,), IInfo): + info.update(adapter.to_dict()) + results.append(info) + + if not results: + raise NotFoundError("No Objects could be found") + + return results + + +# --------------------------------------------------------------------------- +# Low-level building blocks +# --------------------------------------------------------------------------- + +def find_target_container(record): + """Resolve the target container from the record's parent_* keys. + + Reads and pops `parent_uid` or `parent_path` from the record. + Raises NotFoundError when neither resolves to an object. + """ + parent_uid = record.pop("parent_uid", None) + parent_path = record.pop("parent_path", None) + + target = None + if parent_uid: + target = get_object_by_uid(parent_uid) + elif parent_path: + target = get_object_by_path(parent_path) + + if not target: + raise NotFoundError("No target container found") + + return target + + +def create_object(container, portal_type, **data): + """Create a content object inside the container. + + Handles three paths: + + 1. A registered `ICreate` adapter with `is_creation_delegated()` + True fully takes over. + 2. `AnalysisRequest` gets special handling via + `create_analysisrequest` because of the interim / partition + machinery that a plain `bika_api.create` skips. + 3. Everything else: create the minimum viable object and then + apply the record data via `update_object_with_data` so the + fieldmanagers handle coercion. + """ + if "id" in data: + # SENAITE generates its own ID; drop any client-supplied one. + supplied_id = data.pop("id") + logger.warn( + "Passed in ID '{}' omitted! Senaite LIMS generates a proper " + "ID for you".format(supplied_id)) + + try: + adapter = queryAdapter(container, ICreate, name=portal_type) + if adapter and adapter.is_creation_delegated(): + logger.info( + "Delegating 'create' operation of '{}' in '{}'".format( + portal_type, bika_api.get_path(container))) + return adapter.create_object(**data) + + if portal_type == "AnalysisRequest": + # For AR, physical paths get resolved here; for other types + # the fieldmanager handles it. + data = convert_physical_paths_to_objects(data) + obj = create_analysisrequest(container, **data) + data = u.omit(data, "SampleType", "Analyses") + data["Client"] = container + return obj + + # Standard creation: minimum viable object + record data below. + obj = bika_api.create(container, portal_type) + except Unauthorized: + raise ForbiddenError("You are not allowed to create this content") + + update_object_with_data(obj, data) + return obj + + +def create_analysisrequest(container, **data): + """Create a minimum-viable AnalysisRequest via the bika helper.""" + container = get_object(container) + request = req.get_request() + return _create_ar(container, request, data) + + +def update_object_with_data(content, record): + """Update `content` with the fields from `record`. + + Delegates to a registered `IUpdate` adapter if one exists; + otherwise uses the default `IDataManager` machinery. + """ + content = get_object(content) + + adapter = queryAdapter(content, IUpdate) + if adapter: + logger.info( + "Delegating 'update' operation of '{}'".format( + bika_api.get_path(content))) + adapter.update_object(**record) + else: + dm = IDataManager(content) + if dm is None: + raise BadRequestError("Update for this object is not allowed") + + purged = copy.deepcopy(record) + for key in SKIP_UPDATE_FIELDS: + purged.pop(key, None) + + for k, v in purged.items(): + try: + success = dm.set(k, v, **record) + except Unauthorized: + raise ForbiddenError("Not allowed to set the field '%s'" % k) + except ValueError as exc: + raise BadRequestError(str(exc)) + + if success is False: + logger.warning("update_object_with_data::skipping key=%r", k) + continue + logger.debug("update_object_with_data::field %r updated", k) + + invalid = bika_api.validate(content) + if invalid: + raise BadRequestError(u.to_json(invalid)) + + if record.get("transition", None): + t = record.get("transition") + logger.debug(">>> Do Transition '%s' for Object %s", t, content.getId()) + do_transition_for(content, t) + + content.reindexObject() + return content + + +@deprecate("Use senaite.core.api.validate instead") +def validate_object(brain_or_object, data): + """Validate the entire object. Deprecated; use bika_api.validate.""" + obj = get_object(brain_or_object) + return bika_api.validate(obj) + + +def deactivate_object(brain_or_object): + """Deactivate the given object (SENAITE's soft-delete). + + Refuses the site root explicitly; other refusals come from the + workflow guard on `deactivate`. + """ + obj = get_object(brain_or_object) + if is_root(obj): + raise ForbiddenError("Deactivating the Portal is not allowed") + try: + do_transition_for(brain_or_object, "deactivate") + except Unauthorized: + raise ForbiddenError( + "Not allowed to deactivate object '%s'" % obj.getId()) + + +# --------------------------------------------------------------------------- +# Permission checks +# --------------------------------------------------------------------------- + +def is_creation_allowed(portal_type, container): + """True if `portal_type` may be created inside `container`. + + Never allows creation directly in the portal root, the classic + `bika_setup` folder, or the dexterity `senaite_setup` folder. + Falls back to the container's own `allowed_content_types` filter, + then to an optional `ICreate` adapter's opinion. + """ + if container == bika_api.get_portal(): + return False + if container == bika_api.get_setup(): + return False + if container == bika_api.get_senaite_setup(): + return False + + container_info = container.getTypeInfo() + if container_info.filter_content_types: + if portal_type not in container_info.allowed_content_types: + return False + + adapter = queryAdapter(container, ICreate, name=portal_type) + if adapter: + return adapter.is_creation_allowed() + + return True + + +def is_update_allowed(obj): + """True if `obj` may be updated. + + Same denylist as `is_creation_allowed` (portal, bika_setup, + senaite_setup), applied to the object's parent, plus an optional + `IUpdate` adapter's opinion. + """ + if bika_api.is_portal(obj): + return False + + parent = bika_api.get_parent(obj) + if bika_api.is_portal(parent): + return False + if parent == bika_api.get_setup(): + return False + if parent == bika_api.get_senaite_setup(): + return False + + adapter = queryAdapter(obj, IUpdate) + if adapter: + return adapter.is_update_allowed() + + return True diff --git a/src/senaite/jsonapi/tests/doctests/create.rst b/src/senaite/jsonapi/tests/doctests/create.rst index 96ff404..cd48c9a 100644 --- a/src/senaite/jsonapi/tests/doctests/create.rst +++ b/src/senaite/jsonapi/tests/doctests/create.rst @@ -344,7 +344,7 @@ instead of the plone's default creation. Creation restrictions ~~~~~~~~~~~~~~~~~~~~~ -We get a 401 error if we try to create an object inside portal root: +We get a 403 error if we try to create an object inside portal root: >>> data = {"title": "My clients folder", ... "portal_type": "ClientsFolder", @@ -352,9 +352,9 @@ We get a 401 error if we try to create an object inside portal root: >>> post("create", data) Traceback (most recent call last): [...] - HTTPError: HTTP Error 401: Unauthorized + HTTPError: HTTP Error 403: Forbidden -We get a 401 error if we try to create an object inside setup folder: +We get a 403 error if we try to create an object inside setup folder: >>> data = {"title": "My Analysis Categories folder", ... "portal_type": "AnalysisCategories", @@ -362,9 +362,9 @@ We get a 401 error if we try to create an object inside setup folder: >>> post("create", data) Traceback (most recent call last): [...] - HTTPError: HTTP Error 401: Unauthorized + HTTPError: HTTP Error 403: Forbidden -We get a 401 error when we try to create an object from a type that is not +We get a 403 error when we try to create an object from a type that is not allowed by the container: >>> data = {"title": "My Method", @@ -373,4 +373,4 @@ allowed by the container: >>> post("create", data) Traceback (most recent call last): [...] - HTTPError: HTTP Error 401: Unauthorized + HTTPError: HTTP Error 403: Forbidden diff --git a/src/senaite/jsonapi/tests/doctests/update.rst b/src/senaite/jsonapi/tests/doctests/update.rst index 00aa764..b12a332 100644 --- a/src/senaite/jsonapi/tests/doctests/update.rst +++ b/src/senaite/jsonapi/tests/doctests/update.rst @@ -164,16 +164,16 @@ We can update and transition at same time: Update restrictions ~~~~~~~~~~~~~~~~~~~ -We get a 401 error if we try to update an object from inside portal root: +We get a 403 error if we try to update an object from inside portal root: >>> data = {"title": "My clients folder", ... "uid": api.get_uid(clients),} >>> post("update", data) Traceback (most recent call last): [...] - HTTPError: HTTP Error 401: Unauthorized + HTTPError: HTTP Error 403: Forbidden -We get a 401 error if we try to update an object from inside setup folder: +We get a 403 error if we try to update an object from inside setup folder: >>> cats_uid = api.get_uid(portal.setup.analysiscategories) >>> data = {"title": "My Analysis Categories folder", @@ -181,7 +181,7 @@ We get a 401 error if we try to update an object from inside setup folder: >>> post("update", data) Traceback (most recent call last): [...] - HTTPError: HTTP Error 401: Unauthorized + HTTPError: HTTP Error 403: Forbidden We cannot update the `id` of an object: