import sys from server.models.models import Jsonifiable from flask import Flask, request, Response, jsonify from typing import Type, Final, List, Optional from datetime import datetime from neomodel import StructuredNode, RelationshipManager, RelationshipDefinition, db from neomodel.cardinality import ZeroOrOne, ZeroOrMore, One, OneOrMore import json """ This document declares some abstract api functions that will help you implement a REST api for you StructuredNodes fast and efficiently. The simplest way to achieve this is by using the register_api_object(cls: Type[StructuredNode]) method. Example: given the node type: (note that you need to add the mixin Jsonifiable) class Person(StructuredNode, Jsonifiable): uid = UniqueIdProperty() name = StringProperty(required=True) children = RelationshipTo('Person', 'CHILDREN') # you can register the endpoints /api/person and /api/person/ by calling register_api_object(Person) # if you want to preotect some attributes from being set this way, simply include them in the BLOCKED_PROPERTIES global variable in this module """ BLOCKED_PROPERTIES: Final[List[str]] = ['id', 'uid', 'password'] class ApiError(Exception): """Defines an error that occurred during the handling of an API request""" def __init__(self, msg, status=400, data=None): super().__init__() self.msg = msg self.status = status self.data = data def __str__(self): return json.dumps(dict(msg=self.msg, status=self.status)) def to_response(self) -> Response: """Converts the error into a response""" resp: Response = jsonify(msg=self.msg, data=self.data) resp.status_code = self.status return resp @staticmethod def not_found(cls: Type[StructuredNode], uid=None): return ApiError(f"Cannot find {cls.__name__}" + (f" with uid {uid}" if uid else "") + "!", status=404) @staticmethod def bad_request(msg: str, data=None): return ApiError(msg, status=400, data=data) def _connect_callback(attr, target): """returns a callback to connect attr with target, this is wrapped because we need to get attr and target into a separate closure""" return lambda: attr.connect(target) def _get_target_class(rel: RelationshipDefinition) -> Type[StructuredNode]: target_cls = rel._raw_class return getattr(sys.modules[rel.module_name], target_cls) if isinstance(target_cls, str) else target_cls def construct_object_from_request(cls: Type[StructuredNode], uid=None, data: dict = dict()): """Construct an obect of class cls from request data""" # properties that you do not want to initialize or change this way if uid: obj: Final = cls.nodes.get_or_none(uid=uid) if not obj: raise ApiError.not_found(cls, uid) else: obj: Final = cls() relationship_attach_callbacks = list() for name, value in cls.__dict__.items(): if name[0] == '_' or name in BLOCKED_PROPERTIES or not name[0].islower(): continue if name not in data: continue cls_name = value.__class__.__name__ convert_with = None if cls_name == 'StringProperty': convert_with = str elif cls_name == 'DateTimeProperty': convert_with = datetime.fromisoformat elif cls_name == 'DateProperty': convert_with = lambda d: datetime.fromisoformat(d).date() elif cls_name == 'IntegerProperty': convert_with = int elif cls_name == 'FloatProperty': convert_with = float elif cls_name == 'RelationshipDefinition': pass #target_cls: Type[StructuredNode] = _get_target_class(value) # #for val in data[name] if # target = target_cls.nodes.first_or_none(uid=val) # if not target: # raise ApiError(f'Cannot find referenced object uid={val} of type {target_cls.__name__}!') # # relationship_attach_callbacks.append(_connect_callback(getattr(obj, name), target)) if not convert_with: # skip this property continue setattr(obj, name, convert_with(data[name])) return obj, lambda: [c() for c in relationship_attach_callbacks] def handle_object_api_request(cls: Type[StructuredNode]): """basic method for handling (GET|POST) /api/type calls""" if request.method == 'GET': limit = request.args.get('$limit', 100, int) offset = request.args.get('$offset', 0, int) options = {k: v for k, v in request.args.items() if k[0] != '$'} return jsonify([ obj.json(include_relations=False) for obj in cls.nodes.filter(**options).order_by('uid').all()[offset:offset+limit] ]) if request.method == 'POST': with db.transaction: obj, attach_relationships = construct_object_from_request(cls, data=request.get_json()) obj.save() attach_relationships() return jsonify(obj.json()) def handle_object_api_request_id(cls: Type[StructuredNode], uid: str): """basic method for handling (GET|PUT|DELETE) /api/type/uid calls""" if request.method == 'GET': obj = cls.nodes.get_or_none(uid=uid) if not obj: raise ApiError.not_found(cls, uid) return jsonify(obj.json()) if request.method == 'PUT': obj, attach_relationships = construct_object_from_request(cls, uid=uid, data=request.get_json()) obj.save() attach_relationships() return jsonify(obj.json()) if request.method == 'DELETE': obj = cls.nodes.get_or_none(uid=uid) if not obj: raise ApiError.not_found(cls, uid) obj.delete() return '', 204 def handle_object_api_request_for_relation(cls: Type[StructuredNode], uid, relation, reluid): obj = cls.nodes.get_or_none(uid=uid) if not obj: raise ApiError.not_found(cls, uid) rel = getattr(cls, relation, None) if not (rel and isinstance(rel, RelationshipDefinition)): raise ApiError.bad_request(f'Object of type {cls.__name__} has no relation {relation}!') target_cls = _get_target_class(rel) target_obj = target_cls.nodes.get_or_none(uid=reluid) if not target_obj: raise ApiError.not_found(target_cls, reluid) if request.method == 'DELETE': getattr(obj, relation).disconnect(target_obj) elif request.method == 'POST': if isinstance(rel.manager, (ZeroOrOne, One)): getattr(obj, relation).disconnect_all() getattr(obj, relation).connect(target_obj) return jsonify(obj.json()) def register_api_object(app: Flask, cls: Type[StructuredNode], name: Optional[str] = None): """Register api paths for the object type cls routes registered: GET /api/ POST /api/ GET /api// PUT /api// DELETE /api// POST /api//// connect relationships DELETE /api//// disconnect relationships where defaults to the lowercase class name of cls internally it registers the three view functions api., api..with-id and api..rel """ name = name or cls.__name__.lower() def func_handle(): return handle_object_api_request(cls) def func_handle_id(uid): return handle_object_api_request_id(cls, uid) def func_handle_rel(uid, rel, reluid): return handle_object_api_request_for_relation(cls, uid, rel, reluid) app.view_functions[f'api.{name}'] = func_handle app.view_functions[f'api.{name}.wth-id'] = func_handle_id app.view_functions[f'api.{name}.rel'] = func_handle_rel app.add_url_rule(f'/api/{name}', f'api.{name}', func_handle, methods=['GET', 'POST']) app.add_url_rule(f'/api/{name}/', f'api.{name}.with-id', func_handle_id, methods=['GET', 'PUT', 'DELETE']) app.add_url_rule(f'/api/{name}///', f'api.{name}.rel', func_handle_rel, methods=['POST', 'DELETE'])