Skip to content

Instantly share code, notes, and snippets.

@rorycl

rorycl/django_endpoint_example.py

Last active July 31, 2025 14:13
Show Gist options

  • Select an option

    Learn more about clone URLs
  • Save rorycl/2d860433944ec7eaa1ca1291ff6b45b9 to your computer and use it in GitHub Desktop.

Select an option

Learn more about clone URLs
Save rorycl/2d860433944ec7eaa1ca1291ff6b45b9 to your computer and use it in GitHub Desktop.
Download ZIP
Raw
django_endpoint_example.py
"""
Defines the Django views for an example search and user list API.
This module contains view functions that handle HTTP requests. It follows a
service-layer architecture, where views are responsible for:
- Handling HTTP-specific concerns (parsing requests, forming responses).
- Validating input using Pydantic schemas.
- Orchestrating calls to the business logic layer (services).
- Handling exceptions and rendering appropriate templates or JSON responses.
"""
import json
import logging
from typing import Union
from django.http import HttpRequest, HttpResponse, JsonResponse
from django.shortcuts import render
from django.views.decorators.http import require_http_methods
from pydantic import ValidationError
# Assumed local imports
from .decorators import session_handler
from .schemas import DocumentSearchSchema, ListItemAddSchema
from .services import (
DatabaseService,
SearchService,
db_exceptions,
search_exceptions,
)
# example setup, with dependency injection.
search_service = SearchService()
db_service = DatabaseService()
log = logging.getLogger(__name__)
@require_http_methods(["GET"])
@session_handler(guest_ok=True)
def document_search(request: HttpRequest) -> HttpResponse:
"""
Searches documents based on query parameters.
Validates search parameters from the request's query string, performs a
search using the SearchService, and renders an HTML results page.
Args:
request: The Django HttpRequest object. Query params are in request.GET.
Returns:
An HttpResponse object, typically a rendered HTML page.
"""
try:
# Validate input using a Pydantic schema. request.GET is a QueryDict.
search_params = DocumentSearchSchema.model_validate(request.GET.dict())
except ValidationError as e:
log.info(f"Invalid search request: {e.errors()}")
return render(
request,
"caselaw/errors/422.html",
{"errors": e.errors(), "search_params": request.GET},
status=422, # unprocessable entity error
)
try:
# Call the service layer.
search_results = search_service.find_documents(
user=request.user, # Important! The decorator attaches the user to the request.
query=search_params.terms,
page=search_params.page,
)
return render(
request,
"caselaw/search_results.html",
{"results": search_results, "search_params": search_params},
)
except search_exceptions.NoResultsFound:
log.info(f"No results found for query: '{search_params.terms}'")
return render(
request,
"caselaw/search_results.html",
{"results": None, "search_params": search_params},
)
except search_exceptions.SearchServiceError:
log.exception(
f"Search service failed for query: '{search_params.terms}'. User: {request.user.id}"
)
return render(request, "caselaw/errors/500.html", status=500)
@require_http_methods(["POST"])
@session_handler(guest_ok=False) # Requires a logged-in user (always provide the argument)
def add_item_to_list(request: HttpRequest) -> Union[JsonResponse, HttpResponse]:
"""
Adds a document to a user's list from a JSON request body.
For authenticated users only. Validates a JSON request body, calls the
DatabaseService, and returns either a JSON response or an HTML response
based on the 'Accept' header for content negotiation.
Args:
request: The Django HttpRequest object. The body should contain JSON.
Returns:
A JsonResponse with the updated list or an HttpResponse for redirects/messages.
"""
if request.content_type != "application/json":
return JsonResponse(
{"error": "Request content type must be application/json"}, status=415
)
# validate parameters
try:
request_data = json.loads(request.body)
item_data = ListItemAddSchema.model_validate(request_data)
except (json.JSONDecodeError, ValidationError) as e:
errors = e.errors() if isinstance(e, ValidationError) else str(e)
log.info(f"Invalid list item data from user {request.user.id}: {errors}")
return JsonResponse({"errors": errors}, status=422) # invalid entity
# Make database change.
try:
updated_list = db_service.add_document_to_list(
user=request.user,
doc_id=item_data.doc_id,
list_id=item_data.list_id,
)
# Content negotiation based on Accept header.
accept_header = request.headers.get("accept", "")
if "application/json" in accept_header:
return JsonResponse(updated_list.model_dump(), status=200)
else:
# from django.shortcuts import redirect
# return redirect('caselaw:show_list', list_id=updated_list.id)
return HttpResponse("Item added successfully.", status=200)
except db_exceptions.PermissionDeniedError:
log.warning(
f"Permission denied: User {request.user.id} tried to add doc {item_data.doc_id}."
)
return JsonResponse({"error": "Permission denied to access this document."}, status=403) # 403
except db_exceptions.DocumentNotFoundError:
log.info(f"User {request.user.id} tried to add non-existent doc {item_data.doc_id}.")
return JsonResponse({"error": f"Document with ID {item_data.doc_id} not found."}, status=404) # 404
except db_exceptions.DatabaseServiceError:
log.exception(
f"Database service failed for user {request.user.id} adding doc {item_data.doc_id}."
)
return JsonResponse({"error": "An internal server error occurred."}, status=500) # 500
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment