-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Add Update CI.yml Update middlewares.py Update middlewares.py
- Loading branch information
1 parent
83ebb0c
commit 5e55de1
Showing
4 changed files
with
195 additions
and
129 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -1,44 +1,68 @@ | ||
from collections.abc import Callable | ||
from typing import Any, Union | ||
from collections.abc import Awaitable, Callable | ||
from typing import Any, TypeAlias | ||
|
||
from asgiref.sync import iscoroutinefunction, sync_to_async | ||
from django.http import HttpRequest, HttpResponse | ||
from django.utils.decorators import sync_and_async_middleware | ||
|
||
TARGET_METHODS = ("PUT", "PATCH") | ||
# Type aliases for improved readability and type clarity | ||
ResponseType: TypeAlias = HttpResponse | Any | ||
RequestHandler: TypeAlias = Callable[[HttpRequest], ResponseType] | ||
AsyncRequestHandler: TypeAlias = Callable[[HttpRequest], Awaitable[ResponseType]] | ||
|
||
TARGET_METHODS = {"PUT", "PATCH"} | ||
|
||
|
||
@sync_and_async_middleware | ||
def process_put_patch( | ||
get_response: Union[ | ||
Callable[[HttpRequest], HttpResponse], Callable[[HttpRequest], Any] | ||
], | ||
) -> Union[Callable[[HttpRequest], Any], Callable[[HttpRequest], HttpResponse]]: | ||
async def async_middleware(request: HttpRequest) -> Union[HttpResponse, Any]: | ||
get_response: RequestHandler | AsyncRequestHandler, | ||
) -> RequestHandler | AsyncRequestHandler: | ||
""" | ||
Middleware to handle PUT and PATCH requests when content type is not JSON. | ||
Converts the request method to POST temporarily to enable file/post data parsing. | ||
Args: | ||
get_response: The next middleware or view handler in the request processing chain. | ||
Returns: | ||
An async or sync middleware function based on the input handler type. | ||
""" | ||
|
||
async def async_middleware(request: HttpRequest) -> ResponseType: | ||
if ( | ||
request.method in TARGET_METHODS | ||
and request.content_type != "application/json" | ||
): | ||
initial_method = request.method | ||
request.method = "POST" | ||
request.META["REQUEST_METHOD"] = "POST" | ||
|
||
# Sync operation made async | ||
await sync_to_async(request._load_post_and_files)() | ||
|
||
# Restore original method | ||
request.META["REQUEST_METHOD"] = initial_method | ||
request.method = initial_method | ||
|
||
return await get_response(request) | ||
|
||
def sync_middleware(request: HttpRequest) -> Union[HttpResponse, Any]: | ||
def sync_middleware(request: HttpRequest) -> ResponseType: | ||
if ( | ||
request.method in TARGET_METHODS | ||
and request.content_type != "application/json" | ||
): | ||
initial_method = request.method | ||
request.method = "POST" | ||
request.META["REQUEST_METHOD"] = "POST" | ||
|
||
# Standard sync file/post data loading | ||
request._load_post_and_files() | ||
|
||
# Restore original method | ||
request.META["REQUEST_METHOD"] = initial_method | ||
request.method = initial_method | ||
|
||
return get_response(request) | ||
|
||
# Choose middleware based on the input handler's async nature | ||
return async_middleware if iscoroutinefunction(get_response) else sync_middleware |
Oops, something went wrong.