{{ post.title }}
{{ post.content }}
* 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: ** Build real-time SPAs with Python, not JavaScript 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. ** See it in action Here's a complete example: a button that loads the latest blog article with a single click. *HTML:* #+BEGIN_SRC html
#+END_SRC *Python:* #+BEGIN_SRC python from liveview import liveview_handler, send from django.template.loader import render_to_string @liveview_handler("load_latest_article") def load_latest_article(consumer, content): # Get the latest article from database article = Article.objects.latest('published_at') # Render with Django templates html = render_to_string('article.html', { 'article': article }) # Send to frontend send(consumer, { "target": "#article-container", "html": html }) #+END_SRC *Result (after clicking the button):* #+BEGIN_SRC htmlDjango Channels extends Django to handle WebSockets, long-running connections, and background tasks...
Read moreHello, World!
" }) #+END_SRC HTML code: #+BEGIN_SRC htmlHello, {name}! Welcome to Django LiveView.
" }) #+END_SRC HTML code: #+BEGIN_SRC html{{ user.bio }}
{% if is_online %} Online {% endif %}{{ post.content }}
You have a new message
Loading more...
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 %}{{ field }}: {{ error_list.0 }}
{% endfor %}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 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
This is a comment