Added syncapi, made presentation input into a class

This commit is contained in:
aliqandil 2025-03-13 22:51:08 +03:30
parent 74ff118612
commit 1334da35bb
3 changed files with 115 additions and 2 deletions

View File

@ -63,7 +63,7 @@ class BaseModel( models.Model ):
if presentation_name:
assert presentation in self._model_dict_presentations, f"This model '{self.__class__}' does not have a '{presentation_name}' presentation!"
fields.extend( self._model_dict_presentations[presentation_name] )
fields.extend( self._model_dict_presentations[presentation_name].fields )
results = {}
for f in fields:

113
asyncron/syncapi.py Normal file
View File

@ -0,0 +1,113 @@
##
#
# -- syncapi.py
# Automatically generates urlpatterns from signatures and guards
#
##
import logging; logger = logging.getLogger(__name__)
import collections, functools, inspect
import json
from django.core.serializers.json import DjangoJSONEncoder
from django.views.decorators.csrf import csrf_exempt
from django.http import HttpResponse, JsonResponse
from django.urls import path, re_path, include
from django.utils import timezone
class CustomJSONEncoder( DjangoJSONEncoder ):
def default( self, o ):
if isinstance(o, timezone.datetime): o = o.replace( microsecond = 0 ) #IOS can't handle microseconds!
return super().default(o)
def forced_identity( f ):
@functools.wraps(f)
def decorated( x ):
f( x )
return x
return decorated
urlpatterns = []
route_to_index = {} #path route -> number
route_to_others = collections.defaultdict( dict ) #path router -> decorated_apis[]
def Endpoint( sig, *guard_args ):
method, route = sig.split(' /', 1)
@forced_identity #no point messing with the original function
def decorator( f ):
f_args_specs = inspect.getfullargspec(f)
@functools.wraps(f)
@csrf_exempt
def decorated( request, *args, **kwargs ):
if request.method != method: return HttpResponse("Bad Method", 400)
request.is_json = False
if request.body and request.headers['Content-Type'].startswith('application/json'): #Coule be: application/json; charset=utf-8
try: request.json = json.loads( request.body )
except: request.is_json = False
else: request.is_json = True
if isinstance( request.json, dict ):
#Security check bellow (unsafe_kwargs), should make this a non issue
kwargs.update({
k : v
for k, v in request.json.items()
if k in f_args_specs.kwonlyargs
and k not in kwargs #Still Extra Security
})
extended_args = [] # v for v in kwargs.values() ]
request.guard_blocked = False
for guard in guard_args:
response = guard(request)
if request.guard_blocked == True: break
extended_args.append( response )
else:
response = f( request, *extended_args, **kwargs )
if isinstance(response, HttpResponse):
return response
if isinstance(response, tuple):
assert len(response) == 2
status_code, response = response
assert isinstance(status_code, int) #TODO: accept http.HTTPStatus() instances
else: status_code = 200
return JsonResponse( response, status = status_code, encoder = CustomJSONEncoder, safe = False )
route_to_others[route][method] = decorated
endoint_path = path(route, decorated)
unsafe_kwargs = set( f_args_specs.kwonlyargs ) & set( endoint_path.pattern.converters )
if unsafe_kwargs:
logger.warning( f"Skipping '{sig}' due to Security Issue: Keyword only arguments {unsafe_kwargs} can only be provided from user input." )
return
if route not in route_to_index: #If it's the first time seeing this route, just append the decorated endpoint
route_to_index[route] = len(urlpatterns)
urlpatterns.append( endoint_path )
return
@csrf_exempt
def conjoined( request, *args, **kwargs ):
try:
decorated = route_to_others[route][request.method]
except KeyError:
return HttpResponse("Bad Method", 400)
else:
return decorated( request, *args, **kwargs )
urlpatterns[ route_to_index[route] ] = path(route, conjoined)
return decorator

View File

@ -2,7 +2,7 @@ from setuptools import setup, find_packages
setup(
name='asyncron',
version='0.1.7',
version='0.1.8',
packages=find_packages(),
#include_package_data=True, # Include static files from MANIFEST.in
install_requires=[