* Home :PROPERTIES: :ONE: one-custom-default-home :CUSTOM_ID: / :TITLE: :DESCRIPTION: Build real-time, reactive interfaces with Django using WebSockets: write Python, not JavaScript. :NAVIGATOR-ACTIVE: home :END: ** What is HTML over the Wire? HTML over the Wire, or HTML over WebSockets, is a strategy for creating real-time SPAs by establishing a WebSocket connection between a client and server. It allows JavaScript to request actions (its only responsibility is to handle events) while the backend handles the business logic and renders HTML. This means you can create dynamic pages without reloading, without AJAX or APIs. This technology provides a secure, stable, and low-latency connection for real-time web applications. #+ATTR_HTML: :class center-block image image--home [[#/img/step-1.avif][Architecture send]] #+ATTR_HTML: :class center-block image image--home [[#/img/step-2.avif][Architecture receive]] ** What is Django LiveView? 🚀 Django LiveView is a framework for creating real-time, interactive web applications entirely in Python 🐍, inspired by [[https://hexdocs.pm/phoenix_live_view/][Phoenix LiveView]] and [[https://laravel-livewire.com/][Laravel Livewire]]. It is built on top of Django Channels. Build rich, dynamic user experiences ✨ with server-rendered HTML without writing a single line of JavaScript. Perfect for Django developers who want real-time features ⚡ without the complexity of a separate frontend framework. Let's illustrate with an example. I want to print article number 2. 1. A WebSocket connection (a channel) is established between the client and the server. 2. JavaScript sends a message via WebSocket to the server (Django). #+ATTR_HTML: :class center-block image image--home [[#/img/step-3.avif][Send string]] 3. Django interprets the message and renders the HTML of the article through the template system and the database. 4. Django sends the HTML to JavaScript via the channel and specifies which selector to embed it in. #+ATTR_HTML: :class center-block image image--home [[#/img/step-4.avif][Send JSON]] 5. JavaScript renders the received HTML in the indicated selector. #+ATTR_HTML: :class center-block image image--home [[#/img/step-5.avif][Place HTML]] The same process is repeated for each action, such as clicking a button, submitting a form, etc. ** What are your superpowers? 💪 - 🎯 **Create SPAs without using APIs**: No REST or GraphQL needed - 🎨 **Uses Django's template system** to render the frontend (without JavaScript frameworks) - 🐍 **Logic stays in Python**: No split between backend and frontend - 🛠️ **Use all of Django's tools**: ORM, forms, authentication, admin, etc. - ⚡ **Everything is asynchronous by default**: Built on Django Channels - 📚 **Zero learning curve**: If you know Python and Django, you're ready - 🔄 **Real-time by design**: All interactions happen over WebSockets - 🔋 **Batteries included**: JavaScript assets bundled, automatic reconnection with exponential backoff - 💡 **Type hints** and modern Python (3.10+) - 📡 **Broadcast support** for multi-user real-time updates - 🔐 **Middleware system** for authentication and authorization Are you ready to create your first real-time SPA? Let's go to the [[#/quick-start/][Quick start]]. * Install :PROPERTIES: :ONE: one-custom-default-doc :CUSTOM_ID: /docs/install/ :TITLE: Install :DESCRIPTION: Install Django LiveView. :NAVIGATOR-ACTIVE: docs :END: ** Requirements - Python 3.10+ - Django 4.2+ - Redis (for Channels layer) - Channels 4.0+ ** Installation Install Django LiveView with pip: #+BEGIN_SRC sh pip install django-liveview #+END_SRC ** Configure Django Add to your ~settings.py~: #+BEGIN_SRC python # settings.py INSTALLED_APPS = [ "daphne", # Must be first for ASGI support "channels", "liveview", # ... your other apps ] # ASGI configuration ASGI_APPLICATION = "your_project.asgi.application" # Configure Channels with Redis CHANNEL_LAYERS = { "default": { "BACKEND": "channels_redis.core.RedisChannelLayer", "CONFIG": { "hosts": [("127.0.0.1", 6379)], }, }, } #+END_SRC ** Setup ASGI routing Create or update ~asgi.py~: #+BEGIN_SRC python # asgi.py import os from django.core.asgi import get_asgi_application from channels.routing import ProtocolTypeRouter, URLRouter from channels.auth import AuthMiddlewareStack from channels.security.websocket import AllowedHostsOriginValidator from liveview.routing import get_liveview_urlpatterns os.environ.setdefault("DJANGO_SETTINGS_MODULE", "your_project.settings") application = ProtocolTypeRouter({ "http": get_asgi_application(), "websocket": AllowedHostsOriginValidator( AuthMiddlewareStack( URLRouter( get_liveview_urlpatterns() ) ) ), }) #+END_SRC ** Add JavaScript to your base template #+BEGIN_SRC html {% load static %} {% load liveview %} {% block title %}My Site{% endblock %} {% block content %}{% endblock %} #+END_SRC **Important attributes:** - ~data-room="{% liveview_room_uuid %}"~ on ~~: unique room ID for each page load - ~data-controller="page"~ on ~~: activates the Stimulus controller The ~{% liveview_room_uuid %}~ template tag generates a random UUID for each request, ensuring isolated sessions and preventing room ID enumeration attacks. **Alternative: User-specific Room ID** (use with caution) #+BEGIN_SRC html #+END_SRC ⚠️ *Warning:* Using predictable IDs like user IDs can be a security risk. An attacker could subscribe to another user's room by guessing their ID. Only use this if you implement proper authorization checks in your handlers. We strongly recommend that you follow the [[#/quick-start/][Quick start]] to see the installation in action. * Handlers :PROPERTIES: :ONE: one-custom-default-doc :CUSTOM_ID: /docs/handlers/ :TITLE: Handlers :DESCRIPTION: LiveView handlers of Django LiveView. :NAVIGATOR-ACTIVE: docs :END: Handlers are Python functions that respond to WebSocket messages from the client. They contain your business logic and can render HTML, update the database, broadcast to multiple users, and more. ** Your First Handler Let's start with the simplest possible handler. This handler will be called when a button is clicked and will update a section of the page. Python code: #+BEGIN_SRC python from liveview import liveview_handler, send @liveview_handler("say_hello") def say_hello(consumer, content): send(consumer, { "target": "#greeting", "html": "

Hello, World!

" }) #+END_SRC HTML code: #+BEGIN_SRC html
#+END_SRC When you click the button: 1. The frontend sends a WebSocket message with ~function: "say_hello"~ 2. Django LiveView calls the ~say_hello~ handler 3. The handler sends back HTML to replace the content of ~#greeting~ 4. The page updates instantly without a full reload ** Working with Form Data Most handlers need to receive data from the user. All form inputs within the same container are automatically sent to your handler. Python code: #+BEGIN_SRC python @liveview_handler("greet_user") def greet_user(consumer, content): # Get the name from form data name = content["form"].get("name", "Anonymous") send(consumer, { "target": "#greeting", "html": f"

Hello, {name}! Welcome to Django LiveView.

" }) #+END_SRC HTML code: #+BEGIN_SRC html
#+END_SRC The ~content["form"]~ dictionary contains all input values with their ~name~ attribute as the key. ** Using Templates for Complex HTML For anything beyond simple strings, use Django templates to render your HTML. Python code: #+BEGIN_SRC python from django.template.loader import render_to_string @liveview_handler("show_profile") def show_profile(consumer, content): user_id = content["form"].get("user_id") # Get user from database from .models import User user = User.objects.get(id=user_id) # Render template with context html = render_to_string("profile_card.html", { "user": user, "is_online": True }) send(consumer, { "target": "#profile-container", "html": html }) #+END_SRC HTML code: #+BEGIN_SRC html
#+END_SRC Template (~profile_card.html~): #+BEGIN_SRC html
{{ user.name }}

{{ user.name }}

{{ user.bio }}

{% if is_online %} Online {% endif %}
#+END_SRC ** Understanding the Content Parameter Every handler receives two parameters: ~consumer~ and ~content~. The ~content~ dictionary contains all the information from the client. #+BEGIN_SRC python @liveview_handler("example") def example(consumer, content): # content structure: # { # "function": "example", # Handler name # "form": {...}, # All form inputs # "data": {...}, # Custom data-data-* attributes # "lang": "en", # Current language # "room": "user_123" # WebSocket room identifier # } pass #+END_SRC ** Auto-discovery Django LiveView automatically discovers handlers in ~liveview_components/~ directories within your installed apps: #+BEGIN_SRC my_app/ ├── liveview_components/ │ ├── __init__.py │ ├── users.py │ ├── posts.py │ └── comments.py #+END_SRC Handlers are loaded on startup with this output: #+BEGIN_SRC ✓ Imported: my_app.liveview_components.users ✓ Imported: my_app.liveview_components.posts ✓ Imported: my_app.liveview_components.comments #+END_SRC ** Using Custom Data Attributes Sometimes you need to send additional data that isn't part of a form. Use ~data-data-*~ attributes for this. Python code: #+BEGIN_SRC python @liveview_handler("delete_comment") def delete_comment(consumer, content): # Access custom data from data-data-* attributes comment_id = content["data"]["comment_id"] post_id = content["data"]["post_id"] # Delete from database from .models import Comment Comment.objects.filter(id=comment_id).delete() send(consumer, { "target": f"#comment-{comment_id}", "remove": True # Remove the element from DOM }) #+END_SRC HTML code: #+BEGIN_SRC html

This is a comment

#+END_SRC The attribute ~data-data-comment-id~ becomes ~comment_id~ (with underscores, not camelCase) within ~content["data"]~. **Important**: Django LiveView automatically converts all attribute names from camelCase to snake_case. HTML attributes with hyphens (~data-data-comment-id~) are first converted by the browser to camelCase (~dataCommentId~), which Django LiveView then converts to snake_case (~comment_id~) in your handlers. ** Appending Content to Lists When building dynamic lists (like infinite scroll or chat messages), you want to add items without replacing the entire list. Python code: #+BEGIN_SRC python @liveview_handler("load_more_posts") def load_more_posts(consumer, content): page = int(content["form"].get("page", 1)) # Get next page of posts from .models import Post posts = Post.objects.all()[(page-1)*10:page*10] # Render new posts html = render_to_string("posts_list.html", { "posts": posts }) send(consumer, { "target": "#posts-container", "html": html, "append": True # Add to the end instead of replacing }) #+END_SRC HTML code: #+BEGIN_SRC html
#+END_SRC Template (~posts_list.html~): #+BEGIN_SRC html {% for post in posts %}

{{ post.title }}

{{ post.content }}

{% endfor %} #+END_SRC ** Removing Elements from the DOM Instead of hiding elements with CSS, you can completely remove them from the page. Python code: #+BEGIN_SRC python @liveview_handler("archive_notification") def archive_notification(consumer, content): notification_id = content["data"]["notification_id"] # Archive in database from .models import Notification Notification.objects.filter(id=notification_id).update(archived=True) # Remove from page send(consumer, { "target": f"#notification-{notification_id}", "remove": True }) #+END_SRC HTML code: #+BEGIN_SRC html

You have a new message

#+END_SRC ** Updating the URL Without Page Reload Create SPA-like navigation by updating both content and the browser URL. Python code: #+BEGIN_SRC python @liveview_handler("navigate_to_profile") def navigate_to_profile(consumer, content): user_id = content["data"]["user_id"] # Get user data from .models import User user = User.objects.get(id=user_id) # Render profile page html = render_to_string("profile_page.html", { "user": user }) send(consumer, { "target": "#main-content", "html": html, "url": f"/profile/{user.username}/", # Update browser URL "title": f"{user.name} - Profile" # Update page title }) #+END_SRC HTML code: #+BEGIN_SRC html

Home Page

#+END_SRC The browser's back/forward buttons will work correctly, and users can bookmark or share the URL. ** Scrolling to Elements After updating content, you often want to scroll to a specific element or to the top of the page. Python code: #+BEGIN_SRC python @liveview_handler("show_product_details") def show_product_details(consumer, content): product_id = content["data"]["product_id"] from .models import Product product = Product.objects.get(id=product_id) html = render_to_string("product_details.html", { "product": product }) send(consumer, { "target": "#product-details", "html": html, "scroll": "#product-details" # Smooth scroll to this element }) @liveview_handler("search_products") def search_products(consumer, content): query = content["form"].get("q") from .models import Product products = Product.objects.filter(name__icontains=query) html = render_to_string("products_list.html", { "products": products }) send(consumer, { "target": "#products-list", "html": html, "scrollTop": True # Scroll to top of page }) #+END_SRC HTML code: #+BEGIN_SRC html
#+END_SRC * Frontend Integration :PROPERTIES: :ONE: one-custom-default-doc :CUSTOM_ID: /docs/frontend/ :TITLE: Frontend Integration :DESCRIPTION: How to call handlers from the frontend. :NAVIGATOR-ACTIVE: docs :END: The frontend is responsible for capturing events and sending messages over WebSocket. No logic, rendering, or state is in the frontend: the backend does all the work. Django LiveView uses [[https://stimulus.hotwired.dev/][Stimulus]] to manage DOM events and avoid collisions. The JavaScript assets are bundled within the package. ** Calling Handlers To call a handler from a button click: #+BEGIN_SRC html #+END_SRC - ~data-liveview-function~: The handler name registered with ~@liveview_handler~ - ~data-action~: Stimulus action (event->controller#action) ** Sending Form Data All form fields are automatically extracted and sent in ~content["form"]~: #+BEGIN_SRC html
#+END_SRC ** Sending Custom Data Use ~data-data-*~ attributes to send additional data: #+BEGIN_SRC html #+END_SRC Access in Python: #+BEGIN_SRC python @liveview_handler("open_modal") def open_modal(consumer, content): data = content.get("data", {}) modal_id = data.get("modal_id") # from data-data-modal-id user_id = data.get("user_id") # from data-data-user-id #+END_SRC ** Stimulus Actions Reference Available Stimulus actions: - ~data-action="click->page#run"~: Execute LiveView function on click - ~data-action="input->page#run"~: Execute on input change (real-time) - ~data-action="submit->page#run"~: Execute on form submit - ~data-action="change->page#run"~: Execute on change event - ~data-action="blur->page#run"~: Execute when element loses focus - ~data-action="page#stop"~: Stop event propagation * Forms :PROPERTIES: :ONE: one-custom-default-doc :CUSTOM_ID: /docs/forms/ :TITLE: Forms :DESCRIPTION: Form handling in Django LiveView. :NAVIGATOR-ACTIVE: docs :END: Django LiveView works seamlessly with Django forms. Form data is sent via WebSocket instead of HTTP. ** Form Handling Example Python handler: #+BEGIN_SRC python @liveview_handler("submit_contact") def submit_contact(consumer, content): from .forms import ContactForm form = ContactForm(content["form"]) if form.is_valid(): # Save to database contact = form.save() # Show success message html = render_to_string("contact_success.html", { "message": "Thank you! We'll be in touch." }) else: # Show form with errors html = render_to_string("contact_form.html", { "form": form }) send(consumer, { "target": "#contact-container", "html": html }) #+END_SRC HTML template: #+BEGIN_SRC html
#+END_SRC ** Real-time Validation #+BEGIN_SRC python @liveview_handler("validate_field") def validate_field(consumer, content): field_name = content["data"]["field"] field_value = content["form"].get(field_name, "") # Validate error = None if field_name == "email" and "@" not in field_value: error = "Invalid email address" elif field_name == "name" and len(field_value) < 3: error = "Name must be at least 3 characters" # Show error or success html = f'{error or "✓"}' send(consumer, { "target": f"#error-{field_name}", "html": html }) #+END_SRC #+BEGIN_SRC html #+END_SRC * Broadcasting :PROPERTIES: :ONE: one-custom-default-doc :CUSTOM_ID: /docs/broadcasting/ :TITLE: Broadcasting :DESCRIPTION: Broadcasting updates to multiple users. :NAVIGATOR-ACTIVE: docs :END: Send updates to all connected clients using the ~broadcast~ parameter: ** Simple Broadcast #+BEGIN_SRC python @liveview_handler("notify_all") def notify_all(consumer, content): message = content["form"]["message"] html = render_to_string("notification.html", { "message": message }) send(consumer, { "target": "#notifications", "html": html, "append": True }, broadcast=True) # Sends to ALL connected users #+END_SRC ** Background Thread Broadcast with Auto-removal #+BEGIN_SRC python from threading import Thread from time import sleep from uuid import uuid4 @liveview_handler("send_notification") def send_notification(consumer, content): notification_id = str(uuid4().hex) message = "New update available!" def broadcast_notification(): # Send notification html = render_to_string("notification.html", { "id": notification_id, "message": message }) send(consumer, { "target": "#notifications", "html": html, "append": True }, broadcast=True) # Remove after 5 seconds sleep(5) send(consumer, { "target": f"#notification-{notification_id}", "remove": True }, broadcast=True) Thread(target=broadcast_notification).start() #+END_SRC * Advanced Features :PROPERTIES: :ONE: one-custom-default-doc :CUSTOM_ID: /docs/advanced/ :TITLE: Advanced Features :DESCRIPTION: Advanced features of Django LiveView. :NAVIGATOR-ACTIVE: docs :END: ** Intersection Observer (Infinite Scroll) Trigger functions when elements enter or exit the viewport: #+BEGIN_SRC python ITEMS_PER_PAGE = 10 @liveview_handler("load_more") def load_more(consumer, content): page = int(content["data"].get("page", 1)) # Fetch items start = (page - 1) * ITEMS_PER_PAGE end = start + ITEMS_PER_PAGE items = Item.objects.all()[start:end] is_last_page = end >= Item.objects.count() # Append items to list send(consumer, { "target": "#items-list", "html": render_to_string("items_partial.html", { "items": items }), "append": True }) # Update or remove intersection observer trigger if is_last_page: html = "" else: html = render_to_string("load_trigger.html", { "next_page": page + 1 }) send(consumer, { "target": "#load-more-trigger", "html": html }) #+END_SRC HTML template: #+BEGIN_SRC html

Loading more...

#+END_SRC **Attributes:** - ~data-liveview-intersect-appear="function_name"~: Call when element appears - ~data-liveview-intersect-disappear="function_name"~: Call when element disappears - ~data-liveview-intersect-threshold="200"~: Trigger 200px before entering viewport (default: 0) ** Auto-focus Automatically focus elements after rendering: #+BEGIN_SRC html #+END_SRC ** Init Functions Execute functions when elements are first rendered: #+BEGIN_SRC html
#+END_SRC ** Debounce Reduce server calls by adding a delay before sending requests. Perfect for search inputs and real-time validation: #+BEGIN_SRC html #+END_SRC The ~data-liveview-debounce="500"~ attribute waits 500ms after the user stops typing before sending the request. This dramatically reduces server load and provides a better user experience. **Example: Real-time search with debounce** #+BEGIN_SRC python from liveview import liveview_handler, send from django.template.loader import render_to_string @liveview_handler("search_articles") def search_articles(consumer, content): query = content["form"]["search"] articles = Article.objects.filter(title__icontains=query) html = render_to_string("search_results.html", { "articles": articles }) send(consumer, { "target": "#search-results", "html": html }) #+END_SRC Without debounce, typing "python" would send 6 requests (one per letter). With ~data-liveview-debounce="500"~, it sends only 1 request after the user stops typing for 500ms. ** Middleware System Add middleware to run before handlers for authentication, logging, or rate limiting: #+BEGIN_SRC python from liveview import liveview_registry, send def auth_middleware(consumer, content, function_name): """Check if user is authenticated before running handler""" user = consumer.scope.get("user") if not user or not user.is_authenticated: send(consumer, { "target": "#error", "html": "

You must be logged in

" }) return False # Cancel handler execution return True # Continue to handler def logging_middleware(consumer, content, function_name): """Log all handler calls""" import logging logger = logging.getLogger(__name__) user = consumer.scope.get("user") logger.info(f"Handler '{function_name}' called by {user}") return True # Continue to handler # Register middleware liveview_registry.add_middleware(auth_middleware) liveview_registry.add_middleware(logging_middleware) #+END_SRC * Error Handling :PROPERTIES: :ONE: one-custom-default-doc :CUSTOM_ID: /docs/error-handling/ :TITLE: Error Handling :DESCRIPTION: Handle errors and exceptions in Django LiveView. :NAVIGATOR-ACTIVE: docs :END: Proper error handling is essential for production applications. Django LiveView provides several patterns for handling errors gracefully. ** Basic Error Handling Wrap your handler logic in try-except blocks to catch and handle errors: #+BEGIN_SRC python from liveview import liveview_handler, send from django.template.loader import render_to_string @liveview_handler("load_article") def load_article(consumer, content): article_id = content.get("data", {}).get("article_id") try: from .models import Article article = Article.objects.get(id=article_id) html = render_to_string("article_detail.html", { "article": article }) send(consumer, { "target": "#article-container", "html": html }) except Article.DoesNotExist: # Show error message to user send(consumer, { "target": "#article-container", "html": "

Article not found

" }) except Exception as e: # Log unexpected errors import logging logger = logging.getLogger(__name__) logger.error(f"Error loading article {article_id}: {e}") # Show generic error to user send(consumer, { "target": "#article-container", "html": "

An error occurred. Please try again.

" }) #+END_SRC ** Validation Errors Handle validation errors from Django forms: #+BEGIN_SRC python @liveview_handler("submit_form") def submit_form(consumer, content): from .forms import ArticleForm form = ArticleForm(content.get("form", {})) if form.is_valid(): article = form.save() # Show success message html = render_to_string("article_success.html", { "article": article }) send(consumer, { "target": "#form-container", "html": html }) else: # Show form with errors html = render_to_string("article_form.html", { "form": form, "errors": form.errors }) send(consumer, { "target": "#form-container", "html": html }) #+END_SRC Template (~article_form.html~): #+BEGIN_SRC html {% if errors %}
{% for field, error_list in errors.items %}

{{ field }}: {{ error_list.0 }}

{% endfor %}
{% endif %}
{{ form.as_p }}
#+END_SRC ** Permission and Authentication Errors Check permissions before executing operations: #+BEGIN_SRC python @liveview_handler("delete_article") def delete_article(consumer, content): user = consumer.scope.get("user") article_id = content.get("data", {}).get("article_id") # Check authentication if not user or not user.is_authenticated: send(consumer, { "target": "#error-message", "html": "

You must be logged in to delete articles

" }) return try: from .models import Article article = Article.objects.get(id=article_id) # Check permissions if article.author != user and not user.is_staff: send(consumer, { "target": "#error-message", "html": "

You don't have permission to delete this article

" }) return article.delete() # Show success and remove from DOM send(consumer, { "target": f"#article-{article_id}", "remove": True }) send(consumer, { "target": "#success-message", "html": "

Article deleted successfully

" }) except Article.DoesNotExist: send(consumer, { "target": "#error-message", "html": "

Article not found

" }) #+END_SRC ** Database Transaction Errors Use database transactions to ensure data integrity: #+BEGIN_SRC python from django.db import transaction @liveview_handler("create_post_with_tags") def create_post_with_tags(consumer, content): from .models import Post, Tag try: with transaction.atomic(): # Create post post = Post.objects.create( title=content["form"]["title"], content=content["form"]["content"] ) # Add tags tag_names = content["form"]["tags"].split(",") for tag_name in tag_names: tag, _ = Tag.objects.get_or_create(name=tag_name.strip()) post.tags.add(tag) # Success html = render_to_string("post_created.html", { "post": post }) send(consumer, { "target": "#post-container", "html": html }) except Exception as e: # Transaction is automatically rolled back import logging logger = logging.getLogger(__name__) logger.error(f"Error creating post: {e}") send(consumer, { "target": "#error-message", "html": "

Failed to create post. Please try again.

" }) #+END_SRC ** Timeout Errors Handle long-running operations with timeouts: #+BEGIN_SRC python import asyncio from threading import Thread @liveview_handler("process_large_file") def process_large_file(consumer, content): file_id = content.get("data", {}).get("file_id") def process_with_timeout(): try: from .models import File file_obj = File.objects.get(id=file_id) # Simulate long operation result = file_obj.process() # This might take a while # Send success html = render_to_string("process_success.html", { "result": result }) send(consumer, { "target": "#result-container", "html": html }, broadcast=True) except Exception as e: send(consumer, { "target": "#result-container", "html": f"

Processing failed: {str(e)}

" }) # Start processing in background Thread(target=process_with_timeout).start() # Show immediate feedback send(consumer, { "target": "#result-container", "html": "

Processing file... This may take a moment.

" }) #+END_SRC ** Global Error Handler with Middleware Create a middleware to catch all errors: #+BEGIN_SRC python from liveview import liveview_registry, send import logging logger = logging.getLogger(__name__) def error_handling_middleware(consumer, content, function_name): """Catch all exceptions and show user-friendly error""" try: # Continue to handler return True except Exception as e: # Log the error logger.error(f"Error in handler '{function_name}': {e}", exc_info=True) # Show error to user send(consumer, { "target": "#global-error", "html": "

Something went wrong. Our team has been notified.

" }) # Don't continue to handler return False # Register middleware liveview_registry.add_middleware(error_handling_middleware) #+END_SRC Note: Middleware runs ~before~ the handler. To catch errors ~during~ handler execution, wrap your handler logic in try-except blocks. ** Logging Best Practices #+BEGIN_SRC python import logging logger = logging.getLogger(__name__) @liveview_handler("important_operation") def important_operation(consumer, content): user = consumer.scope.get("user") # Log the operation attempt logger.info(f"User {user.id if user else 'anonymous'} attempting operation") try: # Perform operation result = perform_operation() # Log success logger.info(f"Operation successful for user {user.id if user else 'anonymous'}") send(consumer, { "target": "#result", "html": f"

Success: {result}

" }) except ValueError as e: # Log validation errors as warnings logger.warning(f"Validation error for user {user.id if user else 'anonymous'}: {e}") send(consumer, { "target": "#result", "html": f"

Invalid input: {e}

" }) except Exception as e: # Log unexpected errors as errors logger.error(f"Unexpected error for user {user.id if user else 'anonymous'}: {e}", exc_info=True) send(consumer, { "target": "#result", "html": "

An unexpected error occurred

" }) #+END_SRC * Testing :PROPERTIES: :ONE: one-custom-default-doc :CUSTOM_ID: /docs/testing/ :TITLE: Testing :DESCRIPTION: Testing Django LiveView handlers and applications. :NAVIGATOR-ACTIVE: docs :END: Testing LiveView handlers ensures your real-time features work correctly. Django LiveView handlers are regular Python functions, so you can test them like any other Django code. ** Unit Testing Handlers Test handlers directly by calling them with mock data: #+BEGIN_SRC python from django.test import TestCase from unittest.mock import Mock, patch from myapp.liveview_components.articles import load_article class ArticleHandlerTest(TestCase): def setUp(self): from myapp.models import Article self.article = Article.objects.create( title="Test Article", content="Test content" ) def test_load_article_success(self): # Create mock consumer consumer = Mock() # Create content dict content = { "data": {"article_id": str(self.article.id)}, "form": {}, "function": "load_article" } # Patch the send function to capture calls with patch("myapp.liveview_components.articles.send") as mock_send: # Call handler load_article(consumer, content) # Verify send was called mock_send.assert_called_once() # Verify the call arguments call_args = mock_send.call_args[0] self.assertEqual(call_args[0], consumer) # Verify HTML contains article title html = call_args[1]["html"] self.assertIn("Test Article", html) self.assertIn("Test content", html) def test_load_article_not_found(self): consumer = Mock() content = { "data": {"article_id": "99999"}, "form": {} } with patch("myapp.liveview_components.articles.send") as mock_send: load_article(consumer, content) # Verify error message is sent html = mock_send.call_args[0][1]["html"] self.assertIn("not found", html.lower()) #+END_SRC ** Integration Testing with WebSocket Client Test the full WebSocket flow using Django Channels testing utilities: #+BEGIN_SRC python from channels.testing import WebsocketCommunicator from django.test import TransactionTestCase from myproject.asgi import application import json class LiveViewIntegrationTest(TransactionTestCase): async def test_websocket_connection(self): # Create WebSocket communicator communicator = WebsocketCommunicator( application, "/ws/liveview/test-room/" ) # Connect connected, _ = await communicator.connect() self.assertTrue(connected) # Send message to handler await communicator.send_json_to({ "function": "say_hello", "form": {"name": "Django"}, "data": {}, "lang": "en", "room": "test-room" }) # Receive response response = await communicator.receive_json_from() # Verify response self.assertEqual(response["target"], "#greeting") self.assertIn("Hello, Django", response["html"]) # Disconnect await communicator.disconnect() async def test_form_validation(self): from myapp.models import Article communicator = WebsocketCommunicator( application, "/ws/liveview/test-room/" ) connected, _ = await communicator.connect() self.assertTrue(connected) # Send invalid form data await communicator.send_json_to({ "function": "submit_article", "form": {"title": ""}, # Empty title should fail "data": {}, "lang": "en", "room": "test-room" }) response = await communicator.receive_json_from() # Verify error is shown self.assertIn("error", response["html"].lower()) # Verify article was not created self.assertEqual(Article.objects.count(), 0) await communicator.disconnect() #+END_SRC ** Testing Broadcasting Test that handlers broadcast to all connected users: #+BEGIN_SRC python from channels.testing import WebsocketCommunicator from django.test import TransactionTestCase from myproject.asgi import application class BroadcastTest(TransactionTestCase): async def test_broadcast_to_all_users(self): # Connect multiple users user1 = WebsocketCommunicator( application, "/ws/liveview/room-1/" ) user2 = WebsocketCommunicator( application, "/ws/liveview/room-1/" ) await user1.connect() await user2.connect() # User 1 sends a broadcast message await user1.send_json_to({ "function": "notify_all", "form": {"message": "Hello everyone!"}, "data": {}, "lang": "en", "room": "room-1" }) # Both users should receive the message response1 = await user1.receive_json_from() response2 = await user2.receive_json_from() self.assertIn("Hello everyone!", response1["html"]) self.assertIn("Hello everyone!", response2["html"]) await user1.disconnect() await user2.disconnect() #+END_SRC ** Testing with Authenticated Users Test handlers that require authentication: #+BEGIN_SRC python from channels.testing import WebsocketCommunicator from django.contrib.auth import get_user_model from django.test import TransactionTestCase from channels.db import database_sync_to_async User = get_user_model() class AuthenticatedHandlerTest(TransactionTestCase): async def test_authenticated_handler(self): # Create user user = await database_sync_to_async(User.objects.create_user)( username="testuser", password="testpass123" ) # Create authenticated communicator communicator = WebsocketCommunicator( application, "/ws/liveview/user-room/", ) communicator.scope["user"] = user connected, _ = await communicator.connect() self.assertTrue(connected) # Send request that requires authentication await communicator.send_json_to({ "function": "delete_article", "data": {"article_id": "123"}, "form": {}, "lang": "en", "room": "user-room" }) response = await communicator.receive_json_from() # Should succeed (not show "must be logged in" error) self.assertNotIn("must be logged in", response["html"].lower()) await communicator.disconnect() #+END_SRC ** Testing Middleware Test that middleware correctly filters requests: #+BEGIN_SRC python from django.test import TestCase from unittest.mock import Mock, patch from myapp.middleware import auth_middleware class MiddlewareTest(TestCase): def test_auth_middleware_blocks_anonymous(self): # Create mock consumer with anonymous user consumer = Mock() consumer.scope = {"user": None} content = {"function": "protected_handler"} with patch("myapp.middleware.send") as mock_send: # Call middleware result = auth_middleware(consumer, content, "protected_handler") # Should block the request self.assertFalse(result) # Should send error message mock_send.assert_called_once() self.assertIn("logged in", mock_send.call_args[0][1]["html"]) def test_auth_middleware_allows_authenticated(self): from django.contrib.auth import get_user_model User = get_user_model() # Create mock consumer with authenticated user user = User(username="testuser", is_authenticated=True) consumer = Mock() consumer.scope = {"user": user} content = {"function": "protected_handler"} # Call middleware result = auth_middleware(consumer, content, "protected_handler") # Should allow the request self.assertTrue(result) #+END_SRC ** End-to-End Testing with Selenium Test the full user experience including JavaScript: #+BEGIN_SRC python from django.contrib.staticfiles.testing import StaticLiveServerTestCase from selenium import webdriver from selenium.webdriver.common.by import By from selenium.webdriver.support.ui import WebDriverWait from selenium.webdriver.support import expected_conditions as EC class E2ETest(StaticLiveServerTestCase): @classmethod def setUpClass(cls): super().setUpClass() cls.driver = webdriver.Chrome() cls.driver.implicitly_wait(10) @classmethod def tearDownClass(cls): cls.driver.quit() super().tearDownClass() def test_real_time_update(self): # Open the page self.driver.get(f"{self.live_server_url}/") # Type in input field input_field = self.driver.find_element(By.NAME, "name") input_field.send_keys("Django") # Click button button = self.driver.find_element( By.CSS_SELECTOR, "[data-liveview-function='say_hello']" ) button.click() # Wait for update greeting = WebDriverWait(self.driver, 5).until( EC.presence_of_element_located((By.ID, "greeting")) ) # Verify content updated self.assertIn("Hello, Django", greeting.text) #+END_SRC ** Testing Best Practices 1. **Test handler logic separately** from WebSocket connection 2. **Mock the ~send~ function** to verify what gets sent to clients 3. **Test error cases** to ensure proper error handling 4. **Test permissions** to prevent unauthorized access 5. **Use ~TransactionTestCase~** for WebSocket tests (not ~TestCase~) 6. **Test broadcasts** to ensure multi-user features work correctly 7. **Write E2E tests** for critical user flows * Deployment :PROPERTIES: :ONE: one-custom-default-doc :CUSTOM_ID: /docs/deployment/ :TITLE: Deployment :DESCRIPTION: Deploy Django LiveView applications to production. :NAVIGATOR-ACTIVE: docs :END: Deploying Django LiveView requires an ASGI server (like Daphne or Uvicorn), Redis for the channel layer, and proper WebSocket support. ** Production Requirements - **ASGI Server**: Daphne, Uvicorn, or Hypercorn - **Redis**: For channel layer (shared state between workers) - **WebSocket Support**: Reverse proxy must support WebSockets (Nginx, Caddy, Traefik) - **Process Manager**: Supervisor, systemd, or Docker - **SSL/TLS**: Required for ~wss://~ connections in production ** Using Daphne (Recommended) Install Daphne: #+BEGIN_SRC sh pip install daphne #+END_SRC Run Daphne in production: #+BEGIN_SRC sh daphne -b 0.0.0.0 -p 8000 myproject.asgi:application #+END_SRC With multiple workers for better performance: #+BEGIN_SRC sh daphne -b 0.0.0.0 -p 8000 --workers 4 myproject.asgi:application #+END_SRC ** Using Uvicorn Install Uvicorn: #+BEGIN_SRC sh pip install uvicorn[standard] #+END_SRC Run Uvicorn: #+BEGIN_SRC sh uvicorn myproject.asgi:application --host 0.0.0.0 --port 8000 --workers 4 #+END_SRC ** Redis Configuration for Production Update ~settings.py~ for production Redis: #+BEGIN_SRC python # settings.py CHANNEL_LAYERS = { "default": { "BACKEND": "channels_redis.core.RedisChannelLayer", "CONFIG": { "hosts": [ { "address": ("redis", 6379), # Redis hostname "password": "your-redis-password", "db": 0, } ], "capacity": 1500, # Max messages per channel "expiry": 10, # Message expiry in seconds }, }, } #+END_SRC For Redis Sentinel (high availability): #+BEGIN_SRC python CHANNEL_LAYERS = { "default": { "BACKEND": "channels_redis.core.RedisChannelLayer", "CONFIG": { "hosts": [ { "sentinels": [ ("sentinel1", 26379), ("sentinel2", 26379), ("sentinel3", 26379), ], "master_name": "mymaster", "password": "your-redis-password", } ], }, }, } #+END_SRC ** Nginx Configuration Configure Nginx to proxy WebSocket connections: #+BEGIN_SRC nginx upstream django { server 127.0.0.1:8000; } server { listen 80; server_name example.com; # Redirect HTTP to HTTPS return 301 https://$server_name$request_uri; } server { listen 443 ssl http2; server_name example.com; ssl_certificate /path/to/cert.pem; ssl_certificate_key /path/to/key.pem; # Static files location /static/ { alias /path/to/static/; expires 30d; add_header Cache-Control "public, immutable"; } location /media/ { alias /path/to/media/; expires 7d; } # WebSocket support location /ws/ { proxy_pass http://django; proxy_http_version 1.1; proxy_set_header Upgrade $http_upgrade; proxy_set_header Connection "upgrade"; proxy_set_header Host $host; proxy_set_header X-Real-IP $remote_addr; proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for; proxy_set_header X-Forwarded-Proto $scheme; # Increase timeout for long-lived connections proxy_read_timeout 86400; proxy_send_timeout 86400; } # Regular HTTP requests location / { proxy_pass http://django; proxy_set_header Host $host; proxy_set_header X-Real-IP $remote_addr; proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for; proxy_set_header X-Forwarded-Proto $scheme; } } #+END_SRC ** Docker Deployment ~Dockerfile~: #+BEGIN_SRC dockerfile FROM python:3.11-slim WORKDIR /app # Install dependencies COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt # Copy project COPY . . # Collect static files RUN python manage.py collectstatic --noinput # Expose port EXPOSE 8000 # Run Daphne CMD ["daphne", "-b", "0.0.0.0", "-p", "8000", "myproject.asgi:application"] #+END_SRC ~docker-compose.yml~: #+BEGIN_SRC yaml version: '3.8' services: redis: image: redis:7-alpine restart: always volumes: - redis_data:/data command: redis-server --requirepass your-redis-password web: build: . restart: always ports: - "8000:8000" depends_on: - redis environment: - REDIS_HOST=redis - REDIS_PASSWORD=your-redis-password - DJANGO_SETTINGS_MODULE=myproject.settings volumes: - static_volume:/app/staticfiles - media_volume:/app/media nginx: image: nginx:alpine restart: always ports: - "80:80" - "443:443" volumes: - ./nginx.conf:/etc/nginx/nginx.conf:ro - static_volume:/static:ro - media_volume:/media:ro - ./ssl:/etc/nginx/ssl:ro depends_on: - web volumes: redis_data: static_volume: media_volume: #+END_SRC ** Systemd Service Create ~/etc/systemd/system/django-liveview.service~: #+BEGIN_SRC ini [Unit] Description=Django LiveView Application After=network.target redis.service [Service] Type=simple User=www-data Group=www-data WorkingDirectory=/path/to/project Environment="PATH=/path/to/venv/bin" ExecStart=/path/to/venv/bin/daphne -b 0.0.0.0 -p 8000 myproject.asgi:application Restart=always RestartSec=10 [Install] WantedBy=multi-user.target #+END_SRC Enable and start: #+BEGIN_SRC sh sudo systemctl enable django-liveview sudo systemctl start django-liveview sudo systemctl status django-liveview #+END_SRC ** Environment Variables Store sensitive configuration in environment variables: #+BEGIN_SRC python # settings.py import os SECRET_KEY = os.environ.get("DJANGO_SECRET_KEY") DEBUG = os.environ.get("DJANGO_DEBUG", "False") == "True" ALLOWED_HOSTS = os.environ.get("DJANGO_ALLOWED_HOSTS", "").split(",") CHANNEL_LAYERS = { "default": { "BACKEND": "channels_redis.core.RedisChannelLayer", "CONFIG": { "hosts": [{ "address": ( os.environ.get("REDIS_HOST", "127.0.0.1"), int(os.environ.get("REDIS_PORT", 6379)) ), "password": os.environ.get("REDIS_PASSWORD"), }], }, }, } #+END_SRC ** Performance Tuning 1. **Use Redis persistence** (~appendonly yes~) for channel layer reliability 2. **Increase worker count** based on CPU cores (~workers = CPU cores * 2 + 1~) 3. **Set proper ~capacity~ and ~expiry~** in channel layer config 4. **Enable HTTP/2** in Nginx for better performance 5. **Use connection pooling** for database 6. **Configure WebSocket timeouts** appropriately 7. **Monitor Redis memory** usage and set ~maxmemory~ limits ** Monitoring Monitor your Django LiveView application: #+BEGIN_SRC python # settings.py LOGGING = { "version": 1, "disable_existing_loggers": False, "handlers": { "file": { "level": "INFO", "class": "logging.FileHandler", "filename": "/var/log/django/liveview.log", }, }, "loggers": { "liveview": { "handlers": ["file"], "level": "INFO", "propagate": False, }, }, } #+END_SRC Monitor key metrics: - WebSocket connection count - Redis memory usage - Handler execution time - Error rates - Message throughput ** Security Checklist - ✓ Use HTTPS/WSS in production - ✓ Set ~ALLOWED_HOSTS~ correctly - ✓ Use ~AllowedHostsOriginValidator~ in ASGI - ✓ Enable Redis password authentication - ✓ Set proper CORS headers if needed - ✓ Use authentication middleware for protected handlers - ✓ Validate and sanitize all user input - ✓ Set WebSocket rate limiting - ✓ Monitor for suspicious activity ** Scaling Horizontally To scale beyond one server: 1. **Use external Redis** (not Redis on the same server) 2. **Share sessions** across servers (database or Redis sessions) 3. **Load balance** WebSocket connections (sticky sessions not required) 4. **Use Redis Sentinel or Cluster** for Redis high availability 5. **Monitor all instances** with centralized logging * API Reference :PROPERTIES: :ONE: one-custom-default-doc :CUSTOM_ID: /docs/api-reference/ :TITLE: API Reference :DESCRIPTION: Complete API reference for Django LiveView. :NAVIGATOR-ACTIVE: docs :END: Complete reference for all Django LiveView functions, decorators, and classes. ** Decorators *** ~@liveview_handler(name)~ Register a function as a LiveView handler. **Parameters:** - ~name~ (str): The handler name that will be called from the frontend **Returns:** Decorated function **Example:** #+BEGIN_SRC python from liveview import liveview_handler @liveview_handler("my_handler") def my_handler(consumer, content): pass #+END_SRC ** Functions *** ~send(consumer, data, broadcast=False)~ Send data to the client(s). **Parameters:** - ~consumer~ (WebsocketConsumer): The WebSocket consumer instance - ~data~ (dict): Dictionary containing the update instructions - ~broadcast~ (bool, optional): If True, send to all connected clients. Default: False **Data dictionary keys:** - ~target~ (str, required): CSS selector of element to update - ~html~ (str, optional): HTML content to insert - ~append~ (bool, optional): If True, append instead of replace. Default: False - ~remove~ (bool, optional): If True, remove the target element. Default: False - ~url~ (str, optional): Update browser URL (uses ~history.pushState~) - ~title~ (str, optional): Update page title - ~scroll~ (str, optional): CSS selector to scroll to - ~scrollTop~ (bool, optional): If True, scroll to top of page - ~script~ (str, optional): JavaScript code to execute **Returns:** None **Examples:** #+BEGIN_SRC python from liveview import send # Replace content send(consumer, { "target": "#container", "html": "

New content

" }) # Append content send(consumer, { "target": "#list", "html": "
  • New item
  • ", "append": True }) # Remove element send(consumer, { "target": "#notification-123", "remove": True }) # Update URL and title send(consumer, { "target": "#main", "html": "

    New Page

    ", "url": "/new-page/", "title": "New Page - My Site" }) # Execute JavaScript send(consumer, { "script": "console.log('Hello from server');" }) # Broadcast to all users send(consumer, { "target": "#notifications", "html": "

    New message!

    ", "append": True }, broadcast=True) #+END_SRC ** Handler Content Parameter Every handler receives a ~content~ dictionary with the following structure: #+BEGIN_SRC python { "function": str, # Handler name that was called "form": dict, # All form inputs {name: value} "data": dict, # Custom data-data-* attributes "lang": str, # Current language (from ) "room": str # WebSocket room identifier } #+END_SRC **Example access:** #+BEGIN_SRC python @liveview_handler("example") def example(consumer, content): function_name = content["function"] # "example" user_input = content["form"]["name"] # From custom_id = content["data"]["item_id"] # From data-data-item-id language = content["lang"] # "en", "es", etc. room_id = content["room"] # "user_123" or UUID #+END_SRC ** Registry *** ~liveview_registry~ Global registry for handlers and middleware. **Methods:** **** ~add_middleware(middleware_func)~ Add a middleware function to run before handlers. **Parameters:** - ~middleware_func~ (callable): Function with signature ~(consumer, content, function_name) -> bool~ **Returns:** None **Example:** #+BEGIN_SRC python from liveview import liveview_registry def my_middleware(consumer, content, function_name): # Run before handler print(f"Calling {function_name}") # Return True to continue, False to cancel return True liveview_registry.add_middleware(my_middleware) #+END_SRC **** ~list_functions()~ Get list of all registered handler names. **Returns:** List[str] **Example:** #+BEGIN_SRC python from liveview import liveview_registry handlers = liveview_registry.list_functions() print(handlers) # ['say_hello', 'load_articles', ...] #+END_SRC ** Routing *** ~get_liveview_urlpatterns()~ Get URL patterns for WebSocket routing. **Returns:** List of URL patterns **Example:** #+BEGIN_SRC python from liveview.routing import get_liveview_urlpatterns from channels.routing import URLRouter application = URLRouter( get_liveview_urlpatterns() ) #+END_SRC ** Frontend Attributes *** HTML Attributes **** ~data-controller="page"~ Required on ~~ or ~~ to activate Stimulus controller. #+BEGIN_SRC html #+END_SRC **** ~data-room="room-id"~ WebSocket room identifier. Use ~{% liveview_room_uuid %}~ to generate a random UUID for each request. #+BEGIN_SRC html {% load liveview %} #+END_SRC **** ~data-liveview-function="handler-name"~ Specifies which handler to call. #+BEGIN_SRC html #+END_SRC **** ~data-action="event->controller#action"~ Stimulus action binding. #+BEGIN_SRC html
    #+END_SRC **** ~data-data-*~ Custom data attributes sent to handler in ~content["data"]~. #+BEGIN_SRC html

    Hello, World!

    {% endblock %} #+END_SRC ** Step 7: Run your project #+BEGIN_SRC sh # Run Django with Daphne (ASGI server) python manage.py runserver #+END_SRC That's it! Click the button and see real-time updates. 🎉 Congratulations! You have created your first real-time SPA using Django LiveView. ** Live Demo Want to see Django LiveView in action? Check out our [[https://django-liveview-demo.andros.dev/][live demo]] showcasing a real-time alert system with WebSocket communication, form validation, and broadcast notifications. The source code is available at [[https://github.com/Django-LiveView/demo-alarms][GitHub]]. * Source code :PROPERTIES: :ONE: one-custom-default-page :CUSTOM_ID: /source-code/ :TITLE: Source code :DESCRIPTION: List of all related source code. :NAVIGATOR-ACTIVE: source code :END: You can find all the source code in the following repositories: - [[https://github.com/Django-LiveView/liveview][LiveView]]: Source code of the Django framework published on PyPI as django-liveview - [[https://github.com/Django-LiveView/docs][Website and Docs]]: All documentation, including this page - Templates - [[https://github.com/Django-LiveView/starter-template][Starter]]: Check all the features of Django LiveView - [[https://github.com/Django-LiveView/minimal-template][Minimal]]: The minimal template to get started - Demos - [[https://github.com/Django-LiveView/demo-alarms][Alert System]]: Real-time alert system with WebSocket communication ([[https://django-liveview-demo.andros.dev/][live demo]]) - [[https://github.com/Django-LiveView/demo-snake][Snake]]: The classic game of Snake * Books :PROPERTIES: :ONE: one-custom-default-page :CUSTOM_ID: /books/ :TITLE: :DESCRIPTION: Books about Django LiveView. :NAVIGATOR-ACTIVE: books :END: There are no books specifically about Django LiveView yet, but you can find books about Django working with HTML over the Wire technology. ** Building SPAs with Django and HTML Over the Wire #+ATTR_HTML: :class center-block image image--cover-book [[#/img/books/building-spas.avif][Building SPAs with Django and HTML Over the Wire]] Learn to build real-time single page applications with Python. Buy: - [[https://www.packtpub.com/product/building-spas-with-django-and-html-over-the-wire/9781803240190][Packt]] - [[https://www.amazon.com/Real-time-Django-over-Wire-Channels-ebook/dp/B0B3DV54ZT/][Amazon.com]] - [[https://www.amazon.es/Real-time-Django-over-Wire-Channels-ebook/dp/B0B3DV54ZT/][Amazon.es]]