Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
if getattr(self.SAFRSObject, "allow_client_generated_ids", False) is True:
# todo, this isn't required per the jsonapi spec, doesn't work well and isn't documented, maybe later
id = data.get("id")
self.SAFRSObject.id_type.get_pks(id)
# Create the object instance with the specified id and json data
# If the instance (id) already exists, it will be updated with the data
# pylint: disable=not-callable
instance = self.SAFRSObject(**attributes)
if not instance._s_auto_commit:
#
# The item has not yet been added/commited by the SAFRSBase,
# in that case we have to do it ourselves
#
safrs.DB.session.add(instance)
try:
safrs.DB.session.commit()
except sqlalchemy.exc.SQLAlchemyError as exc:
# Exception may arise when a db constrained has been violated
# (e.g. duplicate key)
safrs.log.warning(str(exc))
raise GenericError(str(exc))
return instance
def duplicate(self):
"""
description: Duplicate an object - copy it and give it a new id
"""
from sqlalchemy.orm.session import make_transient
session = safrs.DB.session
session.expunge(self)
make_transient(self)
self.id = self.id_type()
session.add(self)
session.commit()
response = SAFRSFormattedResponse(self)
return response
def _s_clone(self, **kwargs):
"""
Clone an object: copy the parameters and create a new id
:param *kwargs: TBD
"""
make_transient(self)
# pylint: disable=attribute-defined-outside-init
self.id = self.id_type()
for parameter in self._s_jsonapi_attrs:
value = kwargs.get(parameter, None)
if value is not None:
setattr(self, parameter, value)
safrs.DB.session.add(self)
if self._s_auto_commit:
safrs.DB.session.commit()
return self
# db_args now contains the class attributes. Initialize the DB model with them
# All subclasses should have the DB.Model as superclass.
# (SQLAlchemy doesn't work when using DB.Model as SAFRSBase superclass)
try:
safrs.DB.Model.__init__(self, **db_args)
except Exception as exc:
# OOPS .. things are going bad, this might happen using sqla automap
safrs.log.error("Failed to instantiate {}".format(self))
safrs.log.debug("db args: {}".format(db_args))
safrs.log.exception(exc)
safrs.DB.Model.__init__(self)
if self._s_auto_commit:
# Add the object to the database if specified by the class parameters
safrs.DB.session.add(self)
try:
safrs.DB.session.commit()
except sqlalchemy.exc.SQLAlchemyError as exc:
# Exception may arise when a DB constrained has been violated (e.g. duplicate key)
raise GenericError(exc)
message = exc.message
except werkzeug.exceptions.NotFound:
status_code = 404
message = "Not Found"
except Exception as exc:
status_code = getattr(exc, "status_code", 500)
traceback.print_exc()
safrs.log.error(exc)
if safrs.log.getEffectiveLevel() > logging.DEBUG:
message = "Logging Disabled"
else:
message = str(exc)
safrs.DB.session.rollback()
safrs.log.error(message)
errors = dict(detail=message)
abort(status_code, errors=[errors])
200 OK
A server MUST return a 200 OK status code if a deletion request
is successful and the server responds with only top-level meta data.
404 NOT FOUND
A server SHOULD return a 404 Not Found status code
if a deletion request fails due to the resource not existing.
"""
id = kwargs.get(self.object_id, None)
if not id:
# This endpoint shouldn't be exposed so this code is not reachable
raise ValidationError("", status_code=HTTPStatus.METHOD_NOT_ALLOWED)
instance = self.SAFRSObject.get_instance(id)
safrs.DB.session.delete(instance)
return {}, HTTPStatus.NO_CONTENT