"""
Utilities module for Django.
"""
from io import BytesIO
from django.http.response import (
HttpResponse as DjangoHttpResponse,
StreamingHttpResponse as DjangoStreamingHttpResponse,
)
from django.http.request import (
HttpRequest as DjangoHttpRequest,
RawPostDataException,
)
from django.core.files.uploadedfile import SimpleUploadedFile
from django.core.management import execute_from_command_line, call_command
from duck.http.request import HttpRequest
from duck.http.response import (
HttpResponse,
StreamingHttpResponse,
)
from duck.http.fileuploads.handlers import BaseFileUpload
[docs]
def run_from_command_line(command):
"""
Runs any Django management command as if it was run from command line.
Args:
command (str): The command e.g., "manage.py migrate"
"""
cmd = [i.strip() for i in command.split(" ")] if isinstance(command, str) else command
execute_from_command_line(cmd)
[docs]
def run_django_command(command, *args, **kwargs):
"""
Runs any Django management command programmatically.
Args:
command (str): The command name, e.g., "migrate", "makemigrations"
*args: Positional arguments for the command.
**kwargs: Keyword arguments for the command.
"""
call_command(command, *args, **kwargs)
[docs]
def duck_url_to_django_syntax(url: str) -> str:
"""
Converts a Duck URL path to Django's url path.
Args:
url (str): The URL path in duck syntax (e.g., /path/<some_variable>).
Returns:
str: The URL path in explicit Django syntax (e.g., /path/<str:some_variable>).
"""
parts = url.split("/")
output = []
for part in parts:
if "<" in part and ">" in part:
# Replace the part with the converter syntax
converter = "str" # Default converter for strings
var_name = part[part.find("<") + 1:part.find(">")]
output.append(f"<{converter}:{var_name}>")
else:
output.append(part)
return "/".join(output)
[docs]
def duck_to_django_response(response: HttpResponse):
"""
Converts Duck's http response to Django's http response object.
"""
headers = response.title_headers
# Retrieve content type from headers or the automatic calculated content type
content_type = response.get_header(
'content-type',
response.content_type,
)
if "Content-Type" in headers:
# Django response doesn't allow setting both repsponse content_type arg and
# also Content-Type header.
headers.pop("Content-Type")
if not isinstance(response, StreamingHttpResponse):
return DjangoHttpResponse(
content=response.content,
status=response.status_code,
headers=headers,
content_type=content_type,
)
else:
return DjangoStreamingHttpResponse(
streaming_content=response.stream,
status=response.status_code,
headers=headers,
content_type=content_type,
)
[docs]
def to_django_uploadedfile(fileupload: BaseFileUpload) -> SimpleUploadedFile:
"""
Function to convert Duck fileupload to django UploadedFile object
"""
simple_uploadedfile = SimpleUploadedFile(
name=fileupload.filename,
content=fileupload.read(),
content_type=fileupload.content_type if hasattr(fileupload, "content_type") else None,
)
simple_uploadedfile.save = lambda *_, **__: fileupload.save()
return simple_uploadedfile
[docs]
def duck_to_django_request(request):
"""
Converts Duck's HTTP request to Django's HttpRequest.
"""
meta = request.META
# Create a new Django HttpRequest object
new_request = DjangoHttpRequest()
new_request.method = request.method
new_request.path = request.path
new_request.META.update(meta) # Copy META
# Copy GET, POST, FILES data
new_request.GET.update(request.GET)
new_request.POST.update(request.POST)
new_request.FILES.update(request.FILES)
# Assign authentication and session data
new_request.AUTH = request.AUTH
new_request.session = request.SESSION
# Check if 'request.raw' exists and is valid (adding a fallback in case it's missing)
try:
new_request._stream = BytesIO(request.raw)
except AttributeError:
new_request._stream = BytesIO()
# Set request headers (handling case-insensitivity for headers)
for h, v in request.headers.items():
new_request.headers._store[h.lower()] = (h.title(), v)
if hasattr(request, "user"):
new_request.user = request.user
return new_request
[docs]
def get_raw_django_payload(request: DjangoHttpRequest) -> bytes:
"""
Retrieves the raw byte representation of the HTTP header section from a Django request.
This function constructs the raw headers, starting with the request line (method, path,
protocol) and followed by all headers in the request, in the standard HTTP header format.
Args:
request (DjangoHttpRequest): The Django HTTP request object containing the request
and headers.
Returns:
bytes: A byte string representing the raw HTTP header section of the request, including
the request line and all headers. The result will be suitable for use in raw HTTP
request construction or logging.
"""
# Start with the request line (method, path, protocol)
fullpath = request.path
query_string = request.META.get("QUERY_STRING", "")
if query_string:
fullpath += "?" + query_string
topheader = f'{request.method} {fullpath} {request.META.get("SERVER_PROTOCOL", "HTTP/1.1")}\r\n'
raw = bytes(topheader, "utf-8")
# Add headers to the raw request
headers = request.headers
for header, value in headers.items():
# Ensure header title-cased and formatted correctly
raw += bytes(header.title(), "utf-8") + b": " + bytes(value, "utf-8") + b"\r\n"
return raw
[docs]
def django_to_duck_request(django_request: DjangoHttpRequest):
"""
Converts a Django request to a Duck request.
"""
payload = get_raw_django_payload(django_request)
new_request = HttpRequest()
body = b''
try:
body = django_request.body
except RawPostDataException:
pass
# Parse the raw request using Duck's HttpRequest parser
new_request.parse_raw_request(payload + b"\r\n" + body)
# Copy authentication and session data
new_request.AUTH = getattr(django_request, "AUTH", new_request.AUTH)
new_request.SESSION = getattr(django_request, "session", new_request.SESSION)
# Set user if it exists in the Django request
if hasattr(django_request, "user"):
new_request.user = django_request.user
# Return the new Duck HttpRequest object
return new_request
