Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
for rel_name in self._s_relationship_names:
rel_attr = kwargs.get(rel_name, None)
if rel_attr:
# This shouldn't in the work in the web context
# because the relationships should already have been removed by SAFRSRestAPI.post
db_args[rel_name] = rel_attr
# 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)
return attr_val
if attr_val is None:
return attr_val
try:
column.type.python_type
except NotImplementedError as exc:
"""
This happens when a custom type has been implemented, in which case the user/dev should know how to handle it:
override this method and implement the parsing
https://docs.python.org/2/library/exceptions.html#exceptions.NotImplementedError :
In user defined base classes, abstract methods should raise this exception when they require derived classes to override the method.
=> simply return the attr_val for user-defined classes
"""
safrs.log.debug(exc)
return attr_val
"""
Parse datetime and date values for some common representations
If another format is uses, the user should create a custom column type or custom serialization
"""
if attr_val and column.type.python_type == datetime.datetime:
date_str = str(attr_val)
try:
if "." in date_str:
# str(datetime.datetime.now()) => "%Y-%m-%d %H:%M:%S.%f"
attr_val = datetime.datetime.strptime(date_str, "%Y-%m-%d %H:%M:%S.%f")
else:
# JS datepicker format
attr_val = datetime.datetime.strptime(date_str, "%Y-%m-%d %H:%M:%S")
except (NotImplementedError, ValueError) as exc:
instance = self.SAFRSObject
method = getattr(instance, self.method_name, None)
if not method:
# Only call methods for Campaign and not for superclasses (e.g. safrs.DB.Model)
raise ValidationError('Invalid method "{}"'.format(self.method_name))
if not is_public(method):
raise ValidationError("Method is not public")
args = dict(request.args)
payload = request.get_jsonapi_payload()
if payload:
args = payload.get("meta", {}).get("args", {})
safrs.log.debug("method {} args {}".format(self.method_name, args))
result = method(**args)
if isinstance(result, SAFRSFormattedResponse):
response = result
else:
response = {"meta": {"result": result}}
return jsonify(response) # 200 : default
if callable(column.default.arg):
# todo: check how to display the default args
continue
else:
arg = column.default.arg
try:
if column.type.python_type == int:
arg = 0
if column.type.python_type == datetime.datetime:
arg = str(datetime.datetime.now())
elif column.type.python_type == datetime.date:
arg = str(datetime.date.today())
else:
arg = column.type.python_type()
except NotImplementedError:
safrs.log.debug("Failed to get python type for column {} (NotImplementedError)".format(column))
arg = None
except Exception as exc:
safrs.log.debug("Failed to get python type for column {} ({})".format(column, exc))
sample[column.name] = arg
return sample
raise ValidationError("Invalid ID")
else:
# No ID was supplied, apply method to the class itself
instance = self.SAFRSObject
method = getattr(instance, self.method_name, None)
if not method:
# Only call methods for Campaign and not for superclasses (e.g. safrs.DB.Model)
raise ValidationError('Invalid method "{}"'.format(self.method_name))
if not is_public(method):
raise ValidationError("Method is not public")
args = dict(request.args)
safrs.log.debug("method {} args {}".format(self.method_name, args))
result = method(**args)
if isinstance(result, SAFRSFormattedResponse):
response = result
else:
response = {"meta": {"result": result}}
return jsonify(response) # 200 : default
def __init__(self, message, status_code=HTTPStatus.INTERNAL_SERVER_ERROR.value):
Exception.__init__(self)
self.status_code = status_code
if safrs.log.getEffectiveLevel() <= logging.DEBUG:
self.message = str(message)
else:
self.message += HIDDEN_LOG
safrs.log.debug(traceback.format_exc())
safrs.log.error("Generic Error: %s", message)
if relationship in [r.key for r in instance._s_relationships]:
included = getattr(instance, relationship)
# relationship direction in (ONETOMANY, MANYTOMANY):
if included and isinstance(included, SAFRSBase) and included not in result:
# convert single instance to a list so we can generically add the includes
included = [included]
elif isinstance(included, sqlalchemy.orm.collections.InstrumentedList):
pass
elif not included or included in result:
continue
try:
# This works on sqlalchemy.orm.dynamic.AppenderBaseQuery
included = included[:limit]
except Exception as exc:
safrs.log.debug("Failed to add included for {} (included: {} - {}): {}".format(relationship, type(included), included, exc))
try:
result = result.union(included)
except Exception as exc:
safrs.log.warning(
"Failed to unionize included for {} (included: {} - {}): {}".format(relationship, type(included), included, exc)
)
result.add(included)
if INCLUDE_ALL in includes:
for nested_included in [get_included(result, limit, level=level + 1) for obj in result]:
# Removed recursion with get_included(result, limit, INCLUDE_ALL)
result = result.union(nested_included)
elif nested_rel:
for nested_included in [get_included(result, limit, nested_rel, level=level + 1) for obj in result]: