A Brief History of My Web Development Antics
Okay, this is one for the history books (at least mine). I’ve been doing systems
like Ruby on Rails since before RoR. There was this system from Microsoft that
used to work with SQLServer in the 90s called IDC/HTX for internet database
connector** and html template language. It was from before they had
Active Server Pages, that tech that gave all those webpages .asp
extensions.
Yeah, I go back in webdev to before that. But ever since JavaScript really took
off, I’ve hated it with a passion. I am not a ReactJS guy. Even Ruby on
Rails, as “joyful” of a web framework as it was, it was opinonated in all
the wrong ways for me. I tried making a snappy AJAX-driven generalized web
frameork for this thing I did called HitTail from back in 2006, but I got
distracted, lost motivation and let it fizzle out.
Returning to Web Development with a Modern Twist
I’m back! It seems this guy Jeremy Howard, the guy behind a lot of the Pythonic “for humans” (simplified APIs in the spirit of the Requests package) took on Web Development the Python way, and I’m interested again. It’s time to dust off my… no, I never really stopped. I’m a vim/NeoVim guy and even though I swore off WebDev, I type and I type and I type almost every day. I’m the opposite of DRY (don’t repeat yourself). I’m WET (we enjoy typing). So as it turns out as I pivot my career once again, being a year in at the Botify SEO automation tools company (could there be a better fit) and with AI changing everything, I’m full of motivation and raring to go. Correction, I have gone. And now I document and soft-launch. If you’re joining me now, you’re in for something special (if you’re mclovin some superbad tech).
A whole bunch of prompting.
Certainly! Below is the revised introductory guide for Botifython, incorporating your provided code examples. This guide maintains the comprehensive overview and installation instructions while ensuring that all code snippets accurately reflect your project’s implementation.
Welcome to Botifython
Welcome to Botifython! This guide will walk you through everything you need to get started with our cutting-edge web framework designed specifically for SEO and Data Science, with a particular emphasis on AI. From effortless installation to exploring key features and understanding the core architecture, this guide ensures you have all the information necessary to hit the ground running.
Table of Contents
- Introduction
- Installation Guide
- System Overview
- Key Features
- Core Components
- Behind the Scenes
- Distribution Models
- Auto-Update Mechanism
- Next Steps
- Support and Resources
Introduction
Botifython is a powerful web framework built on top of the FastHTML ecosystem, leveraging the robust and deterministic environment management provided by Nix Flakes. Whether you’re focused on SEO optimization, data science projects, or integrating advanced AI functionalities, Botifython provides a seamless and consistent development experience across various operating systems.
Why Botifython?
- Deterministic Environments: Eliminate the “Not on my machine” problem with reproducible setups.
- AI Integration: Harness the power of AI to enhance your SEO and data science tasks.
- Ease of Use: Minimal commands to get started, allowing you to focus on development.
- Flexible Distribution: Choose between open-source and proprietary versions based on your needs.
Installation Guide
Getting started with Botifython is straightforward and designed to solve the “Not on my machine” problem with minimal effort. Follow the steps below to set up your development environment seamlessly.
Prerequisites
Before you begin, ensure that your machine meets the following prerequisites:
- Access to a Terminal: Command-line interface for executing commands.
- Operating System:
- Linux: Native support.
- macOS: Native support.
- Windows: Use WSL2 (Windows Subsystem for Linux) for compatibility.
All dependencies, including Python, are managed by Nix, simplifying the setup process.
Downloading Botifython
You have two options to obtain Botifython:
-
Clone the Repository Using Git:
If you have Git installed, you can clone the repository directly:
git clone git@github.com:YourUsername/Botifython.git
-
Download the ZIP Archive:
If you prefer not to use Git, you can download the repository as a ZIP file:
- Navigate to the Botifython GitHub repository.
- Click on the “Code” button.
- Select “Download ZIP” and extract it to your desired location.
Running the Determinate Nix Installer
Once you have Botifython downloaded, navigate to the project directory and run the Determinate Nix Installer to set up the environment. The Determinate Nix Installer simplifies the Nix installation process with enhanced features over the official installer.
Why Use the Determinate Nix Installer?
- Fast and Reliable: Quick and stable installations across various platforms.
- Multi-Platform Support: Works on macOS (Intel and ARM), Linux (Intel and ARM), Windows via WSL2, and even SteamOS.
- Opinionated Configuration: Sets up Nix with flakes and the unified CLI enabled by default.
- Seamless Uninstallation: Provides a straightforward way to uninstall Nix if needed.
- CI/CD Integration: Optimized for continuous integration and deployment environments.
Installation Command:
curl --proto '=https' --tlsv1.2 -sSf -L https://install.determinate.systems/nix | sh -s -- install
Note: Ensure that you have an active internet connection to download the installer.
Entering the Development Environment
After the Determinate Nix Installer completes the setup, enter the Botifython development environment using Nix:
cd Botifython
nix develop
This command initializes the environment based on the flake.nix
configuration, ensuring all dependencies and tools are correctly set up. With just three simple steps—download the repository (via Git or ZIP), run the Determinate Nix Installer, and execute nix develop
—your deterministic development environment is ready!
No need to manually manage permissions or install Python separately; Nix handles all dependencies seamlessly.
System Overview
Understanding the underlying architecture of Botifython will help you make the most of its features. Let’s delve into the core components that power our web framework.
Nix Package System
Nix is a powerful package management system that aims to provide reproducible, declarative, and reliable software builds and deployments. It is built on several key principles:
- Pure Functional Model: Packages are built in isolation with explicitly declared dependencies, ensuring that builds are reproducible and side-effect free.
- Immutable Store: Each package version is stored with a unique hash, allowing multiple versions to coexist without conflicts.
- Declarative Configuration: System configurations are defined in Nix expressions, enabling reproducible environments across different machines.
- Atomic Upgrades and Rollbacks: Changes to the system are atomic, and you can easily revert to previous configurations if issues arise.
Nix Flakes Explained
Nix Flakes are an experimental feature introduced to enhance Nix’s capabilities, addressing some limitations of the original system. They provide a more standardized and reproducible way to manage Nix projects.
Key Features of Nix Flakes:
- Standardized Structure: Flakes provide a consistent format for Nix projects, improving discoverability and composition.
- Explicit Dependencies: Flakes require explicit declaration of dependencies, enhancing reproducibility.
- Improved Hermetic Builds: Flakes enforce pure evaluation mode, disallowing access to mutable state like environment variables.
- Lock File Mechanism: Flakes use a
flake.lock
file to pin dependency versions, similar to other package managers. - Simplified CLI: The experimental
nix
command offers a more user-friendly interface for working with flakes.
Key Components of a Flake:
- flake.nix: The main configuration file defining inputs and outputs.
- inputs: Dependencies of the flake, typically other flakes or Nix packages.
- outputs: A function that produces an attribute set containing packages, NixOS modules, or other Nix artifacts.
- flake.lock: An automatically generated file that locks input versions for consistent builds.
Benefits of Flakes:
- Improved Composability: Easily depend on and compose with other flakes.
- Better Reproducibility: Locked dependencies ensure consistent builds across different environments.
- Standardized Interface: Provides a uniform way to interact with Nix projects.
- Enhanced Distribution: Flakes can be easily shared and referenced using URLs.
Development Shells
Botifython defines separate development shells for different operating systems, ensuring optimal compatibility and performance.
- Linux Development Shell: Includes additional packages like GCC and CUDA support if available.
- macOS Development Shell: Tailored for Unix-based macOS environments.
- Windows Development Shell: Utilize WSL2 for compatibility with Linux-based development tools.
Each shell is configured with:
- Common Packages: Managed by Nix, including Python, pip, virtualenv, git, curl, wget, cmake, htop, and more.
- Shell Hooks: Scripts that set up the environment upon entering the shell, including activating virtual environments and customizing the prompt.
Nix handles all dependencies, eliminating the need for manual installations or permission management.
Key Features
Botifython is packed with features designed to streamline your development workflow. Let’s explore them in detail.
Reproducible Environments
With Nix Flakes, Botifython guarantees that every developer operates within an identical environment. This eliminates the infamous “it works on my machine” issue, fostering smoother collaboration and consistent builds.
Benefits:
- Consistency: Every dependency and tool is version-controlled.
- Portability: Easily replicate environments across different machines and operating systems.
- Efficiency: Save time on environment setup and troubleshooting.
Automated Setup Scripts
Our automated setup scripts handle the heavy lifting of environment configuration, allowing you to focus on development rather than setup.
Features:
- Virtual Environment Activation: Automatically activates Python virtual environments.
- Welcome Messages: Uses
pyfiglet
to display stylish ASCII art welcome messages. - Package Installation: Installs necessary Python packages seamlessly.
- Chatbot Integration: Sets up and manages chatbot interactions within the development environment.
Integrated Chatbot Support
Botifython integrates a powerful AI-driven chatbot to assist with your development tasks, providing interactive and intelligent support.
Features:
- AI-Powered Interactions: Engage with the chatbot to receive assistance and suggestions.
- Streaming Responses: Enjoy real-time, word-by-word streaming of chatbot responses for a dynamic experience.
- Customizable Prompts: Tailor the chatbot’s behavior and responses to fit your project’s needs.
Seamless Git Integration
Version control is at the heart of modern development. Botifython offers robust Git integration to manage your codebase effectively.
Capabilities:
- Version Control: Track changes, manage branches, and collaborate with team members effortlessly.
- Automated Updates: The system can perform
git pull
operations using securely generated SSH keys to keep your repository up-to-date. - Conflict Resolution: Tools and scripts to handle merge conflicts and ensure smooth collaboration.
AI-Powered SEO Tools
Botifython leverages AI to enhance SEO and data science tasks, providing intelligent features that automate and optimize your workflows.
Features:
- Automated SEO Audits: AI-driven tools analyze your website for SEO performance, providing actionable insights.
- Data Analysis: Advanced data science tools integrated with AI capabilities for predictive analytics and trend analysis.
- Custom AI Models: Easily integrate and deploy custom AI models tailored to your specific needs.
Core Components
Botifython’s architecture is modular and extensible, allowing for easy integration of new features and applications. Below are the core components that form the backbone of the framework.
BaseApp Class
The BaseApp
class serves as the foundation for creating application components with common CRUD operations. It provides a template for building application components that interact with database tables and handle basic Create, Read, Update, Delete (CRUD) operations.
class BaseApp:
"""
A base class for creating application components with common CRUD operations.
This class provides a template for building application components that interact
with database tables and handle basic Create, Read, Update, Delete (CRUD) operations.
It includes methods for registering routes, rendering items, and performing various
database operations.
The class is designed to be flexible and extensible, allowing subclasses to override
or extend its functionality as needed for specific application components.
"""
def __init__(self, name, table, toggle_field=None, sort_field=None, sort_dict=None):
self.name = name
self.table = table
self.toggle_field = toggle_field
self.sort_field = sort_field
self.sort_dict = sort_dict or {'id': 'id', sort_field: sort_field}
def register_routes(self, rt):
# Register routes: create, read, update, delete, toggle, and sort
rt(f'/{self.name}', methods=['POST'])(self.insert_item)
rt(f'/{self.name}/', methods=['POST'])(self.update_item) # Changed to POST
rt(f'/{self.name}/delete/', methods=['DELETE'])(self.delete_item)
rt(f'/{self.name}/toggle/', methods=['POST'])(self.toggle_item)
rt(f'/{self.name}_sort', methods=['POST'])(self.sort_items)
def get_action_url(self, action, item_id):
"""
Generate a URL for a specific action on an item.
Args:
action (str): The action method (e.g., 'delete', 'toggle').
item_id (int): The ID of the item.
Returns:
str: The constructed URL.
"""
return f"/{self.name}/{action}/{item_id}"
def render_item(self, item):
# A wrapper function currently serving as a passthrough for item rendering.
# This method is part of the system's "styling" mechanism, transforming
# dataclasses into HTML or other instructions for display or HTMX operations.
# Subclasses are expected to override this method with context-aware implementations.
return item
async def delete_item(self, request, item_id: int):
"""
Delete an item from the table.
Args:
request: The incoming request object.
item_id (int): The ID of the item to delete.
Returns:
str: An empty string indicating successful deletion.
"""
try:
logger.debug(f"Attempting to delete item ID: {item_id}")
self.table.delete(item_id)
prompt = f"Item {item_id} deleted. Brief, sassy reaction."
await chatq(prompt)
logger.info(f"Deleted item ID: {item_id}")
return ''
except Exception as e:
logger.error(f"Error deleting item: {str(e)}")
return f"Error deleting item: {str(e)}", 500
async def toggle_item(self, request, item_id: int):
"""
Toggle a boolean field of an item.
Args:
request: The incoming request object.
item_id (int): The ID of the item to toggle.
Returns:
dict: The rendered updated item.
"""
try:
logger.debug(f"Toggling {self.toggle_field} for item ID: {item_id}")
item = self.table[item_id]
current_status = getattr(item, self.toggle_field)
setattr(item, self.toggle_field, not current_status)
updated_item = self.table.update(item)
logger.info(f"Toggled {self.toggle_field} for item ID {item_id} to {getattr(updated_item, self.toggle_field)}")
prompt = f"Item {item_id} toggled. Brief, sassy reaction."
await chatq(prompt)
return self.render_item(updated_item) # Use the subclass's render_item method
except Exception as e:
logger.error(f"Error toggling item: {str(e)}")
return f"Error toggling item: {str(e)}", 500
async def sort_items(self, request):
"""
Update the order of items based on the received values.
"""
logger.debug(f"Received request to sort {self.name}.")
try:
values = await request.form() # Get form data from request
items = json.loads(values.get('items', '[]')) # Decode JSON string to list
logger.debug(f"Parsed items: {items}")
for item in items:
logger.debug(f"Updating item: {item}")
update_dict = {self.sort_field: int(item['priority'])} # Use priority
self.table.update(id=int(item['id']), **update_dict) # Update table entry
logger.info(f"{self.name.capitalize()} order updated successfully")
prompt = f"The {self.name} list was reordered. Make a brief, witty remark about sorting or prioritizing. Keep it under 20 words."
await chatq(prompt)
return ''
except json.JSONDecodeError as e:
logger.error(f"JSON decode error: {str(e)}")
return "Invalid data format", 400
except Exception as e:
logger.error(f"Error updating {self.name} order: {str(e)}")
return str(e), 500
async def insert_item(self, request):
"""
Create a new item in the table.
"""
try:
form = await request.form()
new_item_data = self.prepare_insert_data(form)
if not new_item_data: # If prepare_insert_data returns empty string or empty dict
return '' # Return empty string, which won't be rendered in the DOM
new_item = await self.create_item(**new_item_data)
return self.render_item(new_item)
except Exception as e:
logger.error(f"Error inserting {self.name}: {str(e)}")
return str(e), 500
async def update_item(self, request, item_id: int):
"""
Update an existing item in the table.
"""
try:
form = await request.form()
update_data = self.prepare_update_data(form)
if not update_data: # If prepare_update_data returns empty string or empty dict
return '' # Return empty string, which won't be rendered in the DOM
item = self.table[item_id]
for key, value in update_data.items():
setattr(item, key, value)
updated_item = self.table.update(item)
logger.info(f"Updated {self.name} item {item_id}")
return self.render_item(updated_item)
except Exception as e:
logger.error(f"Error updating {self.name} {item_id}: {str(e)}")
return str(e), 500
def prepare_insert_data(self, form):
"""
Prepare data for insertion. To be overridden by subclasses.
"""
raise NotImplementedError("Subclasses must implement prepare_insert_data")
def prepare_update_data(self, form):
"""
Prepare data for update. To be overridden by subclasses.
"""
raise NotImplementedError("Subclasses must implement prepare_update_data")
async def create_item(self, **kwargs):
"""
Create a new item in the table.
Args:
**kwargs: The fields and values for the new item.
Returns:
The newly created item.
"""
try:
logger.debug(f"Creating new {self.name} with data: {kwargs}")
new_item = self.table.insert(kwargs)
logger.info(f"Created new {self.name}: {new_item}")
return new_item
except Exception as e:
logger.error(f"Error creating {self.name}: {str(e)}")
raise e
TodoApp and ProfileApp
Building upon the BaseApp
class, TodoApp
and ProfileApp
handle specific functionalities related to managing tasks and user profiles, respectively.
class TodoApp(BaseApp):
def __init__(self, table):
# Extract the name from the table object
super().__init__(
name=table.name,
table=table,
toggle_field='done',
sort_field='priority'
)
def render_item(self, todo):
return render_todo(todo)
def prepare_insert_data(self, form):
title = form.get('title', '').strip()
if not title:
return '' # Return empty string instead of raising an exception
current_profile_id = db.get("last_profile_id", 1)
max_priority = max((t.priority or 0 for t in self.table()), default=-1) + 1
return {
"title": title,
"done": False,
"priority": max_priority,
"profile_id": current_profile_id,
}
def prepare_update_data(self, form):
title = form.get('title', '').strip()
if not title:
return '' # Return empty string instead of raising an exception
return {
"title": title,
"done": form.get('done', '').lower() == 'true',
}
class ProfileApp(BaseApp):
def __init__(self, table):
super().__init__(
name=table.name,
table=table,
toggle_field='active',
sort_field='priority'
)
def render_item(self, profile):
return render_profile(profile)
def prepare_insert_data(self, form):
profile_name = form.get('profile_name', '').strip()
if not profile_name:
return '' # Return empty string instead of raising an exception
max_priority = max((p.priority or 0 for p in self.table()), default=-1) + 1
return {
"name": profile_name,
"address": form.get('profile_address', '').strip(),
"code": form.get('profile_code', '').strip(),
"active": True,
"priority": max_priority,
}
def prepare_update_data(self, form):
profile_name = form.get('profile_name', '').strip()
if not profile_name:
return '' # Return empty string instead of raising an exception
return {
"name": profile_name,
"address": form.get('profile_address', '').strip(),
"code": form.get('profile_code', '').strip(),
"active": form.get('active', '').lower() == 'true',
}
DictLikeDB Wrapper
The DictLikeDB
class provides a robust wrapper for dictionary-like persistent storage, ensuring seamless interaction with various storage backends.
class DictLikeDB:
"""
A robust wrapper for dictionary-like persistent storage.
This class provides a familiar dict-like interface to interact with
various types of key-value stores, including databases and file systems.
It emphasizes the power and flexibility of key-value pairs as a
fundamental data structure in programming and system design.
Key features:
1. Persistence: Data survives beyond program execution.
2. Dict-like API: Familiar Python dictionary operations.
3. Adaptability: Can wrap different storage backends.
4. Logging: Built-in logging for debugging and monitoring.
By abstracting the underlying storage mechanism, this class allows
for easy swapping of backends without changing the client code.
This demonstrates the power of Python's duck typing and the
universality of the key-value paradigm across different storage solutions.
"""
def __init__(self, store, Store):
self.store = store
self.Store = Store
def __getitem__(self, key):
"""Retrieve an item from the store by key."""
try:
value = self.store[key].value
logger.debug(f"Retrieved from DB: {key} = {value}")
return value
except NotFoundError:
logger.error(f"Key not found: {key}")
raise KeyError(key)
def __setitem__(self, key, value):
"""Set an item in the store by key."""
try:
self.store.update({"key": key, "value": value})
logger.info(f"Updated persistence store: {key} = {value}")
except NotFoundError:
self.store.insert({"key": key, "value": value})
logger.info(f"Inserted new item in persistence store: {key} = {value}")
def __delitem__(self, key):
"""Delete an item from the store by key."""
try:
self.store.delete(key)
logger.warning(f"Deleted key from persistence store: {key}")
except NotFoundError:
logger.error(f"Attempted to delete non-existent key: {key}")
raise KeyError(key)
def __contains__(self, key):
"""Check if a key exists in the store."""
exists = key in self.store
logger.debug(f"Key '{key}' exists: {exists}")
return exists
def __iter__(self):
"""Iterate over the keys in the store."""
for item in self.store():
yield item.key
def items(self):
"""Return key-value pairs in the store."""
for item in self.store():
yield item.key, item.value
def keys(self):
"""Return a list of keys in the store."""
return list(self)
def values(self):
"""Return values in the store."""
for item in self.store():
yield item.value
def get(self, key, default=None):
"""Get an item from the store, returning default if not found."""
try:
return self[key]
except KeyError:
logger.debug(f"Key '{key}' not found. Returning default.")
return default
Behind the Scenes
To truly appreciate Botifython, let’s take a closer look at the key components and the code that powers them.
Key Components and Code Examples
1. Nix Flake Configuration (flake.nix
)
The flake.nix
file is the heart of Botifython, defining dependencies and development environments.
{
description = "Botifython: A deterministic web framework for SEO and Data Science with AI capabilities.";
inputs = {
nixpkgs.url = "github:NixOS/nixpkgs/nixos-unstable";
flake-utils.url = "github:numtide/flake-utils";
};
outputs = { self, nixpkgs, flake-utils }:
flake-utils.lib.eachDefaultSystem (system:
let
pkgs = import nixpkgs {
inherit system;
config = { allowUnfree = true; };
};
isDarwin = pkgs.stdenv.isDarwin;
isLinux = pkgs.stdenv.isLinux;
commonPackages = with pkgs; [
python311 # Python 3.11 interpreter
python311.pkgs.pip # Package installer for Python
python311.pkgs.virtualenv # Tool to create isolated Python environments
figlet # For creating ASCII art welcome messages
tmux # Terminal multiplexer for managing sessions
zlib # Compression library for data compression
git # Version control system for tracking changes
curl # Command-line tool for transferring data with URLs
wget # Utility for non-interactive download of files from the web
cmake # Cross-platform build system generator
htop # Interactive process viewer for Unix systems
] ++ (with pkgs; pkgs.lib.optionals isLinux [
gcc # GNU Compiler Collection for compiling C/C++ code
stdenv.cc.cc.lib # Standard C library for Linux systems
]);
# This script sets up our Python environment and project
runScript = pkgs.writeShellScriptBin "run-script" ''
#!/usr/bin/env bash
# Activate the virtual environment
source .venv/bin/activate
# Create a fancy welcome message
REPO_NAME=$(basename "$PWD")
PROPER_REPO_NAME=$(echo "$REPO_NAME" | awk '{print toupper(substr($0,1,1)) tolower(substr($0,2))}')
figlet "$PROPER_REPO_NAME"
echo "Welcome to the $PROPER_REPO_NAME development environment on ${system}!"
echo
# Install Python packages from requirements.txt
echo "- Installing pip packages..."
if pip install --upgrade pip --quiet && \
pip install -r requirements.txt --quiet; then
package_count=$(pip list --format=freeze | wc -l)
echo "- Done. $package_count pip packages installed."
else
echo "Warning: An error occurred during pip setup."
fi
# Check if numpy is properly installed
if python -c "import numpy" 2>/dev/null; then
echo "- numpy is importable (good to go!)"
echo
echo "To start JupyterLab, type: start"
echo "To stop JupyterLab, type: stop"
echo
else
echo "Error: numpy could not be imported. Check your installation."
fi
# Create convenience scripts for managing JupyterLab
cat << EOF > .venv/bin/start
#!/bin/sh
echo "A JupyterLab tab will open in your default browser."
tmux kill-session -t jupyter 2>/dev/null || echo "No tmux session named 'jupyter' is running."
tmux new-session -d -s jupyter 'source .venv/bin/activate && jupyter lab --NotebookApp.token="" --NotebookApp.password="" --NotebookApp.disable_check_xsrf=True'
echo "If no tab opens, visit http://localhost:8888"
echo "To view JupyterLab server: tmux attach -t jupyter"
echo "To stop JupyterLab server: stop"
EOF
chmod +x .venv/bin/start
cat << EOF > .venv/bin/stop
#!/bin/sh
echo "Stopping tmux session 'jupyter'..."
tmux kill-session -t jupyter 2>/dev/null || echo "No tmux session named 'jupyter' is running."
echo "The tmux session 'jupyter' has been stopped."
EOF
chmod +x .venv/bin/stop
'';
# Define the development shell for Linux systems (including WSL)
linuxDevShell = pkgs.mkShell {
buildInputs = commonPackages ++ (with pkgs; pkgs.lib.optionals (builtins.pathExists "/usr/bin/nvidia-smi") cudaPackages);
shellHook = ''
# Set up the Python virtual environment
test -d .venv || ${pkgs.python311}/bin/python -m venv .venv
export VIRTUAL_ENV="$(pwd)/.venv"
export PATH="$VIRTUAL_ENV/bin:$PATH"
# Customize the prompt to show we're in a Nix environment
export PS1='$(printf "\033[01;34m(nix) \033[00m\033[01;32m[%s@%s:%s]$\033[00m " "\u" "\h" "\w")'
export LD_LIBRARY_PATH=${pkgs.lib.makeLibraryPath commonPackages}:$LD_LIBRARY_PATH
# Set up CUDA if available
if command -v nvidia-smi &> /dev/null; then
echo "CUDA hardware detected."
export CUDA_HOME=${pkgs.cudatoolkit}
export PATH=$CUDA_HOME/bin:$PATH
export LD_LIBRARY_PATH=$CUDA_HOME/lib64:$LD_LIBRARY_PATH
else
echo "No CUDA hardware detected."
fi
# Run our setup script
${runScript}/bin/run-script
'';
};
# Define the development shell for macOS systems
darwinDevShell = pkgs.mkShell {
buildInputs = commonPackages;
shellHook = ''
# Set up the Python virtual environment
test -d .venv || ${pkgs.python311}/bin/python -m venv .venv
export VIRTUAL_ENV="$(pwd)/.venv"
export PATH="$VIRTUAL_ENV/bin:$PATH"
# Customize the prompt to show we're in a Nix environment
export PS1='$(printf "\033[01;34m(nix) \033[00m\033[01;32m[%s@%s:%s]$\033[00m " "\u" "\h" "\w")'
export LD_LIBRARY_PATH=${pkgs.lib.makeLibraryPath commonPackages}:$LD_LIBRARY_PATH
# Run our setup script
${runScript}/bin/run-script
'';
};
in {
# Choose the appropriate development shell based on the OS
devShell = if isLinux then linuxDevShell else darwinDevShell; # Ensure multi-OS support
});
}
Highlights:
- Description: Provides a brief overview of Botifython’s purpose.
- Inputs: Specifies dependencies like
nixpkgs
andflake-utils
. - Common Packages: Lists essential tools and packages available in the environment.
- Run Script: Automates environment setup tasks.
- Development Shell (
devShell
): Configures the shell with necessary tools and hooks based on the operating system.
2. Automated Setup Script (run-script
)
The run-script
automates the setup process each time you enter the development environment.
#!/usr/bin/env bash
# Activate the virtual environment
source .venv/bin/activate
# Create a fancy welcome message
fig("Botifython")
echo "Welcome to the Botifython development environment on ${system}!"
echo
# Install Python packages from requirements.txt
echo "- Installing pip packages..."
if pip install --upgrade pip --quiet && \
pip install -r requirements.txt --quiet; then
package_count=$(pip list --format=freeze | wc -l)
echo "- Done. $package_count pip packages installed."
else
echo "Warning: An error occurred during pip setup."
fi
# Check if numpy is properly installed
if python -c "import numpy" 2>/dev/null; then
echo "- numpy is importable (good to go!)"
echo
echo "To start JupyterLab, type: start"
echo "To stop JupyterLab, type: stop"
echo
else
echo "Error: numpy could not be imported. Check your installation."
fi
# Create convenience scripts for managing JupyterLab
cat << EOF > .venv/bin/start
#!/bin/sh
echo "A JupyterLab tab will open in your default browser."
tmux kill-session -t jupyter 2>/dev/null || echo "No tmux session named 'jupyter' is running."
tmux new-session -d -s jupyter 'source .venv/bin/activate && jupyter lab --NotebookApp.token="" --NotebookApp.password="" --NotebookApp.disable_check_xsrf=True'
echo "If no tab opens, visit http://localhost:8888"
echo "To view JupyterLab server: tmux attach -t jupyter"
echo "To stop JupyterLab server: stop"
EOF
chmod +x .venv/bin/start
cat << EOF > .venv/bin/stop
#!/bin/sh
echo "Stopping tmux session 'jupyter'..."
tmux kill-session -t jupyter 2>/dev/null || echo "No tmux session named 'jupyter' is running."
echo "The tmux session 'jupyter' has been stopped."
EOF
chmod +x .venv/bin/stop
Features:
- Virtual Environment Activation: Ensures Python dependencies are isolated.
- Package Installation: Automatically installs required Python packages from
requirements.txt
. - Numpy Verification: Checks the integrity of the NumPy installation.
- JupyterLab Management: Provides
start
andstop
scripts for managing JupyterLab sessions viatmux
.
Note: All dependencies, including Python, are managed by Nix, eliminating the need for manual installations.
3. Custom Middleware and Exception Handling
Botifython includes custom middleware for enhanced logging and a custom 404 handler to improve user experience.
class DOMSkeletonMiddleware(BaseHTTPMiddleware):
async def dispatch(self, request, call_next):
# Log the incoming HTTP request
endpoint = request.url.path
method = request.method
f = Figlet(font='slant')
figlet_text = f.renderText(f"{endpoint} {method}")
print(figlet_text)
logger.info(f"HTTP Request: {method} {endpoint}")
# Call the next middleware or request handler
response = await call_next(request)
# Print a rich table of the db key/value pairs
table = Table(title="Database Contents")
table.add_column("Key", style="cyan")
table.add_column("Value", style="magenta")
for key, value in db.items():
table.add_row(key, str(value))
console = Console()
table.columns[1].style = "white"
console.print(table)
return response
def custom_404_handler(request, exc):
"""
Custom 404 page handler.
Args:
request: The request that caused the 404 error.
exc: The exception that was raised.
Returns:
HTML: An HTML response for the 404 error.
"""
return Html(
Head(
Title("404 - Page Not Found"),
Style("""
body { font-family: Arial, sans-serif; text-align: center; padding: 50px; }
h1 { color: #e74c3c; }
a { color: #3498db; text-decoration: none; }
""")
),
Body(
H1("404 - Page Not Found"),
P(f"Sorry, the page '{request.url.path}' you're looking for doesn't exist."),
A("Go back to home", href="/")
)
)
Highlights:
- Logging: Enhanced logging for incoming HTTP requests and database state.
- Custom 404 Page: Provides a user-friendly 404 error page with styling.
Distribution Models
Botifython is distributed in two main versions to cater to different user needs: the Pipulate Free and Open Source (FOSS) version and the Proprietary Botify version. Each version comes with a set of pre-built applications tailored to its specific use case.
Pipulate FOSS Version
The Pipulate FOSS version is designed for users who prefer open-source solutions and want to contribute to the community-driven aspects of Botifython.
Features:
- Open-Source Tools: Access to a wide range of free and open-source SEO and data science tools.
- Community Support: Engage with a vibrant community for support and collaboration.
- Customizability: Full access to source code for customization and extension.
Pre-Built Apps:
- SEO Analyzer: Analyze website SEO performance with comprehensive reports.
- Keyword Research Tool: Discover and analyze target keywords.
- Data Visualization Dashboard: Visualize data insights using interactive dashboards.
Proprietary Botify Version
The Proprietary Botify version offers advanced features and premium tools designed for enterprise users and professionals seeking enhanced capabilities.
Features:
- Advanced AI Integrations: Leverage state-of-the-art AI models for predictive analytics and automation.
- Premium Support: Access to dedicated support channels for personalized assistance.
- Enhanced Security: Additional security features to protect sensitive data and operations.
Pre-Built Apps:
- AI-Powered SEO Optimizer: Utilize AI to optimize SEO strategies and content.
- Automated Reporting Tool: Generate detailed and automated reports for stakeholders.
- Custom Model Deployment: Deploy and manage custom AI models tailored to specific business needs.
Auto-Update Mechanism
Botifython includes an auto-update mechanism that ensures your framework stays up-to-date with the latest features and security patches. This is achieved through a seamless integration with Git and the use of securely managed SSH keys.
SSH Key Setup
-
Generate a Dedicated SSH Key Pair:
Generate a new SSH key pair specifically for the auto-update process:
ssh-keygen -t ed25519 -C "botifython-auto-update" -f ~/.ssh/botifython_auto_update
-
Add the Public Key to GitHub Repository:
- Navigate to your Botifython GitHub repository settings.
- Go to “Deploy Keys” and add the generated public key (
~/.ssh/botifython_auto_update.pub
) with read-only access.
-
Store the Private Key Securely:
Ensure the private key (
~/.ssh/botifython_auto_update
) is stored securely within Botifython’s configuration. It should be encrypted and inaccessible to unauthorized users.
Auto-Update Process
When Botifython runs, it performs the following steps to check for and apply updates:
-
SSH Agent Initialization:
Start an SSH agent and add the private key to it:
eval "$(ssh-agent -s)" ssh-add ~/.ssh/botifython_auto_update
-
Git Configuration:
Set up a temporary Git configuration using the SSH key:
git config --global url."git@github.com:".insteadOf "https://github.com/"
-
Repository Check:
Check for updates by comparing the local and remote repository states:
git fetch origin LOCAL=$(git rev-parse @) REMOTE=$(git rev-parse @{u}) BASE=$(git merge-base @ @{u}) if [ $LOCAL = $REMOTE ]; then echo "Botifython is up to date." elif [ $LOCAL = $BASE ]; then echo "Updating Botifython..." git pull # Trigger any post-update scripts if necessary else echo "Botifython has diverged from the remote repository." # Handle merge conflicts or other issues fi
-
Pull Updates:
If updates are available, perform a
git pull
to fetch and merge the latest changes.
-
Cleanup:
Remove the temporary Git configuration and stop the SSH agent:
git config --global --unset url."git@github.com:".insteadOf ssh-agent -k
Security Considerations
- Limited Access: The deploy key has read-only access to the specific repository, minimizing potential security risks.
- Key Protection: The private key is stored securely within Botifython, ideally encrypted.
- Dedicated Account: A separate GitHub account is used solely for this purpose, isolating it from other operations.
- Secure Communication: All interactions with GitHub use SSH, ensuring encrypted communication.
Next Steps
Now that you’ve familiarized yourself with the installation process, system overview, key features, core components, and the auto-update mechanism, you’re ready to dive deeper into developing with Botifython. Here are some recommended next steps:
- Explore the Codebase: Navigate through the repository to understand the project structure.
- Run the Application: Start the server and interact with the web interface.
- Contribute: Begin making changes, adding features, or fixing bugs.
- Automate Further: Leverage the Python scripts to enhance automation and testing.
- Integrate AI Models: Start integrating and deploying your own AI models to leverage Botifython’s capabilities.
- Customize Your Environment: Modify the
flake.nix
and setup scripts to tailor the environment to your specific needs.
Stay tuned for auto-generated screenshots and further tutorials that will guide you through advanced functionalities and best practices.
Support and Resources
Should you encounter any issues or have questions, our support channels are here to help:
- Documentation: Comprehensive guides and API references.
- Community Forums: Connect with other users and developers.
- Issue Tracker: Report bugs or request features on our GitHub repository.
- Contact Support: Reach out via email or our support portal for personalized assistance.
Conclusion
Welcome to Botifython! By leveraging the power of Nix Flakes and a suite of automated tools, we’ve created a robust and reproducible development environment tailored for SEO and Data Science with a strong emphasis on AI. Whether you’re a seasoned developer or just starting, Botifython is designed to streamline your workflow, foster collaboration, and ensure consistency across projects.
With just three simple commands—download the repository (via Git or ZIP), run the Determinate Nix Installer, and enter the development environment using nix develop
—you can overcome the “Not on my machine” problem and get started quickly. Dive in, explore the features, and enjoy a seamless development experience. Happy coding!
Note: Auto-generated screenshots will be incorporated into this guide shortly, enhancing your visual understanding of the system.
Next article
Comprehensive Documentation for Your Python Web Application
Table of Contents
- Introduction
- System Overview
- Architecture
- Components
- Plugging in New Applications
- Detailed Explanation of
fast_app
Return Values - Examples
- Conclusion
Introduction
Welcome to the comprehensive documentation for your Python web application! This guide is designed to provide an in-depth understanding of your application’s structure, components, and the methodology for integrating new applications seamlessly. Whether you’re a developer familiarizing yourself with the system or a newcomer aiming to extend its functionality, this documentation will serve as your go-to resource.
System Overview
Your application is a modular, extensible web platform built with Python, leveraging several powerful libraries and frameworks. At its core, the application manages various data entities through a set of reusable components, enabling the creation of new functionalities (referred to as “apps”) with minimal friction.
Key Features:
- Modular Architecture: Easily extendable through plugins.
- Database Integration: Robust interaction with persistent storage via a custom wrapper.
- Real-time Chatbot: Integration with a local AI model for dynamic user interactions.
- WebSockets: Support for real-time communication.
- Dynamic Routing: Flexible URL handling with support for both HTTP and WebSocket routes.
- Enhanced UI/UX: Interactive elements powered by HTMX and dynamic JavaScript components.
Architecture
The application follows a modular architecture, where each feature or functionality is encapsulated within its own “app.” This design promotes separation of concerns, making the system easier to maintain and extend.
Core Components:
- Fast App Initialization (
fast_app
): Sets up the core application, including database connections, routing, and middleware. - BaseApp Class: Provides a foundation for creating new applications with common CRUD (Create, Read, Update, Delete) operations.
- Plugin Applications: Specific functionalities like
TodoApp
andProfileApp
that extend the base capabilities. - Database Wrapper (
DictLikeDB
): Abstracts database interactions, offering a dictionary-like interface. - Chatbot Integration: Connects to a local AI model (Ollama) for real-time chatbot responses.
- WebSocket Handling: Manages real-time communication channels for interactive features.
- Middleware and Exception Handling: Enhances request processing and error management.
Components
Fast App Initialization (fast_app
)
The fast_app
function is pivotal in bootstrapping your application. It initializes the core components, sets up database tables, and returns essential handles for further configuration.
Initialization Example:
app, rt, (store, Store), (tasks, Task), (customers, Customer) = fast_app(
"data/data.db",
ws_hdr=True,
live=True,
hdrs=(
SortableJSWithUpdate('.sortable'),
Script(type='module')
),
store={
"key": str,
"value": str,
"pk": "key"
},
task={
"id": int,
"title": str,
"done": bool,
"priority": int,
"profile_id": int,
"pk": "id"
},
customer={
"id": int,
"name": str,
"address": str,
"code": str,
"active": bool,
"priority": int,
"pk": "id"
},
)
Parameters:
- Database Path:
"data/data.db"
specifies the SQLite database file. - WebSocket Headers (
ws_hdr
): Enables WebSocket support. - Live Reload (
live
): Enables live-reloading features for development. - Headers (
hdrs
): Injects JavaScript or other headers into the HTML templates. - Table Definitions: Defines database tables with their respective fields and primary keys.
Return Values:
app
: The main application instance.rt
: A routing function to register endpoints.store
,tasks
,customers
: Database table handles corresponding tostore
,task
, andcustomer
tables.Store
,Task
,Customer
: ORM models or schemas for the respective tables.
BaseApp Class
The BaseApp
class serves as an abstract foundation for creating new application components. It encapsulates common CRUD operations and route registrations, promoting code reuse and consistency across different apps.
Key Features:
- Initialization: Accepts parameters like
name
,table
,toggle_field
,sort_field
, andsort_dict
to configure the app’s behavior. - Route Registration: Automatically registers CRUD routes (
create
,read
,update
,delete
,toggle
,sort
) using the provided routing function. - CRUD Operations: Implements asynchronous methods to handle item creation, retrieval, updating, deletion, and sorting.
- Rendering: Defines a
render_item
method for transforming data items into HTML or other presentation formats. - Error Handling: Includes robust error logging and response mechanisms.
BaseApp Implementation:
class BaseApp:
def __init__(self, name, table, toggle_field=None, sort_field=None, sort_dict=None):
self.name = name
self.table = table
self.toggle_field = toggle_field
self.sort_field = sort_field
self.sort_dict = sort_dict or {'id': 'id', sort_field: sort_field}
def register_routes(self, rt):
# Register routes: create, read, update, delete, toggle, and sort
rt(f'/{self.name}', methods=['POST'])(self.insert_item)
rt(f'/{self.name}/', methods=['POST'])(self.update_item) # Changed to POST
rt(f'/{self.name}/delete/', methods=['DELETE'])(self.delete_item)
rt(f'/{self.name}/toggle/', methods=['POST'])(self.toggle_item)
rt(f'/{self.name}_sort', methods=['POST'])(self.sort_items)
# ... [Other methods like insert_item, update_item, delete_item, toggle_item, sort_items, render_item]
Subclasses Must Implement:
prepare_insert_data(form)
: Prepares data for inserting a new item.prepare_update_data(form)
: Prepares data for updating an existing item.render_item(item)
: Renders the item into an HTML representation.
Plugin Applications (TodoApp
and ProfileApp
)
TodoApp and ProfileApp are concrete implementations of the BaseApp
class, representing specific functionalities within the application. They handle tasks like managing to-do items and user profiles, respectively.
TodoApp Example:
class TodoApp(BaseApp):
def __init__(self, table):
super().__init__(
name=table.name,
table=table,
toggle_field='done',
sort_field='priority'
)
def render_item(self, todo):
return render_todo(todo)
def prepare_insert_data(self, form):
title = form.get('title', '').strip()
if not title:
return '' # Return empty string instead of raising an exception
current_profile_id = db.get("last_profile_id", 1)
max_priority = max((t.priority or 0 for t in self.table()), default=-1) + 1
return {
"title": title,
"done": False,
"priority": max_priority,
"profile_id": current_profile_id,
}
def prepare_update_data(self, form):
title = form.get('title', '').strip()
if not title:
return '' # Return empty string instead of raising an exception
return {
"title": title,
"done": form.get('done', '').lower() == 'true',
}
ProfileApp Example:
class ProfileApp(BaseApp):
def __init__(self, table):
super().__init__(
name=table.name,
table=table,
toggle_field='active',
sort_field='priority'
)
def render_item(self, profile):
return render_profile(profile)
def prepare_insert_data(self, form):
profile_name = form.get('profile_name', '').strip()
if not profile_name:
return '' # Return empty string instead of raising an exception
max_priority = max((p.priority or 0 for p in self.table()), default=-1) + 1
return {
"name": profile_name,
"address": form.get('profile_address', '').strip(),
"code": form.get('profile_code', '').strip(),
"active": True,
"priority": max_priority,
}
def prepare_update_data(self, form):
profile_name = form.get('profile_name', '').strip()
if not profile_name:
return '' # Return empty string instead of raising an exception
return {
"name": profile_name,
"address": form.get('profile_address', '').strip(),
"code": form.get('profile_code', '').strip(),
"active": form.get('active', '').lower() == 'true',
}
Key Points:
- Initialization: Each app specifies its
name
,table
,toggle_field
, andsort_field
. - Data Preparation: Methods to sanitize and structure incoming form data for database operations.
- Rendering: Custom
render_item
methods to transform data into HTML elements.
Database Wrapper (DictLikeDB
)
The DictLikeDB
class is a wrapper that provides a dictionary-like interface for interacting with the database. It abstracts the complexities of direct database operations, allowing for more intuitive data handling.
Features:
- Persistence: Ensures data persists across application runs.
- Dict-like API: Supports standard dictionary operations (
__getitem__
,__setitem__
,__delitem__
, etc.). - Adaptability: Compatible with various storage backends.
- Logging: Integrates with the
loguru
logger for detailed operation logs.
Implementation:
class DictLikeDB:
def __init__(self, store, Store):
self.store = store
self.Store = Store
def __getitem__(self, key):
try:
value = self.store[key].value
logger.debug(f"Retrieved from DB: {key} = {value}")
return value
except NotFoundError:
logger.error(f"Key not found: {key}")
raise KeyError(key)
def __setitem__(self, key, value):
try:
self.store.update({"key": key, "value": value})
logger.info(f"Updated persistence store: {key} = {value}")
except NotFoundError:
self.store.insert({"key": key, "value": value})
logger.info(f"Inserted new item in persistence store: {key} = {value}")
def __delitem__(self, key):
try:
self.store.delete(key)
logger.warning(f"Deleted key from persistence store: {key}")
except NotFoundError:
logger.error(f"Attempted to delete non-existent key: {key}")
raise KeyError(key)
def __contains__(self, key):
exists = key in self.store
logger.debug(f"Key '{key}' exists: {exists}")
return exists
def __iter__(self):
for item in self.store():
yield item.key
def items(self):
for item in self.store():
yield item.key, item.value
def keys(self):
return list(self)
def values(self):
for item in self.store():
yield item.value
def get(self, key, default=None):
try:
return self[key]
except KeyError:
logger.debug(f"Key '{key}' not found. Returning default.")
return default
Usage:
db = DictLikeDB(store, Store)
Common Operations:
- Retrieve Value:
value = db["some_key"]
- Set Value:
db["some_key"] = "new_value"
- Delete Key:
del db["some_key"]
- Check Existence:
"some_key" in db
- Iterate Keys:
for key in db: ...
- Retrieve All Items:
for key, value in db.items(): ...
Chatbot Integration
Your application integrates with a local AI model (Ollama) to provide real-time chatbot interactions. This integration handles generating responses based on user inputs and streaming them back to connected clients via WebSockets.
Key Functions:
get_best_model()
: Selects the most suitable LLaMA model available.chat_with_ollama(model, messages)
: Sends messages to the AI model and retrieves responses.chatq(message)
: Queues messages for streaming to all connected WebSocket users.stream_chat(prompt, quick=False)
: Streams the AI-generated response either instantly or with a typing effect.
WebSocket Event Handlers:
on_conn(ws, send)
: Handles new WebSocket connections.on_disconn(ws)
: Handles WebSocket disconnections.ws(msg)
: Processes incoming WebSocket messages from clients.
Integration Highlights:
- Asynchronous Operations: Utilizes
asyncio
for non-blocking operations. - Streaming Responses: Simulates typing effects for a more natural interaction.
- User Management: Tracks connected users for targeted message delivery.
WebSocket Handling
WebSockets facilitate real-time, bidirectional communication between the server and clients. Your application employs WebSockets to handle chatbot interactions and potentially other real-time features.
WebSocket Management:
- User Tracking: Maintains a
users
dictionary to manage connected clients. - Connection Handlers: Defines
on_conn
andon_disconn
functions to manage connection lifecycle events. - Message Handling: Processes incoming messages and streams responses back to clients.
WebSocket Implementation Example:
@app.ws('/ws', conn=on_conn, disconn=on_disconn)
async def ws(msg: str):
"""Handle incoming WebSocket messages."""
# Process message and generate response
# Stream response back to users
Key Points:
- Connection Establishment: When a client connects,
on_conn
adds them to theusers
dictionary and sends a welcome message. - Disconnection Handling: On disconnection,
on_disconn
removes the user from the tracking dictionary. - Message Processing: Incoming messages trigger chatbot responses, which are streamed back to all connected clients.
Middleware and Exception Handling
Middleware components enhance request processing by injecting additional functionality, such as logging, authentication, or request modification.
DOMSkeletonMiddleware Example:
class DOMSkeletonMiddleware(BaseHTTPMiddleware):
async def dispatch(self, request, call_next):
# Log the incoming HTTP request
endpoint = request.url.path
method = request.method
figlet_text = Figlet(font='slant').renderText(f"{endpoint} {method}")
print(figlet_text)
logger.info(f"HTTP Request: {method} {endpoint}")
# Call the next middleware or request handler
response = await call_next(request)
# Print a rich table of the db key/value pairs
table = Table(title="Database Contents")
table.add_column("Key", style="cyan")
table.add_column("Value", style="magenta")
for key, value in db.items():
table.add_row(key, str(value))
console = Console()
table.columns[1].style = "white"
console.print(table)
return response
Features:
- Request Logging: Logs details of incoming HTTP requests.
- Database State Display: Outputs the current state of the database after each request.
- Integration: Added to the application via
app.add_middleware(DOMSkeletonMiddleware)
.
Exception Handling:
- Custom 404 Handler: Provides a user-friendly page when a requested route is not found.
def custom_404_handler(request, exc): return Html( Head( Title("404 - Page Not Found"), Style(""" body { font-family: Arial, sans-serif; text-align: center; padding: 50px; } h1 { color: #e74c3c; } a { color: #3498db; text-decoration: none; } """) ), Body( H1("404 - Page Not Found"), P(f"Sorry, the page '{request.url.path}' you're looking for doesn't exist."), A("Go back to home", href="/") ) )
Integration:
app.add_exception_handler(404, custom_404_handler)
Plugging in New Applications
One of the application’s strengths lies in its modularity, allowing developers to add new functionalities seamlessly. This section provides a detailed guide on how to integrate new applications, leveraging existing reusable components like TodoApp
.
Prerequisites
Before adding a new application, ensure you have:
- Familiarity with Python: Basic understanding of Python programming.
- Understanding of Web Frameworks: Knowledge of web concepts and frameworks like Starlette or FastAPI.
- Database Schema Design: Ability to design database tables relevant to your new app.
- Knowledge of HTMX and JavaScript: For front-end interactivity and dynamic content updates.
Step-by-Step Guide
1. Define the Database Table
Start by defining the database table structure for your new application. This involves specifying the fields, data types, and primary keys.
Example: Adding a NotesApp
notes = Note # Assuming Note is defined in fast_app
# Define the Note table structure within fast_app
app, rt, (store, Store), (tasks, Task), (customers, Customer), (notes, Note) = fast_app(
"data/data.db",
# ... existing parameters
note={
"id": int,
"title": str,
"content": str,
"done": bool,
"priority": int,
"profile_id": int,
"pk": "id"
},
)
2. Create a Subclass of BaseApp
Create a new class that inherits from BaseApp
, configuring it with appropriate fields and implementing necessary methods.
Example: NotesApp
Implementation
class NotesApp(BaseApp):
def __init__(self, table):
super().__init__(
name=table.name,
table=table,
toggle_field='done',
sort_field='priority'
)
def render_item(self, note):
return render_note(note)
def prepare_insert_data(self, form):
title = form.get('title', '').strip()
content = form.get('content', '').strip()
if not title or not content:
return '' # Return empty string to prevent insertion
current_profile_id = db.get("last_profile_id", 1)
max_priority = max((n.priority or 0 for n in self.table()), default=-1) + 1
return {
"title": title,
"content": content,
"done": False,
"priority": max_priority,
"profile_id": current_profile_id,
}
def prepare_update_data(self, form):
title = form.get('title', '').strip()
content = form.get('content', '').strip()
if not title or not content:
return '' # Return empty string to prevent update
return {
"title": title,
"content": content,
"done": form.get('done', '').lower() == 'true',
}
3. Implement Rendering Functions
Define how individual items are rendered into HTML or other formats for display.
Example: render_note
Function
def render_note(note):
"""
Render a note item as an HTML list item with interactive elements.
Args:
note: The note object containing attributes like id, title, content, done, and priority.
Returns:
Li: An HTML list item representing the interactive note component.
"""
# Implement rendering logic similar to render_todo or render_profile
# Include elements like checkbox, delete button, title link, and update form
pass # Replace with actual implementation
4. Register the New App’s Routes
Instantiate the new app and register its routes using the routing function rt
.
Example: Registering NotesApp
notes_app = NotesApp(table=notes)
notes_app.register_routes(rt)
# Add the new app to the navigation menu items
MENU_ITEMS.append(notes_app.name)
5. Update the Main Routing Logic
Ensure the main route handler can recognize and render the new app’s content.
Example: Updating get
Function
@rt('/')
@rt(f'/{todo_app.name}')
@rt(f'/{profile_app.name}')
@rt(f'/{notes_app.name}') # Add this line
def get(request):
# Existing logic
if menux == profile_app.name:
response = get_profiles_content()
elif menux == notes_app.name:
response = get_notes_content() # Implement this function
else:
# Existing logic for Todo and other apps
pass
# ...
6. Implement Content Retrieval Functions
Define how the main route handler retrieves and displays content for the new app.
Example: get_notes_content
Function
def get_notes_content():
"""
Retrieve and display the list of notes.
Returns:
Container: An HTML container with the notes list and chat interface.
"""
# Implement logic similar to get_profiles_content
pass # Replace with actual implementation
7. Test the New App
After integrating the new app, thoroughly test its functionality:
- CRUD Operations: Ensure creating, reading, updating, and deleting items works as expected.
- UI Elements: Verify that interactive elements like buttons and forms behave correctly.
- Routing: Confirm that navigating to the new app’s endpoints renders the appropriate content.
- Database Interactions: Check that data is correctly stored, retrieved, and manipulated in the database.
Naming Conventions and Table Management
Proper naming conventions are crucial to prevent conflicts and ensure seamless integration of new apps.
Guidelines:
- Unique Names: Ensure each app has a unique name that doesn’t clash with existing components.
- Table Naming: Tables returned from
fast_app
should follow a consistent naming pattern, typically the plural form of the app name (e.g.,tasks
forTodoApp
). - Primary Keys: Define clear primary keys (
pk
) for each table to facilitate CRUD operations.
Example:
# For NotesApp
notes = Note
Avoiding Configuration Collisions
When integrating new apps, it’s essential to manage configuration internals to prevent collisions.
Strategies:
- Namespace Isolation: Encapsulate app-specific configurations within their respective classes or modules.
- Slot Allocation: Allocate dedicated slots or sections for new apps’ configurations, ensuring they don’t interfere with the main app’s settings.
- Consistent Prefixes: Use consistent prefixes or suffixes for configuration keys related to specific apps.
Example:
# When initializing DictLikeDB or other configurations
db = DictLikeDB(store, Store)
Ensure that new apps use their own keys and don’t overwrite or interfere with existing configurations.
Detailed Explanation of fast_app
Return Values
Understanding the return values from fast_app
is pivotal for effectively integrating new applications and managing database interactions.
fast_app
Return Structure:
app, rt, (store, Store), (tasks, Task), (customers, Customer) = fast_app(
# parameters
)
Breakdown:
app
: The main application instance that handles routing, middleware, and server configurations.rt
: A routing function used to register new endpoints within the application.(store, Store)
: Tuple representing thestore
table and its corresponding model or schema (Store
).(tasks, Task)
: Tuple for thetasks
table and its model (Task
).(customers, Customer)
: Tuple for thecustomers
table and its model (Customer
).
Adding New Tables:
When you introduce a new app with its own database table, fast_app
should be updated to include the new table and its schema. This ensures that the application recognizes and correctly manages the new data entity.
Example: Adding notes
Table for NotesApp
app, rt, (store, Store), (tasks, Task), (customers, Customer), (notes, Note) = fast_app(
"data/data.db",
# ... existing parameters
note={
"id": int,
"title": str,
"content": str,
"done": bool,
"priority": int,
"profile_id": int,
"pk": "id"
},
)
Cascade Effect:
Each tuple returned by fast_app
corresponds to a specific table and its model. These tables are used across the application for CRUD operations, route handling, and data rendering. Properly naming and managing these tables ensures that each app can operate independently without interfering with others.
Usage in Plugin Apps:
Plugin applications like TodoApp
and ProfileApp
receive their respective table handles (tasks
, customers
) during initialization. This allows them to perform database operations seamlessly within their own scope.
Example:
todo_app = TodoApp(table=tasks)
profile_app = ProfileApp(table=customers)
Examples
To solidify your understanding, let’s walk through an example of adding a new application to the system.
Adding a New NotesApp
Objective: Integrate a NotesApp
that allows users to create, read, update, and delete personal notes.
1. Define the notes
Table in fast_app
Extend the fast_app
initialization to include the notes
table.
app, rt, (store, Store), (tasks, Task), (customers, Customer), (notes, Note) = fast_app(
"data/data.db",
ws_hdr=True,
live=True,
hdrs=(
SortableJSWithUpdate('.sortable'),
Script(type='module')
),
store={
"key": str,
"value": str,
"pk": "key"
},
task={
"id": int,
"title": str,
"done": bool,
"priority": int,
"profile_id": int,
"pk": "id"
},
customer={
"id": int,
"name": str,
"address": str,
"code": str,
"active": bool,
"priority": int,
"pk": "id"
},
note={ # New table definition
"id": int,
"title": str,
"content": str,
"done": bool,
"priority": int,
"profile_id": int,
"pk": "id"
},
)
2. Create the NotesApp
Class
Implement the NotesApp
by subclassing BaseApp
.
class NotesApp(BaseApp):
def __init__(self, table):
super().__init__(
name=table.name, # 'notes'
table=table,
toggle_field='done',
sort_field='priority'
)
def render_item(self, note):
return render_note(note) # Implement this function as shown earlier
def prepare_insert_data(self, form):
title = form.get('title', '').strip()
content = form.get('content', '').strip()
if not title or not content:
return '' # Prevent insertion if data is incomplete
current_profile_id = db.get("last_profile_id", 1)
max_priority = max((n.priority or 0 for n in self.table()), default=-1) + 1
return {
"title": title,
"content": content,
"done": False,
"priority": max_priority,
"profile_id": current_profile_id,
}
def prepare_update_data(self, form):
title = form.get('title', '').strip()
content = form.get('content', '').strip()
if not title or not content:
return '' # Prevent update if data is incomplete
return {
"title": title,
"content": content,
"done": form.get('done', '').lower() == 'true',
}
3. Implement the render_note
Function
Define how individual notes are displayed in the UI.
def render_note(note):
"""
Render a note item as an HTML list item with interactive elements.
Args:
note: The note object containing attributes like id, title, content, done, and priority.
Returns:
Li: An HTML list item representing the interactive note component.
"""
delete_url = notes_app.get_action_url('delete', note.id)
toggle_url = notes_app.get_action_url('toggle', note.id)
nid = f'note-{note.id}' # Unique ID for the note item
checkbox = Input(
type="checkbox",
name="done" if note.done else None,
checked=note.done,
hx_post=toggle_url,
hx_swap="outerHTML",
hx_target=f"#{nid}",
)
delete = A(
'🗑',
hx_delete=delete_url,
hx_swap='outerHTML',
hx_target=f"#{nid}",
style="cursor: pointer; display: inline;",
cls="delete-icon"
)
title_link = A(
note.title,
href="#",
cls="note-title",
style="text-decoration: none; color: inherit;",
onclick=(
"let updateForm = this.nextElementSibling; "
"let checkbox = this.parentNode.querySelector('input[type=checkbox]'); "
"let deleteIcon = this.parentNode.querySelector('.delete-icon'); "
"if (updateForm.style.visibility === 'hidden' || updateForm.style.visibility === '') { "
" updateForm.style.visibility = 'visible'; "
" updateForm.style.height = 'auto'; "
" checkbox.style.display = 'none'; "
" deleteIcon.style.display = 'none'; "
" this.remove(); "
" const inputField = document.getElementById('note_title_" + str(note.id) + "'); "
" inputField.focus(); "
" inputField.setSelectionRange(inputField.value.length, inputField.value.length); "
"} else { "
" updateForm.style.visibility = 'hidden'; "
" updateForm.style.height = '0'; "
" checkbox.style.display = 'inline'; "
" deleteIcon.style.display = 'inline'; "
" this.style.visibility = 'visible'; "
"}"
)
)
update_form = Form(
Div(
Input(
type="text",
id=f"note_title_{note.id}",
value=note.title,
name="title",
style="flex: 1; padding-right: 10px; margin-bottom: 0px;"
),
Input(
type="text",
id=f"note_content_{note.id}",
value=note.content,
name="content",
style="flex: 2; padding-right: 10px; margin-bottom: 0px;"
),
style="display: flex; align-items: center;"
),
style="visibility: hidden; height: 0; overflow: hidden;",
hx_post=f"/{notes_app.name}/{note.id}",
hx_target=f"#{nid}",
hx_swap="outerHTML",
)
return Li(
delete,
checkbox,
title_link,
update_form,
id=nid,
cls='done' if note.done else '',
style="list-style-type: none;",
data_id=note.id,
data_priority=note.priority
)
4. Instantiate and Register NotesApp
notes_app = NotesApp(table=notes)
notes_app.register_routes(rt)
# Add 'notes' to the navigation menu
MENU_ITEMS.append(notes_app.name)
5. Update the Main Route Handler
Modify the get
function to handle requests for the new NotesApp
.
@rt('/')
@rt(f'/{todo_app.name}')
@rt(f'/{profile_app.name}')
@rt(f'/{notes_app.name}') # Register the new app's route
def get(request):
path = request.url.path.strip('/')
logger.debug(f"Received request for path: {path}")
menux = "home" if not path else path
fig(f"app: {menux}")
logger.info(f"Selected explore item: {menux}")
db["last_app_choice"] = menux
db["last_visited_url"] = request.url.path
current_profile_id = db.get("last_profile_id")
if current_profile_id:
logger.debug(f"Current profile ID: {current_profile_id}")
todos.xtra(profile_id=current_profile_id)
else:
logger.warning("No current profile ID found. Using default filtering.")
todos.xtra(profile_id=None)
if menux == profile_app.name:
response = get_profiles_content()
elif menux == notes_app.name:
response = get_notes_content() # Implement this function similar to get_profiles_content
else:
# Existing TodoApp view handling
is_todo_view = (menux == todo_app.name)
todo_items = sorted(todos(), key=lambda x: x.priority)
logger.info(f"Fetched {len(todo_items)} todo items for profile ID {current_profile_id}.")
response = Container(
nav_group,
Grid(
Div(
Card(
H2(f"{pluralize(menux, singular=True)} {LIST_SUFFIX}"),
Ul(*[render_todo(todo) for todo in todo_items],
id='todo-list',
cls='sortable',
style="padding-left: 0;"),
header=Form(
Group(
Input(
placeholder=f'Add new {todo_app.name.capitalize()}',
id='title',
name='title',
autofocus=True,
),
Button("Add", type="submit"),
),
hx_post=f"/{todo_app.name}",
hx_swap="beforeend",
hx_target="#todo-list",
),
) if is_todo_view else Card(
H2(f"{pluralize(menux, singular=True)}"),
P("This is a placeholder for the selected application."),
),
id="content-container",
),
Div(
Card(
H2(f"{APP_NAME} Chatbot"),
Div(
id='msg-list',
cls='overflow-auto',
style='height: 40vh;',
),
footer=Form(
mk_chat_input_group(),
),
),
),
cls="grid",
style=(
"display: grid; "
"gap: 20px; "
f"grid-template-columns: {GRID_LAYOUT}; "
),
),
Div(
A(
f"Poke {APP_NAME} Chatbot",
hx_post="/poke",
hx_target="#msg-list",
hx_swap="innerHTML",
cls="button",
),
style=(
"bottom: 20px; "
"position: fixed; "
"right: 20px; "
"z-index: 1000; "
),
),
Script("""
document.addEventListener('htmx:afterSwap', function(event) {
if (event.target.id === 'todo-list' && event.detail.successful) {
const form = document.querySelector('form[hx-target="#todo-list"]');
if (form) {
form.reset();
}
}
});
""")
)
logger.debug("Returning response for main GET request.")
last_profile_name = get_profile_name()
return Titled(
f"{APP_NAME} / {pluralize(last_profile_name, singular=True)} / {pluralize(menux, singular=True)}",
response,
hx_ext='ws',
ws_connect='/ws',
data_theme="dark",
)
6. Implement get_notes_content
Function
Define how the notes content is retrieved and displayed.
def get_notes_content():
"""
Retrieve and display the list of notes.
Returns:
Container: An HTML container with the notes list and chat interface.
"""
logger.debug(f"Retrieving {notes_app.name.lower()} for display.")
nav_group = create_nav_menu()
all_notes = notes()
logger.debug("Initial note state:")
for note in all_notes:
logger.debug(f"Note {note.id}: title = {note.title}, priority = {note.priority}")
ordered_notes = sorted(all_notes, key=lambda p: p.priority if p.priority is not None else float('inf'))
logger.debug("Ordered note list:")
for note in ordered_notes:
logger.debug(f"Note {note.id}: title = {note.title}, priority = {note.priority}")
return Container(
nav_group,
Grid(
Div(
Card(
H2(f"{notes_app.name.capitalize()} {LIST_SUFFIX}"),
Ul(*[render_note(note) for note in ordered_notes],
id='note-list',
cls='sortable',
style="padding-left: 0;"),
footer=Form(
Group(
Input(placeholder=f"{notes_app.name.capitalize()} Title", name="title", id="note-title-input"),
Input(placeholder="Content", name="content", id="note-content-input"),
Button("Add", type="submit", id="add-note-button"),
),
hx_post=f"/{notes_app.name}",
hx_target="#note-list",
hx_swap="beforeend",
hx_swap_oob="true",
),
),
id="content-container",
),
Div(
Card(
H2(f"{APP_NAME} Chatbot"),
Div(
id='msg-list',
cls='overflow-auto',
style='height: 40vh;',
),
footer=Form(
mk_chat_input_group(),
),
),
),
cls="grid",
style=(
"display: grid; "
"gap: 20px; "
f"grid-template-columns: {GRID_LAYOUT}; "
),
),
Script("""
document.addEventListener('htmx:afterSwap', function(event) {
if (event.target.id === 'note-list' && event.detail.successful) {
const form = document.getElementById('add-note-button').closest('form');
form.reset();
}
});
""")
)
7. Final Integration Steps
Ensure that all components are correctly integrated and the server recognizes the new app.
# Add the new app's route
for item in MENU_ITEMS:
app.add_route(f'/{item}', get)
8. Test the NotesApp
- Create Notes: Add new notes and verify they appear correctly.
- Edit Notes: Update existing notes and check for proper rendering.
- Delete Notes: Remove notes and ensure they are deleted from the UI and database.
- Toggle Completion: Mark notes as done/undone and verify the state changes.
- Sorting: Drag and drop notes to reorder and confirm priority updates.
Conclusion
This documentation provides a thorough exploration of your Python web application’s architecture, components, and extension methodologies. By following the outlined steps and adhering to the established conventions, developers can efficiently integrate new functionalities, ensuring scalability and maintainability. The emphasis on modularity, reusable components, and clear naming conventions facilitates a robust and flexible system poised for future enhancements.
Whether you’re adding simple list-based apps like TodoApp
or more complex modules, the structured approach detailed here ensures seamless integration and consistent behavior across the entire application.
For further assistance or queries, refer to the in-line comments within the codebase or reach out to the development community associated with this project.
Next article
Comprehensive Documentation for Integrating a Custom Single Page Application (SPA) Plugin
Table of Contents
- Introduction
- System Overview
- Architecture
- Components
- Plugging in the Reporting SPA
- Detailed Explanation of
fast_app
Return Values - Examples
- Conclusion
Introduction
Welcome to the advanced documentation for your Python web application! Building upon the foundational CRUD (Create, Read, Update, Delete) applications like TodoApp
and ProfileApp
, this guide delves into integrating a sophisticated Single Page Application (SPA) plugin named CustomerTasksReportApp
. This plugin leverages existing data sources to generate insightful reports, facilitating enhanced data visualization and user interactions without necessitating additional CRUD operations.
Whether you’re a seasoned developer or a newcomer aiming to extend the application’s capabilities, this documentation provides a thorough walkthrough to seamlessly incorporate new reporting functionalities into your system.
System Overview
Your application is a highly modular and extensible web platform developed in Python, utilizing a suite of robust libraries and frameworks. It is architected to support both backend CRUD operations and frontend reporting and visualization through plugins.
Key Features:
- Modular Architecture: Facilitates the addition of new functionalities without disrupting existing components.
- Database Integration: Manages persistent storage with a custom dictionary-like wrapper.
- Real-time Chatbot: Enhances user experience with AI-driven interactions.
- WebSockets: Supports real-time, bidirectional communication for dynamic features.
- Dynamic Routing: Enables flexible URL handling with support for both HTTP and WebSocket routes.
- Enhanced UI/UX: Provides interactive elements powered by HTMX and dynamic JavaScript components.
Architecture
The application employs a layered architecture, segregating concerns into distinct components for better maintainability and scalability.
Core Components:
- Fast App Initialization (
fast_app
): Bootstraps the core application, setting up database connections, routing, and middleware. - BaseApp Class: Serves as an abstract foundation for creating new application components with common CRUD operations.
- CRUD Plugin Applications (
TodoApp
andProfileApp
): Handle specific data entities and their respective CRUD operations. - Reporting SPA Plugin (
CustomerTasksReportApp
): Generates reports by aggregating data from existing CRUD apps. - Database Wrapper (
DictLikeDB
): Abstracts database interactions, offering a dictionary-like interface for data manipulation. - Chatbot Integration: Connects to a local AI model (Ollama) for real-time chatbot responses.
- WebSocket Handling: Manages real-time communication channels for interactive features.
- Middleware and Exception Handling: Enhances request processing and error management.
Conceptual Distinction:
- CRUD Apps: Serve as the backend, managing data entities and providing CRUD operations.
- Reporting and Process Apps: Utilize data from CRUD apps to generate reports, perform data visualizations, or execute complex processes.
Components
Existing CRUD Applications (TodoApp
and ProfileApp
)
These applications manage core data entities—tasks and customer profiles—providing interfaces for creating, reading, updating, and deleting records. They serve as the backbone for data manipulation within the system.
Key Features:
- Route Registration: Automatically registers endpoints for CRUD operations.
- Data Handling: Manages data through a consistent interface, ensuring data integrity and consistency.
- Rendering: Transforms data into user-friendly HTML representations for seamless UI integration.
- Error Handling: Implements robust logging and error responses to maintain system reliability.
New Reporting SPA (CustomerTasksReportApp
)
The CustomerTasksReportApp
is a Single Page Application (SPA) designed to generate comprehensive reports by aggregating data from TodoApp
and ProfileApp
. Unlike CRUD apps, it focuses on data visualization and user interactions that provide insights into customer tasks.
Key Features:
- Data Aggregation: Pulls data from existing
tasks
andcustomers
tables to generate meaningful reports. - Interactive UI: Offers dynamic interactions such as buttons triggering long-running events.
- Real-time Feedback: Integrates with the chatbot to provide humorous and engaging user experiences during processing.
- File Generation Simulation: Mimics file generation and provides links for users to access generated reports.
Database Wrapper (DictLikeDB
)
The DictLikeDB
class offers a dictionary-like interface for interacting with the database, abstracting the complexities of direct database operations.
Features:
- Persistence: Ensures data remains intact across application restarts.
- Intuitive API: Supports standard dictionary operations for ease of use.
- Adaptability: Compatible with various storage backends, promoting flexibility.
- Logging: Integrates with
loguru
for detailed operational logs.
Chatbot Integration
Integrates a local AI model (Ollama) to facilitate real-time chatbot interactions, enhancing user engagement and providing dynamic responses based on user inputs.
Key Components:
- Model Selection: Dynamically selects the best available LLaMA model.
- Response Generation: Processes user messages and generates contextual responses.
- Streaming Responses: Simulates typing effects to create a natural conversational flow.
WebSocket Handling
Manages real-time, bidirectional communication between the server and clients, enabling features like live chat and real-time updates.
Key Features:
- Connection Management: Tracks active WebSocket connections for targeted message delivery.
- Message Processing: Handles incoming messages and dispatches responses efficiently.
- Real-time Updates: Facilitates instantaneous data synchronization across clients.
Middleware and Exception Handling
Enhances the application’s robustness by managing request processing and gracefully handling errors.
Key Components:
- Custom Middleware: Implements logging and state display functionalities.
- Exception Handlers: Provides user-friendly error pages and logs critical issues for debugging.
Plugging in the Reporting SPA
Integrating a new SPA like CustomerTasksReportApp
involves several steps, leveraging the existing infrastructure of CRUD apps and the modular architecture. This section provides a detailed guide to seamlessly incorporate the reporting SPA into your application.
Prerequisites
Before proceeding, ensure you have:
- Python Proficiency: Solid understanding of Python programming.
- Web Framework Knowledge: Familiarity with web concepts and frameworks such as Starlette or FastAPI.
- Database Schema Understanding: Ability to work with existing database schemas (
tasks
andcustomers
tables). - Front-end Skills: Basic knowledge of HTML, CSS, JavaScript, and HTMX for creating interactive user interfaces.
- Asynchronous Programming: Understanding of asynchronous operations in Python (
asyncio
).
Step-by-Step Guide
1. Define the Reporting SPA Class
Create a new class CustomerTasksReportApp
that encapsulates the reporting logic. This class will handle data aggregation from TodoApp
and ProfileApp
, render the report, and manage user interactions.
Implementation:
class CustomerTasksReportApp:
"""
A Single Page Application (SPA) for reporting customers and their associated tasks.
This class leverages the existing TodoApp and ProfileApp to aggregate data and generate
comprehensive reports. It provides an interactive interface where users can trigger
long-running events, receive humorous feedback from the chatbot, and access generated reports.
"""
def __init__(self, tasks_table, customers_table, rt):
"""
Initialize the CustomerTasksReportApp.
Args:
tasks_table: The tasks table handle from TodoApp.
customers_table: The customers table handle from ProfileApp.
rt: The routing function to register endpoints.
"""
self.tasks = tasks_table
self.customers = customers_table
self.rt = rt
self.register_routes()
def register_routes(self):
"""
Register the routes specific to the CustomerTasksReportApp.
This includes the main report page and endpoints for handling long-running events.
"""
self.rt('/reports/customers', methods=['GET'])(self.get_report_page)
self.rt('/reports/customers/generate/{customer_id}', methods=['POST'])(self.generate_report)
async def get_report_page(self, request):
"""
Serve the main report page listing customers and their tasks.
Args:
request: The incoming HTTP request.
Returns:
HTML: The rendered report page.
"""
logger.debug("Serving CustomerTasksReportApp main page.")
nav = create_nav_menu()
# Fetch all active customers
active_customers = self.customers("active=?", (True,), order_by='priority')
logger.info(f"Fetched {len(active_customers)} active customers.")
# Aggregate tasks for each customer
customer_tasks = {}
for customer in active_customers:
tasks = self.tasks("profile_id=?", (customer.id,), order_by='priority')
customer_tasks[customer] = tasks
logger.debug(f"Customer {customer.name} has {len(tasks)} tasks.")
# Render the report
report_content = Container(
nav,
H2("Customer Tasks Report"),
Ul(
*[
render_customer_with_tasks(customer, tasks)
for customer, tasks in customer_tasks.items()
],
style="padding-left: 0;",
),
Div(
id="report-section",
style="margin-top: 20px;",
),
cls="container",
)
return Titled(
"Customer Tasks Report",
report_content,
data_theme="dark",
)
async def generate_report(self, request, customer_id: int):
"""
Handle the generation of a project plan for a specific customer.
This simulates a long-running process, interacts with the chatbot to provide humorous feedback,
and finally offers a link to the generated report.
Args:
request: The incoming HTTP request.
customer_id (int): The ID of the customer for whom to generate the report.
Returns:
HTML: The updated report section with the generated report link.
"""
logger.debug(f"Initiating report generation for customer ID: {customer_id}")
# Validate customer existence
try:
customer = self.customers[customer_id]
logger.info(f"Generating report for customer: {customer.name}")
except NotFoundError:
logger.error(f"Customer ID {customer_id} not found.")
return Html(
Body(
P("Customer not found."),
)
)
# Simulate long-running event
await chatq(f"Generating project plan for {customer.name}. Please wait...")
# Simulate delay
await asyncio.sleep(5) # Simulate a 5-second long task
# Interact with the chatbot for humorous feedback
await chatq("Finally! Here it is...")
# Simulate report generation and provide a link
report_link = f"/reports/generated/{customer_id}/project_plan.pdf"
# Update the UI with the generated report link
return Div(
A(
"Download Project Plan",
href=report_link,
target="_blank",
cls="button",
style="margin-top: 10px;"
)
)
Explanation:
- Initialization (
__init__
): Accepts handles to thetasks
andcustomers
tables fromTodoApp
andProfileApp
, respectively, and the routing functionrt
to register new endpoints. - Route Registration (
register_routes
): Defines two primary routes:/reports/customers
(GET): Serves the main report page./reports/customers/generate/{customer_id}
(POST): Handles report generation for a specific customer.
- Report Page Handler (
get_report_page
): Fetches active customers and their associated tasks, then renders them in a structured HTML format. - Report Generation Handler (
generate_report
): Simulates a long-running process to generate a project plan, interacts with the chatbot for humorous feedback, and provides a download link for the generated report.
2. Implement Rendering Functions
Define helper functions to render customers with their tasks and handle interactive elements like report generation buttons.
Implementation:
def render_customer_with_tasks(customer, tasks):
"""
Render a customer with their associated tasks and a report generation button.
Args:
customer: The customer object.
tasks: A list of task objects associated with the customer.
Returns:
Li: An HTML list item representing the customer and their tasks.
"""
customer_id = customer.id
customer_name = customer.name
tasks_list = Ul(
*[render_task(task) for task in tasks],
style="padding-left: 20px;",
)
# Button to generate report
generate_button = Button(
"Generate Project Plan",
type="button",
hx_post=f"/reports/customers/generate/{customer_id}",
hx_target=f"#report-section",
hx_swap="outerHTML",
cls="button",
style="margin-left: 10px;"
)
return Li(
Div(
Span(customer_name, style="font-weight: bold;"),
generate_button,
style="display: flex; align-items: center;"
),
tasks_list,
style="list-style-type: none; margin-bottom: 15px;",
)
def render_task(task):
"""
Render a single task item.
Args:
task: The task object.
Returns:
Li: An HTML list item representing the task.
"""
return Li(
f"- {task.title}",
cls='done' if task.done else '',
style="list-style-type: disc;"
)
Explanation:
render_customer_with_tasks
: Takes a customer and their tasks, rendering the customer name, a button to generate the project plan, and a nested list of tasks.render_task
: Renders individual tasks as list items, marking them as done if applicable.
3. Register the Reporting SPA Routes
Instantiate the CustomerTasksReportApp
and register its routes using the routing function rt
.
Implementation:
# Instantiate the CustomerTasksReportApp with existing tasks and customers tables
customer_tasks_report_app = CustomerTasksReportApp(
tasks_table=tasks,
customers_table=customers,
rt=rt
)
Explanation:
- Instantiation: Passes the
tasks
andcustomers
table handles to theCustomerTasksReportApp
. - Route Registration: Automatically registers the necessary routes during initialization.
4. Integrate with Existing CRUD Apps
Ensure that the reporting SPA leverages data from TodoApp
and ProfileApp
without duplicating CRUD operations.
Key Steps:
- Data Access: Use the
tasks
andcustomers
table handles provided during instantiation to fetch and aggregate data. - UI Consistency: Maintain a consistent look and feel by using existing rendering functions and styles.
- Navigation Integration: Add the reporting SPA to the main navigation menu for easy access.
Implementation:
# Add 'Customer Tasks Report' to the navigation menu
MENU_ITEMS.append('customer_tasks_report')
# Update the navigation menu creation function to include the new app
def create_nav_menu():
"""
Create the navigation menu with app, profile, and action dropdowns.
Returns:
Div: An HTML div containing the navigation menu.
"""
# Existing implementation...
# Add the Customer Tasks Report to the MENU_ITEMS list if not already present
if 'customer_tasks_report' not in MENU_ITEMS:
MENU_ITEMS.append('customer_tasks_report')
# ... Rest of the navigation menu creation logic
Explanation:
- Navigation Menu Update: Appends
'customer_tasks_report'
to theMENU_ITEMS
list, ensuring it appears in the navigation bar. - Consistency: Utilizes existing styles and structures to maintain a unified user interface.
5. Handle Long-Running Events and Chatbot Interactions
Implement the logic to simulate long-running events, interact with the chatbot for humorous feedback, and provide users with generated report links.
Implementation:
async def chatq(message: str):
"""
Queue a message for the chat stream.
Args:
message (str): The message to queue for streaming.
"""
asyncio.create_task(stream_chat(message))
logger.debug(f"Message queued for chat: {message}")
async def stream_chat(prompt: str, quick: bool = False):
"""
Generate and stream an AI response to all connected users.
Args:
prompt (str): The prompt to send to the AI model.
quick (bool): If True, send the entire response at once.
Returns:
None
"""
logger.debug(f"Streaming chat response for prompt: {prompt}")
# Generate the AI response using the selected model
response = await run_in_threadpool(
chat_with_ollama,
model,
[{"role": "user", "content": prompt}],
)
if quick:
for u in users.values():
await u(
Div(
response,
id='msg-list',
cls='fade-in',
style=MATRIX_STYLE,
)
)
else:
# Stream the response word by word
words = response.split()
for i in range(len(words)):
partial_response = " ".join(words[: i + 1])
for u in users.values():
await u(
Div(
partial_response,
id='msg-list',
cls='fade-in',
style=MATRIX_STYLE,
_=f"this.scrollIntoView({{behavior: 'smooth'}});",
)
)
await asyncio.sleep(TYPING_DELAY)
logger.debug("Completed streaming chat response.")
Explanation:
chatq
Function: Queues messages to be sent to all connected WebSocket users by creating asynchronous tasks.stream_chat
Function: Generates responses from the AI model and streams them to users. It can send the response either instantly (quick=True
) or simulate typing by sending it word by word.
Integration in Report Generation:
Within the generate_report
method of CustomerTasksReportApp
, the chatbot is invoked to provide feedback during the simulated long-running process.
async def generate_report(self, request, customer_id: int):
"""
Handle the generation of a project plan for a specific customer.
Args:
request: The incoming HTTP request.
customer_id (int): The ID of the customer for whom to generate the report.
Returns:
HTML: The updated report section with the generated report link.
"""
logger.debug(f"Initiating report generation for customer ID: {customer_id}")
# Validate customer existence
try:
customer = self.customers[customer_id]
logger.info(f"Generating report for customer: {customer.name}")
except NotFoundError:
logger.error(f"Customer ID {customer_id} not found.")
return Html(
Body(
P("Customer not found."),
)
)
# Simulate long-running event
await chatq(f"Generating project plan for {customer.name}. Please wait...")
# Simulate delay
await asyncio.sleep(5) # Simulate a 5-second long task
# Interact with the chatbot for humorous feedback
await chatq("Finally! Here it is...")
# Simulate report generation and provide a link
report_link = f"/reports/generated/{customer_id}/project_plan.pdf"
# Update the UI with the generated report link
return Div(
A(
"Download Project Plan",
href=report_link,
target="_blank",
cls="button",
style="margin-top: 10px;"
)
)
Explanation:
- Customer Validation: Ensures the customer exists before proceeding.
- Chatbot Interaction:
- Sends an initial message indicating the start of report generation.
- Waits for a simulated delay representing a long-running task.
- Sends a final message indicating completion.
- Report Link Generation: Simulates the creation of a report file and provides a download link to the user.
6. Update the Main Routing Logic
Ensure that the main route handler recognizes and appropriately delegates requests to the reporting SPA.
Implementation:
@rt('/')
@rt(f'/{todo_app.name}')
@rt(f'/{profile_app.name}')
@rt(f'/reports/customers') # Registering the main report page route
@rt(f'/reports/customers/generate/') # Registering the report generation route
def get(request):
"""
Handle main page and specific page GET requests.
This function serves as the central routing and rendering mechanism for the application.
It processes requests for the main page and specific app pages defined in MENU_ITEMS.
New apps can be easily integrated by adding them to the MENU_ITEMS list and implementing
their corresponding render and CRUD functions.
Args:
request: The incoming HTTP request.
Returns:
Titled: An HTML response with the appropriate title and content for the requested page.
"""
path = request.url.path.strip('/')
logger.debug(f"Received request for path: {path}")
menux = "home" if not path else path
fig(f"app: {menux}")
logger.info(f"Selected explore item: {menux}")
db["last_app_choice"] = menux
db["last_visited_url"] = request.url.path
# Apply the profile filter if necessary
current_profile_id = db.get("last_profile_id")
if current_profile_id:
logger.debug(f"Current profile ID: {current_profile_id}")
todos.xtra(profile_id=current_profile_id)
else:
logger.warning("No current profile ID found. Using default filtering.")
todos.xtra(profile_id=None)
if menux == profile_app.name:
response = get_profiles_content()
elif menux == 'customer_tasks_report':
response = await customer_tasks_report_app.get_report_page(request)
else:
# Existing TodoApp view handling
is_todo_view = (menux == todo_app.name)
todo_items = sorted(todos(), key=lambda x: x.priority)
logger.info(f"Fetched {len(todo_items)} todo items for profile ID {current_profile_id}.")
response = Container(
nav_group,
Grid(
Div(
Card(
H2(f"{pluralize(menux, singular=True)} {LIST_SUFFIX}"),
Ul(*[render_todo(todo) for todo in todo_items],
id='todo-list',
cls='sortable',
style="padding-left: 0;"),
header=Form(
Group(
Input(
placeholder=f'Add new {todo_app.name.capitalize()}',
id='title',
name='title',
autofocus=True,
),
Button("Add", type="submit"),
),
hx_post=f"/{todo_app.name}",
hx_swap="beforeend",
hx_target="#todo-list",
),
) if is_todo_view else Card(
H2(f"{pluralize(menux, singular=True)}"),
P("This is a placeholder for the selected application."),
),
id="content-container",
),
Div(
Card(
H2(f"{APP_NAME} Chatbot"),
Div(
id='msg-list',
cls='overflow-auto',
style='height: 40vh;',
),
footer=Form(
mk_chat_input_group(),
),
),
),
cls="grid",
style=(
"display: grid; "
"gap: 20px; "
f"grid-template-columns: {GRID_LAYOUT}; "
),
),
Div(
A(
f"Poke {APP_NAME} Chatbot",
hx_post="/poke",
hx_target="#msg-list",
hx_swap="innerHTML",
cls="button",
),
style=(
"bottom: 20px; "
"position: fixed; "
"right: 20px; "
"z-index: 1000; "
),
),
Script("""
document.addEventListener('htmx:afterSwap', function(event) {
if (event.target.id === 'todo-list' && event.detail.successful) {
const form = document.querySelector('form[hx-target="#todo-list"]');
if (form) {
form.reset();
}
}
});
""")
)
logger.debug("Returning response for main GET request.")
last_profile_name = get_profile_name()
return Titled(
f"{APP_NAME} / {pluralize(last_profile_name, singular=True)} / {pluralize(menux, singular=True)}",
response,
hx_ext='ws',
ws_connect='/ws',
data_theme="dark",
)
Explanation:
- Route Registration: Ensures that the reporting SPA’s routes (
/reports/customers
and/reports/customers/generate/{customer_id}
) are recognized and appropriately handled. - Request Delegation: Adds conditional logic to delegate requests to
CustomerTasksReportApp
when the path matches'customer_tasks_report'
. - Asynchronous Handling: Ensures that asynchronous functions are properly awaited within the route handler.
Naming Conventions and Table Management
Adhering to consistent naming conventions is crucial for maintaining clarity and avoiding conflicts within the system.
Guidelines:
- App Naming: Use descriptive and unique names for each app (e.g.,
CustomerTasksReportApp
). - Table Naming: Follow a pluralized form for table names (e.g.,
tasks
,customers
,reports
). - Primary Keys: Clearly define primary keys (
pk
) for each table to facilitate efficient CRUD operations. - Route Naming: Use clear and hierarchical route structures (e.g.,
/reports/customers
,/reports/customers/generate/{customer_id}
).
Avoiding Configuration Collisions
When integrating new plugins, it’s essential to ensure that configurations remain isolated to prevent unintended interactions.
Strategies:
- Namespace Isolation: Encapsulate app-specific configurations within their respective classes or modules.
- Dedicated Slots: Allocate unique slots or sections for new apps’ configurations to avoid overlapping with existing settings.
- Consistent Prefixes/Suffixes: Use standardized prefixes or suffixes for configuration keys related to specific apps (e.g.,
customer_tasks_report_
).
Example:
# Within CustomerTasksReportApp class
self.rt('/reports/customers', methods=['GET'])(self.get_report_page)
self.rt('/reports/customers/generate/{customer_id}', methods=['POST'])(self.generate_report)
By maintaining clear boundaries and consistent naming, you ensure that each plugin operates independently without interfering with others.
Detailed Explanation of fast_app
Return Values
Understanding the return values from the fast_app
function is pivotal for effectively integrating new applications and managing database interactions.
fast_app
Return Structure:
app, rt, (store, Store), (tasks, Task), (customers, Customer) = fast_app(
"data/data.db",
ws_hdr=True,
live=True,
hdrs=(
SortableJSWithUpdate('.sortable'),
Script(type='module')
),
store={
"key": str,
"value": str,
"pk": "key"
},
task={
"id": int,
"title": str,
"done": bool,
"priority": int,
"profile_id": int,
"pk": "id"
},
customer={
"id": int,
"name": str,
"address": str,
"code": str,
"active": bool,
"priority": int,
"pk": "id"
},
)
Breakdown:
app
: The main application instance responsible for handling routing, middleware, and server configurations.rt
: A routing function used to register new endpoints within the application.(store, Store)
: Tuple representing thestore
table and its corresponding model or schema (Store
).(tasks, Task)
: Tuple for thetasks
table and its model (Task
).(customers, Customer)
: Tuple for thecustomers
table and its model (Customer
).
Adding New Tables:
When introducing a new app with its own database table, update fast_app
to include the new table and its schema. This ensures the application recognizes and correctly manages the new data entity.
Example: Adding reports
Table for CustomerTasksReportApp
app, rt, (store, Store), (tasks, Task), (customers, Customer), (reports, Report) = fast_app(
"data/data.db",
ws_hdr=True,
live=True,
hdrs=(
SortableJSWithUpdate('.sortable'),
Script(type='module')
),
store={
"key": str,
"value": str,
"pk": "key"
},
task={
"id": int,
"title": str,
"done": bool,
"priority": int,
"profile_id": int,
"pk": "id"
},
customer={
"id": int,
"name": str,
"address": str,
"code": str,
"active": bool,
"priority": int,
"pk": "id"
},
report={
"id": int,
"customer_id": int,
"file_path": str,
"generated_at": datetime,
"pk": "id"
},
)
Cascade Effect:
Each tuple returned by fast_app
corresponds to a specific table and its model. These tables are utilized across the application for CRUD operations, route handling, and data rendering. Proper naming and management ensure each app operates independently without interference.
Usage in Plugin Apps:
Plugin applications like CustomerTasksReportApp
receive their respective table handles (tasks
, customers
, reports
) during initialization. This allows them to perform database operations seamlessly within their own scope.
Example:
customer_tasks_report_app = CustomerTasksReportApp(
tasks_table=tasks,
customers_table=customers,
rt=rt
)
Examples
To solidify your understanding, let’s walk through an example of integrating the CustomerTasksReportApp
into your system.
Integrating CustomerTasksReportApp
Objective: Integrate a reporting SPA that lists customers and their associated tasks, provides interactive buttons to generate project plans, and leverages the chatbot for humorous feedback during processing.
1. Define the reports
Table in fast_app
Extend the fast_app
initialization to include the reports
table required by CustomerTasksReportApp
.
from datetime import datetime
app, rt, (store, Store), (tasks, Task), (customers, Customer), (reports, Report) = fast_app(
"data/data.db",
ws_hdr=True,
live=True,
hdrs=(
SortableJSWithUpdate('.sortable'),
Script(type='module')
),
store={
"key": str,
"value": str,
"pk": "key"
},
task={
"id": int,
"title": str,
"done": bool,
"priority": int,
"profile_id": int,
"pk": "id"
},
customer={
"id": int,
"name": str,
"address": str,
"code": str,
"active": bool,
"priority": int,
"pk": "id"
},
report={
"id": int,
"customer_id": int,
"file_path": str,
"generated_at": datetime,
"pk": "id"
},
)
Explanation:
report
Table: Defines fields necessary for tracking generated reports, includingcustomer_id
,file_path
, andgenerated_at
.
2. Create the CustomerTasksReportApp
Class
Implement the CustomerTasksReportApp
as detailed in the Define the Reporting SPA Class section.
Implementation:
customer_tasks_report_app = CustomerTasksReportApp(
tasks_table=tasks,
customers_table=customers,
rt=rt
)
3. Implement Rendering Functions
Define how customers and their tasks are rendered within the report.
Implementation:
def render_customer_with_tasks(customer, tasks):
"""
Render a customer with their associated tasks and a report generation button.
Args:
customer: The customer object.
tasks: A list of task objects associated with the customer.
Returns:
Li: An HTML list item representing the customer and their tasks.
"""
customer_id = customer.id
customer_name = customer.name
tasks_list = Ul(
*[render_task(task) for task in tasks],
style="padding-left: 20px;",
)
# Button to generate report
generate_button = Button(
"Generate Project Plan",
type="button",
hx_post=f"/reports/customers/generate/{customer_id}",
hx_target=f"#report-section",
hx_swap="outerHTML",
cls="button",
style="margin-left: 10px;"
)
return Li(
Div(
Span(customer_name, style="font-weight: bold;"),
generate_button,
style="display: flex; align-items: center;"
),
tasks_list,
style="list-style-type: none; margin-bottom: 15px;",
)
def render_task(task):
"""
Render a single task item.
Args:
task: The task object.
Returns:
Li: An HTML list item representing the task.
"""
return Li(
f"- {task.title}",
cls='done' if task.done else '',
style="list-style-type: disc;"
)
Explanation:
render_customer_with_tasks
: Renders each customer with their tasks and includes a button to generate a project plan.render_task
: Renders individual tasks, indicating completion status.
4. Register the Reporting SPA Routes
Ensure that CustomerTasksReportApp
registers its routes during initialization.
Implementation:
customer_tasks_report_app = CustomerTasksReportApp(
tasks_table=tasks,
customers_table=customers,
rt=rt
)
Explanation:
- Automatic Route Registration: The
CustomerTasksReportApp
registers its routes (/reports/customers
and/reports/customers/generate/{customer_id}
) upon instantiation.
5. Integrate with Existing CRUD Apps
Leverage data from TodoApp
and ProfileApp
to generate reports without duplicating CRUD operations.
Implementation:
- Data Access: Utilize the
tasks
andcustomers
table handles to fetch and aggregate data withinCustomerTasksReportApp
. - Navigation Integration: Add the reporting SPA to the main navigation menu for user accessibility.
# Add 'Customer Tasks Report' to the navigation menu
MENU_ITEMS.append('customer_tasks_report')
Explanation:
- Seamless Integration: By appending to
MENU_ITEMS
, the reporting SPA becomes accessible via the main navigation, maintaining UI consistency.
6. Handle Long-Running Events and Chatbot Interactions
Implement the logic to simulate long-running events, interact with the chatbot for humorous feedback, and provide users with generated report links.
Implementation:
As detailed in the Handle Long-Running Events and Chatbot Interactions section, the generate_report
method within CustomerTasksReportApp
manages these interactions.
Key Points:
- Simulated Delay: Uses
asyncio.sleep
to mimic a long-running task. - Chatbot Feedback: Sends messages to the chatbot to engage users during processing.
- Report Link Generation: Provides a simulated download link upon completion.
Conclusion
This documentation has provided a comprehensive guide to integrating a custom Single Page Application (SPA) plugin—CustomerTasksReportApp
—into your Python web application. By leveraging existing CRUD applications (TodoApp
and ProfileApp
), the reporting SPA efficiently aggregates and visualizes data, enhancing the overall functionality and user experience of your system.
Key Takeaways:
- Modular Integration: The application’s architecture supports seamless addition of new plugins without disrupting existing components.
- Data Reusability: Leveraging data from backend CRUD apps promotes efficient data management and reduces redundancy.
- Interactive Enhancements: Integrating real-time feedback through chatbots and simulating long-running events enriches user engagement.
- Scalability: The structured approach ensures that the system remains scalable, accommodating future enhancements and data visualization needs.
By following the outlined steps and adhering to established conventions, developers can efficiently extend the application’s capabilities, ensuring a robust and flexible system poised for future growth. Whether creating simple reports or integrating complex data visualizations, the application’s modular design facilitates a smooth and efficient development process.
For further assistance or queries, refer to the in-line comments within the codebase or engage with the development community associated with this project. Embrace the power of modular design and continue building innovative features that elevate your application’s utility and user experience!
Next article
Advanced Integration: Data Science Enhancements with Pandas and Bokeh
Table of Contents
- Introduction
- System Overview
- Architecture
- Components
- Plugging in the Advanced Reporting SPA
- Prerequisites
- Step-by-Step Guide
- Install Required Packages
- Integrate Pandas for Data Manipulation
- Create Bokeh DataTable
- Embed Bokeh DataTable into FastHTML
- Add PicoCSS Range Slider
- Bind Slider to Bokeh DataTable
- Provide CSV Download Link
- Update the Reporting SPA Class
- Implement Frontend Interactivity
- Test the Advanced Reporting SPA
- Naming Conventions and Table Management
- Avoiding Configuration Collisions
- Conclusion
Introduction
Building upon the foundational CRUD applications (TodoApp
and ProfileApp
) and the initial CustomerTasksReportApp
, this advanced guide introduces data science enhancements using Pandas and Bokeh. These integrations empower data scientists to perform sophisticated data manipulations and visualizations, transitioning seamlessly from Jupyter Notebooks to dynamic web applications using the FastHTML framework.
Objectives
- Data Manipulation with Pandas: Leverage Pandas for efficient data processing and analysis.
- Interactive Visualizations with Bokeh: Create dynamic data tables and visualizations.
- Frontend Interactivity with PicoCSS: Implement range sliders to filter data in real-time.
- Enhanced User Experience: Provide downloadable CSVs and interactive reports akin to Jupyter Notebook capabilities.
System Overview
Your application is a modular, extensible web platform built with Python, designed to facilitate both backend data management and frontend data visualization. By integrating data science tools like Pandas and Bokeh, the system caters to data scientists aiming to develop interactive dashboards and reports without leaving the web environment.
Key Features:
- Modular Architecture: Seamlessly extendable through plugins.
- Database Integration: Robust interaction with persistent storage via a custom wrapper.
- Data Science Enhancements: Advanced data manipulation and visualization using Pandas and Bokeh.
- Real-time Chatbot: Engaging user interactions powered by a local AI model (Ollama).
- WebSockets: Support for real-time communication and updates.
- Dynamic Routing: Flexible URL handling with support for both HTTP and WebSocket routes.
- Enhanced UI/UX: Interactive elements powered by HTMX, PicoCSS, and dynamic JavaScript components.
Architecture
The application employs a layered, modular architecture, promoting separation of concerns and facilitating scalability. Each component interacts cohesively, ensuring that backend data management and frontend visualization operate seamlessly.
Core Components:
- Fast App Initialization (
fast_app
): Bootstraps the core application, setting up database connections, routing, and middleware. - BaseApp Class: Serves as an abstract foundation for creating new application components with common CRUD operations.
- CRUD Plugin Applications (
TodoApp
andProfileApp
): Handle specific data entities and their respective CRUD operations. - Reporting SPA Plugin (
CustomerTasksReportApp
): Aggregates data from CRUD apps to generate reports. - Advanced Reporting SPA (
CustomerTasksReportApp
with Pandas and Bokeh): Enhances the reporting SPA with data science capabilities. - Database Wrapper (
DictLikeDB
): Abstracts database interactions, offering a dictionary-like interface for data manipulation. - Chatbot Integration: Connects to a local AI model (Ollama) for real-time chatbot responses.
- WebSocket Handling: Manages real-time communication channels for interactive features.
- Middleware and Exception Handling: Enhances request processing and error management.
Components
Existing CRUD Applications (TodoApp
and ProfileApp
)
These applications manage core data entities—tasks and customer profiles—providing interfaces for creating, reading, updating, and deleting records. They serve as the backbone for data manipulation within the system.
Key Features:
- Route Registration: Automatically registers endpoints for CRUD operations.
- Data Handling: Manages data through a consistent interface, ensuring data integrity and consistency.
- Rendering: Transforms data into user-friendly HTML representations for seamless UI integration.
- Error Handling: Implements robust logging and error responses to maintain system reliability.
Reporting SPA (CustomerTasksReportApp
)
The CustomerTasksReportApp
is a Single Page Application (SPA) designed to generate comprehensive reports by aggregating data from TodoApp
and ProfileApp
. It provides interactive elements that allow users to trigger report generation processes, receive real-time feedback from the chatbot, and access generated reports.
Key Features:
- Data Aggregation: Pulls data from existing
tasks
andcustomers
tables to generate meaningful reports. - Interactive UI: Offers dynamic interactions such as buttons triggering long-running events.
- Real-time Feedback: Integrates with the chatbot to provide humorous and engaging user experiences during processing.
- File Generation Simulation: Mimics file generation and provides links for users to access generated reports.
Advanced Reporting SPA (CustomerTasksReportApp
with Pandas and Bokeh)
This advanced iteration of CustomerTasksReportApp
incorporates data science tools to enhance data manipulation and visualization capabilities. By integrating Pandas for data processing and Bokeh for interactive visualizations, it enables the creation of dynamic, filterable reports that appeal to data scientists transitioning from Jupyter Notebooks to web applications.
Key Enhancements:
- Data Manipulation with Pandas: Perform sophisticated data processing tasks with ease.
- Interactive Visualizations with Bokeh: Create dynamic data tables and visualizations that respond to user inputs.
- Frontend Interactivity with PicoCSS: Implement intuitive range sliders to filter data in real-time.
- Enhanced Reporting Features: Provide downloadable CSVs and interactive visualizations for comprehensive data analysis.
Database Wrapper (DictLikeDB
)
The DictLikeDB
class offers a dictionary-like interface for interacting with the database, abstracting the complexities of direct database operations.
Features:
- Persistence: Ensures data remains intact across application restarts.
- Intuitive API: Supports standard dictionary operations for ease of use.
- Adaptability: Compatible with various storage backends, promoting flexibility.
- Logging: Integrates with
loguru
for detailed operational logs.
Chatbot Integration
Integrates a local AI model (Ollama) to facilitate real-time chatbot interactions, enhancing user engagement and providing dynamic responses based on user inputs.
Key Components:
- Model Selection: Dynamically selects the best available LLaMA model.
- Response Generation: Processes user messages and generates contextual responses.
- Streaming Responses: Simulates typing effects to create a natural conversational flow.
WebSocket Handling
Manages real-time, bidirectional communication between the server and clients, enabling features like live chat and real-time updates.
Key Features:
- Connection Management: Tracks active WebSocket connections for targeted message delivery.
- Message Processing: Handles incoming messages and dispatches responses efficiently.
- Real-time Updates: Facilitates instantaneous data synchronization across clients.
Middleware and Exception Handling
Enhances the application’s robustness by managing request processing and gracefully handling errors.
Key Components:
- Custom Middleware: Implements logging and state display functionalities.
- Exception Handlers: Provides user-friendly error pages and logs critical issues for debugging.
Plugging in the Advanced Reporting SPA
Integrating the CustomerTasksReportApp
with advanced data science capabilities involves several steps. This section provides a detailed, step-by-step guide to enhance your reporting SPA using Pandas for data manipulation and Bokeh for interactive visualizations, coupled with PicoCSS for frontend interactivity.
Prerequisites
Before proceeding, ensure you have:
- Python Proficiency: Solid understanding of Python programming.
- Web Framework Knowledge: Familiarity with web concepts and frameworks such as Starlette or FastAPI.
- Database Schema Understanding: Ability to work with existing database schemas (
tasks
,customers
, andreports
tables). - Front-end Skills: Basic knowledge of HTML, CSS, JavaScript, and HTMX for creating interactive user interfaces.
- Data Science Tools: Familiarity with Pandas for data manipulation and Bokeh for data visualization.
- Asynchronous Programming: Understanding of asynchronous operations in Python (
asyncio
). - Additional Libraries: Knowledge of PicoCSS for styling and Bokeh integration with web frameworks.
Step-by-Step Guide
1. Install Required Packages
To integrate Pandas and Bokeh, install the necessary Python packages.
Installation Command:
pip install pandas bokeh
Explanation:
- Pandas: A powerful data manipulation and analysis library.
- Bokeh: An interactive visualization library that targets modern web browsers for presentation.
2. Integrate Pandas for Data Manipulation
Leverage Pandas to perform sophisticated data processing tasks within your reporting SPA.
Implementation:
Within the CustomerTasksReportApp
, fetch data from the tasks
and customers
tables and load them into Pandas DataFrames for manipulation.
Code Example:
import pandas as pd
class CustomerTasksReportApp:
# Existing __init__ and register_routes methods...
async def get_report_page(self, request):
"""
Serve the main report page listing customers and their tasks with advanced data manipulations.
"""
logger.debug("Serving advanced CustomerTasksReportApp main page.")
nav = create_nav_menu()
# Fetch all active customers
active_customers = self.customers("active=?", (True,), order_by='priority')
logger.info(f"Fetched {len(active_customers)} active customers.")
# Fetch all tasks related to active customers
customer_ids = [customer.id for customer in active_customers]
tasks = self.tasks("profile_id IN ?", (customer_ids,), order_by='priority')
logger.info(f"Fetched {len(tasks)} tasks for active customers.")
# Load data into Pandas DataFrames
customers_df = pd.DataFrame([customer.__dict__ for customer in active_customers])
tasks_df = pd.DataFrame([task.__dict__ for task in tasks])
logger.debug("Data loaded into Pandas DataFrames.")
# Merge DataFrames to associate tasks with customers
merged_df = pd.merge(customers_df, tasks_df, left_on='id', right_on='profile_id', how='left', suffixes=('_customer', '_task'))
logger.debug("DataFrames merged for report generation.")
# Perform any additional data manipulations here if necessary
# Convert the merged DataFrame to a CSV string for download
csv_data = merged_df.to_csv(index=False)
logger.debug("CSV data prepared for download.")
# Render the report with Bokeh DataTable and download link
report_content = Container(
nav,
H2("Customer Tasks Report"),
Div(
A(
"Download Report as CSV",
href=f"/reports/generated/{merged_df.to_csv(index=False)}", # Placeholder, will update in later steps
download="customer_tasks_report.csv",
cls="button",
style="margin-bottom: 10px;"
),
id="download-link",
),
Div(
id="bokeh-table",
style="margin-top: 20px;",
),
Div(
# Placeholder for PicoCSS Range Slider
Label("Filter by Task Priority:"),
Input(
type="range",
min=str(merged_df['priority'].min()),
max=str(merged_df['priority'].max()),
value=str(merged_df['priority'].max()),
step="1",
id="priority-slider",
cls="slider",
style="width: 100%;",
hx_post="/reports/customers/filter",
hx_trigger="change",
hx_target="#bokeh-table",
hx_swap="outerHTML",
),
style="margin-top: 10px;",
),
cls="container",
)
# Generate the initial Bokeh DataTable
bokeh_table = self.create_bokeh_table(merged_df)
# Embed the Bokeh table into the Div
report_content.children.append(bokeh_table)
return Titled(
"Customer Tasks Report",
report_content,
data_theme="dark",
)
def create_bokeh_table(self, df: pd.DataFrame):
"""
Create a Bokeh DataTable from a Pandas DataFrame.
Args:
df (pd.DataFrame): The DataFrame containing the report data.
Returns:
Div: An HTML div containing the Bokeh DataTable.
"""
from bokeh.models import ColumnDataSource
from bokeh.models.widgets import DataTable, TableColumn
from bokeh.embed import components
# Prepare the ColumnDataSource
source = ColumnDataSource(df)
# Define the columns for the DataTable
columns = [
TableColumn(field="name", title="Customer Name"),
TableColumn(field="title", title="Task Title"),
TableColumn(field="done", title="Completed"),
TableColumn(field="priority_task", title="Task Priority"),
# Add more columns as needed
]
# Create the DataTable
data_table = DataTable(source=source, columns=columns, width=800, height=280, sortable=True, selectable=True)
# Embed the DataTable using Bokeh's components
script, div = components(data_table)
# Return as an HTML Div
return Div(
script,
div,
style="border: 1px solid #ddd; padding: 10px; border-radius: 5px;",
)
Explanation:
- Data Fetching: Retrieves active customers and their associated tasks from the database.
- Pandas DataFrames: Loads fetched data into Pandas DataFrames for manipulation.
- Data Merging: Merges
customers_df
andtasks_df
to associate tasks with respective customers. - CSV Preparation: Converts the merged DataFrame into a CSV string for download.
- Bokeh DataTable: Generates an interactive data table using Bokeh, embedding it into the report page.
- PicoCSS Range Slider: Adds a range slider to filter tasks based on priority, triggering updates to the Bokeh table.
3. Create Bokeh DataTable
Utilize Bokeh to create an interactive data table that displays the report data. This table allows users to sort and select data dynamically.
Implementation:
Within the CustomerTasksReportApp
, define a method to create a Bokeh DataTable
from a Pandas DataFrame.
Code Example:
def create_bokeh_table(self, df: pd.DataFrame):
"""
Create a Bokeh DataTable from a Pandas DataFrame.
Args:
df (pd.DataFrame): The DataFrame containing the report data.
Returns:
Div: An HTML div containing the Bokeh DataTable.
"""
from bokeh.models import ColumnDataSource
from bokeh.models.widgets import DataTable, TableColumn
from bokeh.embed import components
# Prepare the ColumnDataSource
source = ColumnDataSource(df)
# Define the columns for the DataTable
columns = [
TableColumn(field="name", title="Customer Name"),
TableColumn(field="title", title="Task Title"),
TableColumn(field="done", title="Completed"),
TableColumn(field="priority_task", title="Task Priority"),
# Add more columns as needed
]
# Create the DataTable
data_table = DataTable(source=source, columns=columns, width=800, height=280, sortable=True, selectable=True)
# Embed the DataTable using Bokeh's components
script, div = components(data_table)
# Return as an HTML Div
return Div(
script,
div,
style="border: 1px solid #ddd; padding: 10px; border-radius: 5px;",
)
Explanation:
- ColumnDataSource: Bridges the Pandas DataFrame with Bokeh’s plotting mechanisms.
- TableColumn: Defines the columns to be displayed in the data table, specifying both the field name and the title.
- DataTable: Creates an interactive table with sorting and selection capabilities.
- Embedding: Uses Bokeh’s
components
to generate the necessary JavaScript and HTML to embed the table into the web page. - Styling: Wraps the table in an HTML
Div
with basic styling for visual appeal.
4. Embed Bokeh DataTable into FastHTML
Integrate the Bokeh DataTable
into the FastHTML framework to render it within the web application seamlessly.
Implementation:
The create_bokeh_table
method returns an HTML Div
containing the embedded Bokeh table, which is then included in the main report content.
Code Example:
# Within the get_report_page method
# Generate the initial Bokeh DataTable
bokeh_table = self.create_bokeh_table(merged_df)
# Embed the Bokeh table into the Div
report_content.children.append(bokeh_table)
Explanation:
- Report Content: The main report page includes a navigation menu, download link, Bokeh table, and a range slider.
- Embedding: Appends the Bokeh table
Div
to the report content, ensuring it appears below the download link.
5. Add PicoCSS Range Slider
Implement a PicoCSS range slider to allow users to filter tasks based on priority dynamically. This slider will interact with the Bokeh DataTable
to update the displayed data in real-time.
Implementation:
Add an Input
element of type range
styled with PicoCSS, and bind it to a backend endpoint that processes the slider value to filter the data table.
Code Example:
Div(
# Existing elements...
Div(
Label("Filter by Task Priority:"),
Input(
type="range",
min=str(merged_df['priority_task'].min()),
max=str(merged_df['priority_task'].max()),
value=str(merged_df['priority_task'].max()),
step="1",
id="priority-slider",
cls="slider",
style="width: 100%;",
hx_post="/reports/customers/filter",
hx_trigger="change",
hx_target="#bokeh-table",
hx_swap="outerHTML",
),
style="margin-top: 10px;",
),
)
Explanation:
- Label: Provides a descriptive label for the slider.
- Input Slider:
- Type:
range
creates a slider. - Min/Max: Set to the minimum and maximum task priority values.
- Value: Initialized to the maximum priority to show all tasks by default.
- Step: Defines the increment steps.
- ID:
priority-slider
uniquely identifies the slider. - Class:
slider
applies PicoCSS styling. - HTMX Attributes:
hx_post
: Specifies the backend endpoint to handle the slider change.hx_trigger
: Triggers on thechange
event.hx_target
: Specifies the target element (#bokeh-table
) to update.hx_swap
: Determines how the response is integrated (outerHTML
replaces the entire target element).
- Type:
- Styling: Adds top margin for spacing.
6. Bind Slider to Bokeh DataTable
Create a backend endpoint that processes the slider value, filters the data accordingly using Pandas, and updates the Bokeh DataTable
.
Implementation:
Define a new route /reports/customers/filter
that accepts the slider value, filters the data, regenerates the Bokeh table, and returns it to the frontend.
Code Example:
class CustomerTasksReportApp:
# Existing methods...
def register_routes(self):
"""
Register the routes specific to the CustomerTasksReportApp.
"""
self.rt('/reports/customers', methods=['GET'])(self.get_report_page)
self.rt('/reports/customers/generate/{customer_id}', methods=['POST'])(self.generate_report)
self.rt('/reports/customers/filter', methods=['POST'])(self.filter_table)
async def filter_table(self, request):
"""
Handle filtering of the Bokeh DataTable based on slider input.
Args:
request: The incoming HTTP request containing the slider value.
Returns:
HTML: The updated Bokeh DataTable after applying the filter.
"""
form = await request.form()
max_priority = int(form.get('priority-slider', self.tasks().priority.max()))
logger.debug(f"Filtering tasks with priority <= {max_priority}")
# Fetch all active customers
active_customers = self.customers("active=?", (True,), order_by='priority')
customer_ids = [customer.id for customer in active_customers]
# Fetch tasks with priority <= max_priority
tasks = self.tasks("profile_id IN ? AND priority <= ?", (customer_ids, max_priority), order_by='priority')
logger.info(f"Fetched {len(tasks)} tasks after applying priority filter.")
# Load data into Pandas DataFrames
customers_df = pd.DataFrame([customer.__dict__ for customer in active_customers])
tasks_df = pd.DataFrame([task.__dict__ for task in tasks])
# Merge DataFrames to associate tasks with customers
merged_df = pd.merge(customers_df, tasks_df, left_on='id', right_on='profile_id', how='left', suffixes=('_customer', '_task'))
# Perform any additional data manipulations here if necessary
# Generate the filtered Bokeh DataTable
bokeh_table = self.create_bokeh_table(merged_df)
return bokeh_table
Explanation:
- Route Registration: Adds a new POST route
/reports/customers/filter
to handle filter requests. filter_table
Method:- Form Data: Retrieves the
priority-slider
value from the POST request. - Data Fetching: Retrieves active customers and their associated tasks with priority less than or equal to the slider value.
- Pandas DataFrames: Loads the filtered data into Pandas DataFrames for manipulation.
- Data Merging: Merges
customers_df
andtasks_df
to associate tasks with respective customers. - Bokeh DataTable: Generates a new Bokeh
DataTable
based on the filtered data. - Return: Sends the updated Bokeh table to the frontend to replace the existing table.
- Form Data: Retrieves the
7. Provide CSV Download Link
Enhance the reporting SPA by allowing users to download the filtered report as a CSV file. This feature complements the interactive Bokeh DataTable
, providing flexibility for offline analysis.
Implementation:
Update the download link to dynamically generate and serve the CSV data based on the current filter.
Code Example:
Modify the download link within the get_report_page
method to point to a new endpoint that serves the CSV.
Div(
A(
"Download Report as CSV",
href=f"/reports/customers/download",
cls="button",
style="margin-bottom: 10px;"
),
id="download-link",
)
Register the Download Route:
class CustomerTasksReportApp:
# Existing methods...
def register_routes(self):
"""
Register the routes specific to the CustomerTasksReportApp.
"""
self.rt('/reports/customers', methods=['GET'])(self.get_report_page)
self.rt('/reports/customers/generate/{customer_id}', methods=['POST'])(self.generate_report)
self.rt('/reports/customers/filter', methods=['POST'])(self.filter_table)
self.rt('/reports/customers/download', methods=['GET'])(self.download_csv)
async def download_csv(self, request):
"""
Serve the filtered report as a downloadable CSV file.
Args:
request: The incoming HTTP request.
Returns:
StreamingResponse: The CSV file as a downloadable response.
"""
logger.debug("Preparing CSV download for Customer Tasks Report.")
# Fetch the current filter from the database or session
max_priority = db.get("last_max_priority", self.tasks().priority.max())
# Fetch all active customers
active_customers = self.customers("active=?", (True,), order_by='priority')
customer_ids = [customer.id for customer in active_customers]
# Fetch tasks with priority <= max_priority
tasks = self.tasks("profile_id IN ? AND priority <= ?", (customer_ids, max_priority), order_by='priority')
logger.info(f"Fetched {len(tasks)} tasks for CSV download.")
# Load data into Pandas DataFrames
customers_df = pd.DataFrame([customer.__dict__ for customer in active_customers])
tasks_df = pd.DataFrame([task.__dict__ for task in tasks])
# Merge DataFrames to associate tasks with customers
merged_df = pd.merge(customers_df, tasks_df, left_on='id', right_on='profile_id', how='left', suffixes=('_customer', '_task'))
# Convert the merged DataFrame to CSV
csv_data = merged_df.to_csv(index=False)
# Update the last_max_priority in the database
db["last_max_priority"] = max_priority
# Serve the CSV as a downloadable response
from starlette.responses import StreamingResponse
from io import StringIO
buffer = StringIO()
buffer.write(csv_data)
buffer.seek(0)
return StreamingResponse(
buffer,
media_type='text/csv',
headers={
'Content-Disposition': f'attachment; filename="customer_tasks_report.csv"'
}
)
Explanation:
- Download Link: Points to
/reports/customers/download
, initiating a GET request to download the CSV. download_csv
Method:- Filter Retrieval: Obtains the current
max_priority
from the database to apply the same filter as the displayed table. - Data Fetching: Retrieves active customers and their associated tasks based on the filter.
- Pandas DataFrames: Loads and merges data into a single DataFrame.
- CSV Conversion: Converts the DataFrame to a CSV string.
- StreamingResponse: Serves the CSV data as a downloadable file with appropriate headers.
- Filter Retrieval: Obtains the current
8. Update the Reporting SPA Class
Integrate the newly implemented methods and ensure that the reporting SPA class (CustomerTasksReportApp
) incorporates all functionalities, including data manipulation, visualization, filtering, and file downloading.
Implementation:
Consolidate all methods within the CustomerTasksReportApp
class, ensuring seamless interaction between components.
Code Example:
import pandas as pd
from bokeh.models import ColumnDataSource
from bokeh.models.widgets import DataTable, TableColumn
from bokeh.embed import components
from starlette.responses import StreamingResponse, HTMLResponse
class CustomerTasksReportApp:
"""
A Single Page Application (SPA) for reporting customers and their associated tasks with data science enhancements.
"""
def __init__(self, tasks_table, customers_table, rt):
"""
Initialize the CustomerTasksReportApp.
Args:
tasks_table: The tasks table handle from TodoApp.
customers_table: The customers table handle from ProfileApp.
rt: The routing function to register endpoints.
"""
self.tasks = tasks_table
self.customers = customers_table
self.rt = rt
self.register_routes()
def register_routes(self):
"""
Register the routes specific to the CustomerTasksReportApp.
"""
self.rt('/reports/customers', methods=['GET'])(self.get_report_page)
self.rt('/reports/customers/generate/{customer_id}', methods=['POST'])(self.generate_report)
self.rt('/reports/customers/filter', methods=['POST'])(self.filter_table)
self.rt('/reports/customers/download', methods=['GET'])(self.download_csv)
async def get_report_page(self, request):
"""
Serve the main report page listing customers and their tasks with advanced data manipulations.
"""
logger.debug("Serving advanced CustomerTasksReportApp main page.")
nav = create_nav_menu()
# Fetch all active customers
active_customers = self.customers("active=?", (True,), order_by='priority')
logger.info(f"Fetched {len(active_customers)} active customers.")
# Fetch all tasks related to active customers
customer_ids = [customer.id for customer in active_customers]
tasks = self.tasks("profile_id IN ?", (customer_ids,), order_by='priority')
logger.info(f"Fetched {len(tasks)} tasks for active customers.")
# Load data into Pandas DataFrames
customers_df = pd.DataFrame([customer.__dict__ for customer in active_customers])
tasks_df = pd.DataFrame([task.__dict__ for task in tasks])
logger.debug("Data loaded into Pandas DataFrames.")
# Merge DataFrames to associate tasks with customers
merged_df = pd.merge(customers_df, tasks_df, left_on='id', right_on='profile_id', how='left', suffixes=('_customer', '_task'))
logger.debug("DataFrames merged for report generation.")
# Convert the merged DataFrame to a CSV string for download
csv_data = merged_df.to_csv(index=False)
logger.debug("CSV data prepared for download.")
# Store the current max_priority in the database for download consistency
max_priority = merged_df['priority_task'].max()
db["last_max_priority"] = max_priority
# Render the report with Bokeh DataTable and download link
report_content = Container(
nav,
H2("Customer Tasks Report"),
Div(
A(
"Download Report as CSV",
href=f"/reports/customers/download",
cls="button",
style="margin-bottom: 10px;"
),
id="download-link",
),
Div(
id="bokeh-table",
style="margin-top: 20px;",
),
Div(
Label("Filter by Task Priority:"),
Input(
type="range",
min=str(merged_df['priority_task'].min()),
max=str(merged_df['priority_task'].max()),
value=str(merged_df['priority_task'].max()),
step="1",
id="priority-slider",
cls="slider",
style="width: 100%;",
hx_post="/reports/customers/filter",
hx_trigger="change",
hx_target="#bokeh-table",
hx_swap="outerHTML",
),
style="margin-top: 10px;",
),
cls="container",
)
# Generate the initial Bokeh DataTable
bokeh_table = self.create_bokeh_table(merged_df)
# Embed the Bokeh table into the Div
report_content.children.append(bokeh_table)
return Titled(
"Customer Tasks Report",
report_content,
data_theme="dark",
)
def create_bokeh_table(self, df: pd.DataFrame):
"""
Create a Bokeh DataTable from a Pandas DataFrame.
Args:
df (pd.DataFrame): The DataFrame containing the report data.
Returns:
Div: An HTML div containing the Bokeh DataTable.
"""
from bokeh.models import ColumnDataSource
from bokeh.models.widgets import DataTable, TableColumn
from bokeh.embed import components
# Prepare the ColumnDataSource
source = ColumnDataSource(df)
# Define the columns for the DataTable
columns = [
TableColumn(field="name", title="Customer Name"),
TableColumn(field="title", title="Task Title"),
TableColumn(field="done", title="Completed"),
TableColumn(field="priority_task", title="Task Priority"),
# Add more columns as needed
]
# Create the DataTable
data_table = DataTable(source=source, columns=columns, width=800, height=280, sortable=True, selectable=True)
# Embed the DataTable using Bokeh's components
script, div = components(data_table)
# Return as an HTML Div
return Div(
script,
div,
style="border: 1px solid #ddd; padding: 10px; border-radius: 5px;",
)
async def filter_table(self, request):
"""
Handle filtering of the Bokeh DataTable based on slider input.
Args:
request: The incoming HTTP request containing the slider value.
Returns:
HTML: The updated Bokeh DataTable after applying the filter.
"""
form = await request.form()
max_priority = int(form.get('priority-slider', self.tasks().priority.max()))
logger.debug(f"Filtering tasks with priority <= {max_priority}")
# Fetch all active customers
active_customers = self.customers("active=?", (True,), order_by='priority')
customer_ids = [customer.id for customer in active_customers]
# Fetch tasks with priority <= max_priority
tasks = self.tasks("profile_id IN ? AND priority <= ?", (customer_ids, max_priority), order_by='priority')
logger.info(f"Fetched {len(tasks)} tasks after applying priority filter.")
# Load data into Pandas DataFrames
customers_df = pd.DataFrame([customer.__dict__ for customer in active_customers])
tasks_df = pd.DataFrame([task.__dict__ for task in tasks])
# Merge DataFrames to associate tasks with customers
merged_df = pd.merge(customers_df, tasks_df, left_on='id', right_on='profile_id', how='left', suffixes=('_customer', '_task'))
# Convert the merged DataFrame to a CSV string for download
csv_data = merged_df.to_csv(index=False)
logger.debug("CSV data updated for download.")
# Update the last_max_priority in the database for download consistency
db["last_max_priority"] = max_priority
# Generate the filtered Bokeh DataTable
bokeh_table = self.create_bokeh_table(merged_df)
return bokeh_table
async def generate_report(self, request, customer_id: int):
"""
Handle the generation of a project plan for a specific customer.
This simulates a long-running process, interacts with the chatbot to provide humorous feedback,
and finally offers a link to the generated report.
Args:
request: The incoming HTTP request.
customer_id (int): The ID of the customer for whom to generate the report.
Returns:
HTML: The updated report section with the generated report link.
"""
logger.debug(f"Initiating report generation for customer ID: {customer_id}")
# Validate customer existence
try:
customer = self.customers[customer_id]
logger.info(f"Generating report for customer: {customer.name}")
except NotFoundError:
logger.error(f"Customer ID {customer_id} not found.")
return Html(
Body(
P("Customer not found."),
)
)
# Simulate long-running event
await chatq(f"Generating project plan for {customer.name}. Please wait...")
# Simulate delay
await asyncio.sleep(5) # Simulate a 5-second long task
# Interact with the chatbot for humorous feedback
await chatq("Finally! Here it is...")
# Simulate report generation and provide a link
report_link = f"/reports/generated/{customer_id}/project_plan.pdf"
# Update the UI with the generated report link
return Div(
A(
"Download Project Plan",
href=report_link,
target="_blank",
cls="button",
style="margin-top: 10px;"
),
style="margin-top: 10px;",
)
async def download_csv(self, request):
"""
Serve the filtered report as a downloadable CSV file.
Args:
request: The incoming HTTP request.
Returns:
StreamingResponse: The CSV file as a downloadable response.
"""
logger.debug("Preparing CSV download for Customer Tasks Report.")
# Fetch the current filter from the database
max_priority = db.get("last_max_priority", self.tasks().priority.max())
# Fetch all active customers
active_customers = self.customers("active=?", (True,), order_by='priority')
customer_ids = [customer.id for customer in active_customers]
# Fetch tasks with priority <= max_priority
tasks = self.tasks("profile_id IN ? AND priority <= ?", (customer_ids, max_priority), order_by='priority')
logger.info(f"Fetched {len(tasks)} tasks for CSV download.")
# Load data into Pandas DataFrames
customers_df = pd.DataFrame([customer.__dict__ for customer in active_customers])
tasks_df = pd.DataFrame([task.__dict__ for task in tasks])
# Merge DataFrames to associate tasks with customers
merged_df = pd.merge(customers_df, tasks_df, left_on='id', right_on='profile_id', how='left', suffixes=('_customer', '_task'))
# Convert the merged DataFrame to CSV
csv_data = merged_df.to_csv(index=False)
# Serve the CSV as a downloadable response
buffer = StringIO()
buffer.write(csv_data)
buffer.seek(0)
return StreamingResponse(
buffer,
media_type='text/csv',
headers={
'Content-Disposition': f'attachment; filename="customer_tasks_report.csv"'
}
)
Explanation:
get_report_page
: Serves the main report page with the Bokeh DataTable and range slider.create_bokeh_table
: Generates a Bokeh DataTable from a Pandas DataFrame.filter_table
: Handles slider input to filter tasks based on priority and updates the Bokeh table accordingly.generate_report
: Simulates report generation with chatbot interactions and provides a download link.download_csv
: Serves the filtered report as a downloadable CSV file.
9. Implement Frontend Interactivity
Ensure that the frontend elements (PicoCSS range slider and Bokeh DataTable) interact seamlessly, providing real-time data filtering and an enhanced user experience.
Implementation:
The HTMX attributes in the range slider facilitate automatic POST requests to the backend when the slider value changes, updating the Bokeh table without a full page reload.
Code Example:
Div(
Label("Filter by Task Priority:"),
Input(
type="range",
min=str(merged_df['priority_task'].min()),
max=str(merged_df['priority_task'].max()),
value=str(merged_df['priority_task'].max()),
step="1",
id="priority-slider",
cls="slider",
style="width: 100%;",
hx_post="/reports/customers/filter",
hx_trigger="change",
hx_target="#bokeh-table",
hx_swap="outerHTML",
),
style="margin-top: 10px;",
)
Explanation:
- HTMX Attributes:
hx_post
: Sends a POST request to/reports/customers/filter
when the slider value changes.hx_trigger
: Specifies that thechange
event triggers the request.hx_target
: Identifies the target element (#bokeh-table
) to update with the response.hx_swap
: Determines that the entire target element (outerHTML
) will be replaced with the response.
- Range Slider:
- Adjusting the slider automatically filters the data displayed in the Bokeh table based on task priority.
10. Test the Advanced Reporting SPA
After implementing the advanced features, thoroughly test the reporting SPA to ensure all components function as intended.
Testing Steps:
- Access the Report Page:
- Navigate to
/reports/customers
to view the main report page. - Verify that the Bokeh DataTable displays all customers and their associated tasks.
- Navigate to
- Use the Range Slider:
- Adjust the PicoCSS range slider to filter tasks by priority.
- Confirm that the Bokeh DataTable updates dynamically to reflect the filtered data.
- Download CSV:
- Click the “Download Report as CSV” button.
- Ensure that the downloaded CSV file contains data matching the current filter.
- Generate Project Plan:
- Click the “Generate Project Plan” button next to a customer.
- Observe the chatbot interactions providing humorous feedback during the simulated delay.
- Confirm that a download link for the project plan appears after the process completes.
- Verify Responsiveness:
- Test the application on different devices and screen sizes to ensure responsive design.
- Check Logs:
- Review application logs to confirm that all operations are logged appropriately and no errors are present.
Example Testing Scenario:
- Scenario: Filtering tasks with priority <= 3.
- Set the range slider to 3.
- Observe that the Bokeh table only displays tasks with priority 3 or lower.
- Download the CSV and verify its contents.
- Generate a project plan for a customer and ensure the download link appears post-processing.
Naming Conventions and Table Management
Adhering to consistent naming conventions is crucial for maintaining clarity and avoiding conflicts within the system.
Guidelines:
- App Naming: Use descriptive and unique names for each app (e.g.,
CustomerTasksReportApp
). - Table Naming: Follow a pluralized form for table names (e.g.,
tasks
,customers
,reports
). - Primary Keys: Clearly define primary keys (
pk
) for each table to facilitate efficient CRUD operations. - Route Naming: Use clear and hierarchical route structures (e.g.,
/reports/customers
,/reports/customers/generate/{customer_id}
). - Consistent Suffixes: Maintain consistent suffixes for methods and endpoints (e.g.,
generate_report
,filter_table
).
Example:
# Table Definitions
tasks = Task # Represents the 'tasks' table
customers = Customer # Represents the 'customers' table
reports = Report # Represents the 'reports' table
Avoiding Configuration Collisions
When integrating new plugins, ensure that configurations remain isolated to prevent unintended interactions.
Strategies:
- Namespace Isolation: Encapsulate app-specific configurations within their respective classes or modules.
- Dedicated Slots: Allocate unique slots or sections for new apps’ configurations to avoid overlapping with existing settings.
- Consistent Prefixes/Suffixes: Use standardized prefixes or suffixes for configuration keys related to specific apps (e.g.,
customer_tasks_report_
). - Unique IDs: Assign unique identifiers to HTML elements to prevent conflicts in the DOM.
Example:
# Within CustomerTasksReportApp class
self.rt('/reports/customers', methods=['GET'])(self.get_report_page)
self.rt('/reports/customers/generate/{customer_id}', methods=['POST'])(self.generate_report)
self.rt('/reports/customers/filter', methods=['POST'])(self.filter_table)
self.rt('/reports/customers/download', methods=['GET'])(self.download_csv)
By maintaining clear boundaries and consistent naming, each plugin operates independently without interfering with others.
Conclusion
This advanced guide has expanded your Python web application by integrating data science tools—Pandas and Bokeh—into the CustomerTasksReportApp
. These enhancements empower data scientists to perform sophisticated data manipulations and create interactive visualizations, bridging the gap between Jupyter Notebooks and dynamic web applications.
Key Takeaways:
- Data Manipulation with Pandas: Efficiently process and analyze data within the web application.
- Interactive Visualizations with Bokeh: Create dynamic and responsive data tables that enhance data interpretation.
- Frontend Interactivity with PicoCSS: Implement intuitive UI elements like range sliders to provide real-time data filtering.
- Seamless Integration: Leverage existing CRUD applications to serve as data sources for advanced reporting features.
- Enhanced User Experience: Offer downloadable reports and interactive visualizations that cater to data scientists’ needs.
By following the detailed, step-by-step instructions, you can confidently extend your application with advanced reporting capabilities, fostering a robust and flexible system poised for future growth. Embrace the power of modular design and continue building innovative features that elevate your application’s utility and user experience.
For further assistance or queries, refer to the in-line comments within the codebase or engage with the development community associated with this project. Transition smoothly from data analysis in Jupyter Notebooks to comprehensive web-based dashboards, harnessing the full potential of the FastHTML framework tailored for data science professionals.
Appendix
Additional Resources
- Pandas Documentation: https://pandas.pydata.org/docs/
- Bokeh Documentation: https://docs.bokeh.org/en/latest/
- PicoCSS Documentation: https://picocss.com/docs/
- HTMX Documentation: https://htmx.org/docs/
- FastHTML Framework: [Link to your FastHTML framework documentation or repository]
Troubleshooting
- Bokeh Table Not Displaying:
- Ensure that Bokeh’s JavaScript and CSS are correctly included in your HTML templates.
- Verify that the
components
function from Bokeh is properly generating the necessary scripts. - Check browser console for any JavaScript errors.
- Range Slider Not Filtering Data:
- Confirm that the HTMX attributes (
hx_post
,hx_target
, etc.) are correctly set on the slider input. - Ensure that the backend endpoint
/reports/customers/filter
is properly handling the POST request and returning the updated Bokeh table. - Verify that the slider’s
min
,max
, andstep
values are correctly derived from the data.
- Confirm that the HTMX attributes (
- CSV Download Issues:
- Check that the
/reports/customers/download
endpoint is correctly generating and serving the CSV data. - Ensure that the
Content-Disposition
header is set toattachment
to prompt the download.
- Check that the
- Chatbot Interactions Not Working:
- Verify that the WebSocket connection is established and that messages are being sent and received.
- Check that the chatbot backend (
chat_with_ollama
function) is operational and responding as expected. - Review application logs for any errors related to WebSocket communication.
By meticulously following this guide, you can elevate your web application to cater to data scientists seeking interactive and dynamic reporting tools, bridging the capabilities of data analysis in notebooks with the accessibility and reach of web applications.