From 4bf684f4b576620005a6dd32ec2954b2ac7f0d4e Mon Sep 17 00:00:00 2001 From: Queue A Date: Fri, 14 Aug 2026 08:30:42 +0200 Subject: [PATCH] syncapi now also supports async. --- asyncron/syncapi.py | 190 +++++++++++++++++++++++++++++++------------- 1 file changed, 134 insertions(+), 56 deletions(-) diff --git a/asyncron/syncapi.py b/asyncron/syncapi.py index 4258daf..2e92c67 100644 --- a/asyncron/syncapi.py +++ b/asyncron/syncapi.py @@ -4,11 +4,10 @@ # Automatically generates urlpatterns from signatures and guards # ## - import logging; logger = logging.getLogger(__name__) import collections, functools, inspect import json - +from asgiref.sync import iscoroutinefunction from django.core.serializers.json import DjangoJSONEncoder from django.views.decorators.csrf import csrf_exempt from django.http import HttpResponse, JsonResponse @@ -20,94 +19,173 @@ class CustomJSONEncoder( DjangoJSONEncoder ): 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) +class Endpoint: - @forced_identity #no point messing with the original function - def decorator( f ): - f_args_specs = inspect.getfullargspec(f) + route_to_index = {} #path route -> number + route_to_others = collections.defaultdict( dict ) #path router -> decorated_apis[] + bad_method_response = functools.partial( HttpResponse, "Bad Method", status = 400 ) #call it to get a fresh instance - @functools.wraps(f) - @csrf_exempt - def decorated( request, *args, **kwargs ): + def __init__( self, sig, *guard_args ): + self.sig = sig + self.method, self.route = sig.split(' /', 1) + self.guard_args = guard_args - if request.method != method: return HttpResponse("Bad Method", status = 400) - request.is_json = False - if request.body and request.headers['Content-Type'].startswith('application/json'): #Coule be: application/json; charset=utf-8 + #This function returns f on success, no point messing with the function. + def __call__( self, f ): + if self.method in Endpoint.route_to_others.get(self.route, {}): + raise ValueError( f"Endpoint '{self.sig}': method '{self.method}' is already registered for route '{self.route}'" ) - try: request.json = json.loads( request.body ) - except: request.is_json = False - else: request.is_json = True + self.f = f + self.f_args_specs = inspect.getfullargspec(f) + # f and all its guards must agree on flavor (all sync or all async) + named_callables = { f"view '{getattr(f, '__qualname__', f)}'": f } + named_callables.update({ + f"guard #{i} '{getattr(guard, '__qualname__', guard)}'": guard + for i, guard in enumerate(self.guard_args) + }) + self.is_async = self.check_same_flavor( named_callables ) + + build_decorated = self.build_async_decorated if self.is_async else self.build_sync_decorated + decorated = build_decorated() + endoint_path = path(self.route, decorated) + + unsafe_kwargs = set( self.f_args_specs.kwonlyargs ) & set( endoint_path.pattern.converters ) + if unsafe_kwargs: + logger.warning( f"Skipping '{self.sig}' due to Security Issue: Keyword only arguments {unsafe_kwargs} can only be provided from user input." ) + return f + + Endpoint.route_to_others[self.route][self.method] = decorated + + if self.route not in Endpoint.route_to_index: #If it's the first time seeing this route, just append the decorated endpoint + Endpoint.route_to_index[self.route] = len(urlpatterns) + urlpatterns.append( endoint_path ) + return f + + #Every method sharing this route must also be either all sync or async + route_flavors = { m: iscoroutinefunction(view) for m, view in Endpoint.route_to_others[self.route].items() } + if len(set(route_flavors.values())) > 1: + detail = ", ".join( f"{m}={'async' if a else 'sync'}" for m, a in route_flavors.items() ) + raise TypeError( f"Route '{self.route}': all methods sharing a route must be the same flavor (sync/async), got: {detail}" ) + + build_conjoined = self.build_async_conjoined if self.is_async else self.build_sync_conjoined + urlpatterns[ Endpoint.route_to_index[self.route] ] = path(self.route, build_conjoined()) + + return f + + def check_same_flavor( self, named_callables ): + flavors = { label: iscoroutinefunction(fn) for label, fn in named_callables.items() } + distinct = set(flavors.values()) + + if len(distinct) > 1: + detail = ", ".join( f"{label}={'async' if is_async else 'sync'}" for label, is_async in flavors.items() ) + raise TypeError( f"Endpoint '{self.sig}': view and guards must all be sync or all be async, got mixed flavors: {detail}" ) + + return distinct.pop() + + def sanity_check( self, request ): + return request.method == self.method + + def prepare_kwargs( self, request, kwargs ): + request.is_json = False + if request.body and request.headers.get('Content-Type', '').startswith('application/json'): #Coule be: application/json; charset=utf-8 + try: + request.json = json.loads( request.body ) + except (json.JSONDecodeError, UnicodeDecodeError): + 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 + if k in self.f_args_specs.kwonlyargs and k not in kwargs #Still Extra Security }) - extended_args = [] # v for v in kwargs.values() ] + @staticmethod + def finalize_response( response ): + 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 ) + + def build_sync_decorated( self ): + f = self.f + guard_args = self.guard_args + + @functools.wraps(f) + @csrf_exempt + def decorated( request, *args, **kwargs ): + if not self.sanity_check(request): return Endpoint.bad_method_response() + + self.prepare_kwargs( request, kwargs ) + extended_args = [] 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 + return self.finalize_response( response ) + return decorated - if isinstance(response, tuple): - assert len(response) == 2 - status_code, response = response - assert isinstance(status_code, int) #TODO: accept http.HTTPStatus() instances + def build_async_decorated( self ): + f = self.f + guard_args = self.guard_args - else: status_code = 200 + @functools.wraps(f) + @csrf_exempt + async def decorated( request, *args, **kwargs ): + if not self.sanity_check(request): return Endpoint.bad_method_response() - return JsonResponse( response, status = status_code, encoder = CustomJSONEncoder, safe = False ) + self.prepare_kwargs( request, kwargs ) + extended_args = [] + request.guard_blocked = False + for guard in guard_args: + response = await guard(request) + if request.guard_blocked == True: break + extended_args.append( response ) + else: + response = await f( request, *extended_args, **kwargs ) - route_to_others[route][method] = decorated - endoint_path = path(route, decorated) + return self.finalize_response( response ) + return 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 + def build_sync_conjoined( self ): + route = self.route @csrf_exempt def conjoined( request, *args, **kwargs ): try: - decorated = route_to_others[route][request.method] + target = Endpoint.route_to_others[route][request.method] except KeyError: - return HttpResponse("Bad Method", status = 400) + return Endpoint.bad_method_response() else: - return decorated( request, *args, **kwargs ) + return target( request, *args, **kwargs ) + return conjoined - urlpatterns[ route_to_index[route] ] = path(route, conjoined) + def build_async_conjoined( self ): + route = self.route - - - return decorator + @csrf_exempt + async def conjoined( request, *args, **kwargs ): + try: + target = Endpoint.route_to_others[route][request.method] + except KeyError: + return Endpoint.bad_method_response() + else: + return await target( request, *args, **kwargs ) + return conjoined