Skip to content

Instantly share code, notes, and snippets.

@evilUrge
Last active Oct 13, 2020
Embed
What would you like to do?
Django external webhook view
import logging
from json import loads
from django.apps import apps
from django.core.exceptions import FieldDoesNotExist, FieldError
from django.core.serializers import serialize
from django.http import JsonResponse
logger = logging.getLogger(__name__)
globals().update({_class.__name__: _class for _class in apps.get_models()})
def _rebuild_object(model, dict_obj):
"""
Small hack to rebuild objects with foreign keys, by finding their related model and attaching it as the proper value
:param model:
:param dict_obj:
:return:
"""
def _foreign_key_replacement(outer_model):
for field in outer_model._meta.get_fields():
try:
if outer_model.objects.filter(**{field.name: value}).exists():
return outer_model.objects.get(**{field.name: value})
except ValueError:
logger.info(f'No relevant field found - skipping for now')
return False
modified_dict_obj = dict_obj
for key, value in dict_obj.items():
current_attr = getattr(model, key)
if hasattr(current_attr, "field") and current_attr.field.many_to_one:
find_foreign_field = _foreign_key_replacement(current_attr.field.related_model)
if find_foreign_field:
modified_dict_obj.update({key: find_foreign_field})
elif getattr(current_attr.field, "empty_strings_allowed", False):
modified_dict_obj.update({key: ''})
elif getattr(current_attr.field, "null", False):
modified_dict_obj.pop(key)
else:
msg = f"Value:{value} for field:{key} does not exist as a foreign key nor any other special relation."
logger.error(msg)
raise ValueError(msg)
return modified_dict_obj
def _object_seeker(model, find=dict, external_filter=dict):
if external_filter:
external_model = globals().get(external_filter.get('model'))
if external_model and external_filter.get('filter') and external_filter.get(
'attributes') and external_filter.get('field'):
for result in external_model.objects.filter(**external_filter['filter']):
try:
# XXX: what is the purpose of this ?:
# FIXME: F821 (undefined) error
for attribute in external_filter['attributes'].split('.'):
value = getattr(value if locals().get('value') else result, attribute) # NOQA: F821
type(value) in (str, int, float, list, tuple) and find.update({external_filter['field']: value})
except AttributeError as error:
logger.error(str(error))
return model.objects.filter(**find) if find and model.objects.filter(**find).exists() else False
def update(request, model):
"""
FOR INTERNAL USAGE ONLY!
A generic webook that allows requests with the proper body and path to modify existing objects within connect models
:param request: POST ONLY with the following req body:
{
"external_filter": { # OPTIONAL!
"model": "OrganisationBranch", # In order to find more complex relations.
"filter": {"old_api_id": "123"}, # In this example we're gonna get:
"attributes": "organisation.debtor.id" # OrganisationBranch.objects.filter(old_api_id=value)
.organisation.debtor.id
"field": "id" # And append it to the filter part as another {id: value}
}
"find": {var:value, var:value}, # OPTIONAL (via using internal filter OR external)
"update": {"var":"value", "var":"value"},
}
:param model: Name of an existing connect model.
:return: json with the update status and an array of changed internal id's.
"""
model, content = globals().get(model), loads(request.body)
if request.method == "POST" and model and content.get("update") \
and content.get("find") or content.get("external_filter"):
try:
results = _object_seeker(model, content.get("find", {}), content.get("external_filter", {}))
if results:
msg = {"status": f"Update {results.update(**_rebuild_object(model, content['update']))} records",
"id": [entry.id for entry in results]}
logger.info(msg)
return JsonResponse(msg)
msg = {"status": "Failed to find any matching records"}
logger.warning({"request_body": content, **msg})
return JsonResponse(msg)
except (FieldError, FieldDoesNotExist, ValueError) as error:
msg = {"status": "Update 0 records", "error": str(error)}
logger.error({"request_body": content, **msg})
return JsonResponse(msg)
return JsonResponse({"status": "invalid"})
def create(request, model):
"""
FOR INTERNAL USAGE ONLY!
A generic webhook that allows the creation of a new object in an existing model.
:param request: POST ONLY; the body should include an array of items to create
{
data: [
{var: value, var2:value, var3:value},
{var: value, var2:value, var3:value},
{var: value, var2:value, var3:value},
]
}
:param model:Name of an existing connect model.
:return: json with the creation status and an array of created internal id's
"""
import_model, content = globals().get(model), loads(request.body)
if request.method == "POST" and import_model and content.get("data"):
error, results = {"error": []}, []
for new_entry in content["data"]:
try:
record = import_model.objects.create(**_rebuild_object(import_model, new_entry))
record.save()
results.append(record.id)
logger.info(f"Created new record {record.id} in model {model}")
except (TypeError, AttributeError, ValueError) as err:
logger.error({"exception": str(err), "requested_entry": new_entry})
error["error"].append(str(err))
error = error if error["error"] else {}
return JsonResponse({"status": f"Created {results.__len__()} records", "id": results, **error})
return JsonResponse({"status": "invalid"})
def get(request, model):
"""
Getting an object from whatever ORM model
:param request: POST ONLY!
:param model: What model do you want to get data from.
:return: a serialized jsonfied array version of your object.
EX: {"response":[{"id": 2,"name": demo}, {"id": 3,"name": demo1}]}
"""
model, content = globals().get(model), loads(request.body)
if request.method == "POST" and model and content.get("find") or content.get("external_filter"):
try:
results = _object_seeker(model, content.get("find", {}), content.get("external_filter", {}))
if results:
return JsonResponse({"response": loads(serialize("json", results))})
msg = {"status": "Failed to find any matching records"}
logger.warning({"request_body": content, **msg})
return JsonResponse(msg)
except (FieldError, FieldDoesNotExist, ValueError) as error:
msg = {"status": f"Got 0 records", "error": str(error)}
logger.error({"request_body": content, **msg})
return JsonResponse(msg)
return JsonResponse({"status": "invalid"})
def delete(request, model):
model, content = globals().get(model), loads(request.body)
if request.method == "POST" and model and content.get("find") or content.get("external_filter"):
try:
results = _object_seeker(model, content.get("find", {}), content.get("external_filter", {}))
if results and all(results):
[result.delete() for result in results]
return JsonResponse({"status": "Deleted {} records".format(results.count())})
msg = {"status": "Failed to find any matching records"}
logger.warning({"request_body": content, **msg})
return JsonResponse(msg)
except (FieldError, FieldDoesNotExist, ValueError) as error:
msg = {"status": "Failed to delete records", "error": str(error)}
logger.error({"request_body": content, **msg})
return JsonResponse(msg)
return JsonResponse({"status": "invalid"})
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment