Taming Tight Coupling: Building a Flexible Plugin System in Python & FastHTML
This article provides a fascinating and highly practical account of tackling a common software architecture challenge: achieving modularity and extensibility within the constraints of existing frameworks. The author does an excellent job detailing the initial problem (dual extension systems), the theoretical approaches considered (including seeking AI assistance), and the often messy reality of implementation where initial plans didn’t quite fit.
The key takeaway – the pivot from attempting a separate, complex registry to adapting the existing, working workflow discovery pattern – is a valuable lesson in pragmatic software development. Recognizing and leveraging what already works often leads to simpler, more maintainable solutions. The detailed breakdown of the “binary search” debugging process and the final successful implementation using the CompetitorApp as a proof-of-concept clearly demonstrate the viability and benefits of the unified plugin architecture.
Overall, it’s a well-documented journey combining technical depth (exploring FastHTML, MiniDataAPI, plugin patterns) with relatable development struggles and the satisfaction of finding an elegant solution. It serves as an insightful case study for developers facing similar challenges with framework limitations or designing extensible systems. The inclusion of code snippets and the clear explanation of the final CompetitorPlugin make the achieved solution concrete and understandable.
CRUD is Dry & Workflows are WET
I’m hitting above my weight class due to AI coding assistance, however the key is not having it code for you. The key is having it pound into your thick noggin what’s going on and why.
Dry CRUD
What? CRUD stands for Create, Read, Update and Delete. Most apps that use data use this traditional rigid sort of database. There’s a table with rows and columns, a lot like a spreadsheet. A single row is a record in a database, and the columns are fields, and the content per field is the cell value. Such table data is known as tabular, and it’s used for all those things like Employee records or sales ledgers. With rigid tabular data comes predictability, and with predictability comes efficient templating techniques that avoid reusing code and repeating yourself. In other words, you Don’t Repeat Yourself. DRY.
Wet Workflows
Well, what the heck are Workflows, then? Well, they’re literally anything you can write out. First do this, then do that. Not only does it break the rigid row & column format, but the complexity of an individual step in a workflow can be anything. Whereas a traditional Employee database is a list, a Workflow is what an employee is instructions for what the Employee is supposed to do all day – like the Employee Handbook documenting procedures. Two procedures that seem similar but different in small but critical ways may have to be written out twice to get it exactly right. Write everything twice. WET.
Pipulate’s CRUD & Workflows
Understanding Web Frameworks
Pipulate is a locally running web framework that has both a CRUD database system and a Workflow procedural system built into it. Something is a framework when it does a lot of work for you by adopting a conventional ways of doing a things, therefore making a lot of assumptions and taking a lot of shortcuts. The apps you build out of frameworks spring to life with much less work on your part than without using a framework, only needing to seed it with some light configuration. And frameworks are built on frameworks are built on frameworks.
Python as a Framework Platform
Pipulate is built on FastHTML which is built on Python, all of them frameworks. Yes, Python is not only a programming language, but is a full fledged framework in its own right, with so many high-level primitives (lists, tuples & dicts) and pre-built standard library components (webservers, databases). That’s why everyone’s favorite sport is making their own framework out of Python. It takes you 80% of the way there already, so all you need do is add the last 20%. That’s what I did! And so I get to focus on fun things like auto-configuring plugin systems, so your extensions to my frameworks are even easier!
Auto-Configuring Plugin Systems
Auto-configuring plugin systems generally let you drop a file into a certain directory, and the system recognizes its presence and extends its capabilities in accordance with the plugin. And I already did that on Pipulate with Workflows. They were getting so long, WET as they were, that to keep them in the core code was becoming ridiculous. And that work is done and behind me, but now I’m getting ready to make the DRY CRUD apps externalized too – or at least one of them, the ToDo app. The other of the two that exist right now, the Profiles app, is probably just too integrated into the app to extract. So even better! One App stays in and the other gets plucked out and becomes the template for further expansion.
FastHTML CRUD Implementation
Now the thing is, these CRUD database apps are implemented with the FastHTML
helper class called fast_app() that has this pattern:
app, rt, (store, Store), (tasks, Task), (profiles, Profile), (pipeline, Pipeline) = fast_app(
"data/data.db",
# ... configuration ...
task={
"id": int,
"name": str,
"done": bool,
"priority": int,
"profile_id": int,
"pk": "id"
},
# ... other schemas ...
)
profile_app = ProfileApp(table=profiles)
profile_app.register_routes(rt)
todo_app = TodoApp(table=tasks)
todo_app.register_routes(rt)
This pattern creates a tight coupling between the application initialization and the database schema definition. The fast_app() function returns the application object, a route registration function, and handles to the database tables we defined. These table handles are then passed to our app classes, which register their routes with the main application.
The Plugin Challenge
The challenge I’m facing is how to externalize the TodoApp component while working within the constraints of this architecture. I can’t modify the fast_app() function since it’s part of the FastHTML library. But I still want to create a plugin system that allows new CRUD apps to be added to Pipulate without modifying the core code.
This is a common challenge in application design: how do you make a system extensible when some of its core components are tightly coupled? The answer lies in creating a plugin architecture.
What is a Plugin Architecture?
A plugin architecture is a design pattern that allows an application to be extended with new features without modifying its core code. It’s like creating a system of “slots” where additional functionality can be plugged in, similar to how you can add new attachments to a kitchen mixer or new apps to your smartphone.
Plugin architectures are valuable for several reasons:
- Maintainability: They keep your core application lean and focused on essential functionality
- Extensibility: They allow your application to grow in ways you didn’t anticipate
- Collaboration: They enable others to contribute features without touching core code
- Customization: They let users adapt the application to their specific needs
- Deployment Flexibility: They allow updating individual components without rebuilding the entire application
Python’s Plugin Approaches
Python’s dynamic nature makes it particularly well-suited for plugin architectures. There are three main approaches to implementing plugins in Python:
1. Naming Convention
This approach discovers plugins based on their module or package names. For example, Flask uses the naming convention flask_{plugin_name} for its extensions.
2. Namespace Packages
This approach uses Python’s namespace packages to organize plugins. Plugins can be installed in a specific namespace, and the application can discover all modules and packages in that namespace.
3. Entry Points
This approach uses package metadata to declare and discover plugins. Plugins register themselves using entry points in their package metadata, and the application can discover all registered plugins.
Pipulate’s Workflow Plugin System
I’ve already implemented a plugin system for Workflows in Pipulate using the naming convention approach. Workflow plugins are Python modules placed in a specific directory, and Pipulate discovers them at startup:
def discover_workflow_files():
workflow_modules = {}
workflows_dir = os.path.join(os.path.dirname(__file__), 'workflows')
# Discovery logic...
return workflow_modules
This system works well for Workflows because they don’t need to define database schemas or register routes. They simply provide rendering functions that integrate with the main UI.
Externalizing TodoApp: A Hybrid Approach
To externalize TodoApp while working within the constraints of FastHTML, I need a hybrid approach that combines elements from all three plugin discovery methods and leverages the flexibility of MiniDataAPI.
The MiniDataAPI Advantage
FastHTML uses MiniDataAPI, a lightweight database abstraction specification that provides a consistent API across different database backends. The key insight is that while fast_app() is a convenience wrapper, MiniDataAPI allows creating separate database connections and tables outside of this wrapper.
This means a plugin can create its own database connection and define its own schema:
from fastlite import database
class TodoAppPlugin:
def __init__(self, app, rt, db_path="data/data.db"):
self.db = database(db_path)
self.tasks = self.db.create({
"id": int,
"name": str,
"done": bool,
"priority": int,
"profile_id": int,
"pk": "id"
})
self.todo_app = TodoApp(table=self.tasks)
self.register_routes(rt)
Building a Plugin System for CRUD Apps
To implement a plugin system for CRUD apps in Pipulate, I’ll need:
- A Plugin Base Class: Defines the interface that all plugins must implement
- A Plugin Registry: Keeps track of available plugins
- A Discovery Mechanism: Finds and loads plugins
- Integration with the Rendering System: Ensures plugin UIs are displayed correctly
Here’s how these components would work together:
1. Plugin Base Class
class PipulatePlugin:
"""Base class for all Pipulate plugins"""
NAME = None
DISPLAY_NAME = None
def __init__(self, app, rt, db=None):
self.app = app
self.rt = rt
self.db = db
def register_routes(self):
"""Register routes with the application"""
pass
async def render(self, render_items=None):
"""Render the plugin UI"""
pass
2. Plugin Registry
class PluginRegistry:
def __init__(self):
self.plugins = {}
def register(self, plugin):
self.plugins[plugin.NAME] = plugin
def get(self, name):
return self.plugins.get(name)
def get_all(self):
return self.plugins.values()
3. Discovery Mechanism
def discover_plugins():
plugin_modules = {}
plugins_dir = os.path.join(os.path.dirname(__file__), 'plugins')
# Similar discovery logic to the workflow system
return plugin_modules
4. Integration with Rendering
async def create_grid_left(menux, render_items=None):
plugin = plugin_registry.get(menux)
if plugin:
return await plugin.render(render_items)
# ... existing code ...
Implementing TodoApp as a Plugin
With this architecture in place, I can implement TodoApp as a plugin:
class TodoAppPlugin(PipulatePlugin):
NAME = "todo"
DISPLAY_NAME = "Tasks"
def __init__(self, app, rt, db_path="data/data.db"):
super().__init__(app, rt)
from fastlite import database
self.db = database(db_path)
self.tasks = self.db.create({
"id": int,
"name": str,
"done": bool,
"priority": int,
"profile_id": int,
"pk": "id"
})
self.todo_app = TodoApp(table=self.tasks)
self.register_routes()
def register_routes(self):
self.todo_app.register_routes(self.rt)
async def render(self, render_items=None):
return await todo_render(self.NAME, render_items)
Future Extensions: Entry Points
For distributing plugins as separate packages, I could add support for entry points:
# In a plugin's pyproject.toml
[project.entry-points.'pipulate.plugins']
todo = 'pipulate_todo_plugin:TodoAppPlugin'
Then discover them in the main application:
import importlib.metadata
def load_entry_point_plugins(app, rt):
plugins = {}
for entry_point in importlib.metadata.entry_points(group='pipulate.plugins'):
plugin_class = entry_point.load()
plugin = plugin_class(app, rt)
plugins[entry_point.name] = plugin
return plugins
The Power of Plugins
By implementing this hybrid plugin architecture, I can externalize TodoApp while working within the constraints of FastHTML and leveraging the flexibility of MiniDataAPI. This approach builds on my existing Workflow plugin system while adding the capability for plugins to define their own database schemas.
The resulting architecture will be more modular, maintainable, and extensible, allowing Pipulate to grow through community contributions while maintaining a stable core. It transforms Pipulate from a static application into an evolving ecosystem, empowering users to customize their experience and enabling developers to collaborate on extending functionality.
Whether you’re building a small utility or a complex application like Pipulate, considering a plugin architecture early in your design process can pay dividends in flexibility, maintainability, and community engagement. Python’s dynamic nature and rich ecosystem make it an excellent language for implementing plugin systems, with multiple approaches to suit different needs and complexity levels.
By embracing plugins, you’re not just building an application—you’re creating a platform for innovation.
Once Bitten, Twice Shy
Now I Demand Non-breaking Step-by-Step Guides From AI Assistants
So that’s the overview of what I have planned. However, I have dove into refactor-hell one too many times lately. Everything is a rabbit hole with cascading unintended consequences, so my new style is to demand non-breaking step-by-step guides from the coding assistants.
Me: Do you see this actually viable and approachable within the current code-base? Could you analyze how I could break this down into step-by-step (non-breaking) steps?
Claude 3.7 Sonnet:
Implementing a Plugin Architecture in Pipulate: A Practical Roadmap
Viability Assessment
After analyzing your codebase, I believe implementing a plugin architecture for externalizing TodoApp is definitely viable. The current architecture already has several elements that make this transition feasible:
- Existing Workflow Plugin System: You’ve already implemented a plugin system for Workflows, providing a proven pattern to follow
- MiniDataAPI Flexibility: The MiniDataAPI specification allows creating database connections and tables outside of
fast_app() - Modular UI Rendering: Your
create_grid_left()function already handles different rendering scenarios based on menu selection - Clean Component Boundaries: TodoApp has relatively clear boundaries with the main application
Step-by-Step Implementation Plan
Here’s a practical roadmap that breaks down the implementation into non-breaking incremental steps:
Phase 1: Prepare the Plugin Infrastructure (Non-Breaking)
- Create a Plugin Base Class:
class PipulatePlugin: """Base class for all Pipulate plugins""" NAME = None DISPLAY_NAME = None def __init__(self, app, rt, db=None): self.app = app self.rt = rt self.db = db def register_routes(self): """Register routes with the application""" pass async def render(self, render_items=None): """Render the plugin UI""" pass - Implement a Plugin Registry:
class PluginRegistry: def __init__(self): self.plugins = {} def register(self, plugin): self.plugins[plugin.NAME] = plugin def get(self, name): return self.plugins.get(name) def get_all(self): return self.plugins.values() - Initialize the Registry:
# After app initialization plugin_registry = PluginRegistry()
These steps don’t modify any existing functionality but lay the groundwork for the plugin system.
Phase 2: Create a TodoApp Plugin Wrapper (Non-Breaking)
- Create a TodoApp Plugin Class:
class TodoAppPlugin(PipulatePlugin): NAME = "todo" DISPLAY_NAME = "Tasks" def __init__(self, app, rt, tasks_table): super().__init__(app, rt) self.tasks = tasks_table self.todo_app = TodoApp(table=tasks_table) def register_routes(self): self.todo_app.register_routes(self.rt) async def render(self, render_items=None): return await todo_render(self.NAME, render_items) - Register the Existing TodoApp as a Plugin:
# After todo_app initialization todo_plugin = TodoAppPlugin(app, rt, tasks) plugin_registry.register(todo_plugin) - Update the Rendering System (Dual-Path):
async def create_grid_left(menux, render_items=None): # Try plugin registry first plugin = plugin_registry.get(menux) if plugin: return await plugin.render(render_items) # Fall back to existing logic if menux == profile_app.name: return await profile_render() elif menux == todo_app.name: return await todo_render(menux, render_items) # ... rest of the existing code ...
This phase wraps the existing TodoApp in a plugin interface without changing its functionality, creating a dual-path for rendering that will eventually be consolidated.
Phase 3: Create a Plugin Discovery Mechanism (Non-Breaking)
- Create a Plugins Directory:
mkdir -p pipulate/plugins touch pipulate/plugins/__init__.py - Implement Plugin Discovery:
def discover_plugins(app, rt): discovered_plugins = {} plugins_dir = os.path.join(os.path.dirname(__file__), 'plugins') if not os.path.exists(plugins_dir): return discovered_plugins for filename in os.listdir(plugins_dir): if filename.endswith('.py') and filename != '__init__.py': module_name = filename[:-3] try: module = importlib.import_module(f'pipulate.plugins.{module_name}') for attr_name in dir(module): attr = getattr(module, attr_name) if (isinstance(attr, type) and issubclass(attr, PipulatePlugin) and attr != PipulatePlugin): plugin_class = attr plugin = plugin_class(app, rt) discovered_plugins[plugin.NAME] = plugin except Exception as e: logger.error(f"Error loading plugin {module_name}: {e}") return discovered_plugins - Load Discovered Plugins:
# After initializing plugin_registry discovered_plugins = discover_plugins(app, rt) for name, plugin in discovered_plugins.items(): plugin_registry.register(plugin)
This phase adds the ability to discover plugins without actually moving TodoApp yet.
Phase 4: Externalize TodoApp (Potentially Breaking, but Controlled)
- Create a TodoApp Plugin File:
# pipulate/plugins/todo_plugin.py from fastlite import database from pipulate.server import PipulatePlugin, TodoApp, todo_render class TodoAppPlugin(PipulatePlugin): NAME = "todo" DISPLAY_NAME = "Tasks" def __init__(self, app, rt, db_path="data/data.db"): super().__init__(app, rt) self.db = database(db_path) self.tasks = self.db.create({ "id": int, "name": str, "done": bool, "priority": int, "profile_id": int, "pk": "id" }) self.todo_app = TodoApp(table=self.tasks) self.register_routes() def register_routes(self): self.todo_app.register_routes(self.rt) async def render(self, render_items=None): return await todo_render(self.NAME, render_items) - Modify the Main Application (Controlled Breaking Change):
# Comment out the TodoApp initialization # todo_app = TodoApp(table=tasks) # todo_app.register_routes(rt) # Remove TodoApp from base_menu_items base_menu_items = ['', 'profile'] # Removed todo_app.name - Update the Rendering System (Final Consolidation):
async def create_grid_left(menux, render_items=None): # Try plugin registry first plugin = plugin_registry.get(menux) if plugin: return await plugin.render(render_items) # Fall back to existing logic for non-plugin components if menux == profile_app.name: return await profile_render() elif menux in workflow_instances: # ... workflow rendering ... else: return await introduction.introduction_render()
This phase externalizes TodoApp as a plugin. It’s potentially breaking, but by following the previous phases, you’ve ensured that the plugin system is ready to take over.
Phase 5: Enhance the Plugin System (Future Improvements)
- Add Entry Point Support:
def load_entry_point_plugins(app, rt): try: from importlib.metadata import entry_points except ImportError: from importlib_metadata import entry_points for entry_point in entry_points(group='pipulate.plugins'): try: plugin_class = entry_point.load() plugin = plugin_class(app, rt) plugin_registry.register(plugin) except Exception as e: logger.error(f"Error loading plugin {entry_point.name}: {e}") - Add Plugin Configuration Support:
def load_plugin_config(): config_path = Path('config/plugins.json') if config_path.exists(): return json.loads(config_path.read_text()) return {} - Implement Plugin Lifecycle Management:
class PluginRegistry: # ... existing methods ... def enable(self, name): plugin = self.plugins.get(name) if plugin and hasattr(plugin, 'enable'): plugin.enable() def disable(self, name): plugin = self.plugins.get(name) if plugin and hasattr(plugin, 'disable'): plugin.disable()
These enhancements add more sophisticated plugin capabilities without breaking existing functionality.
Implementation Considerations
Data Migration
When externalizing TodoApp, you’ll need to consider data migration. Options include:
- Same Database File: Have the plugin use the same database file, accessing the existing
taskstable - Data Migration Script: Create a script to migrate data from the main database to the plugin’s database
- Read-Only Legacy Support: Keep the old table for reading but write to the new one
Testing Strategy
For each phase:
- Unit Tests: Test the plugin infrastructure components in isolation
- Integration Tests: Test the interaction between plugins and the main application
- End-to-End Tests: Test the complete user flow with plugins
Rollback Plan
For each phase, especially Phase 4:
- Code Backup: Keep a backup of the working code before changes
- Feature Flags: Use feature flags to enable/disable the plugin system
- Dual-Path Support: Maintain both old and new paths until confident in the new system
Conclusion
This step-by-step approach makes implementing a plugin architecture in Pipulate both viable and approachable. By breaking it down into non-breaking phases, you can gradually transition to a plugin-based architecture while maintaining a working application throughout the process.
The existing Workflow plugin system provides a solid foundation, and the flexibility of MiniDataAPI allows for clean externalization of TodoApp. With this approach, you’ll create a more modular, maintainable, and extensible application that can grow through community contributions.
Taking The Plunge
Okay, so now I’ve got my allegedly non-breaking (though Claude does fess up to the possibility of breakage) step-by-step guide to externalizing the other plugin system in Pipulate.
Python Plugin Architectures: The Journey to Unified Plugins in Pipulate
The Challenge: Unifying Two Plugin Systems
In our previous installment (above), we explored the theoretical approach to implementing a plugin architecture in Pipulate, focusing on externalizing the ToDo app. Now, let’s document the practical journey we took to unify Pipulate’s plugin systems.
The Starting Point: Two Separate Systems
Pipulate initially had two separate extension mechanisms:
- Workflows: A well-functioning plugin system for procedural, step-by-step processes
- Plugins: A separate system intended for CRUD applications that wasn’t fully integrated
This dual approach created unnecessary complexity and confusion. Our goal was to unify these systems into a single, coherent plugin architecture.
The Discovery Process
Initial Attempts: Plugin Registry Approach
Our first approach was to use the existing plugin_registry system:
class HelloPlugin(PipulatePlugin):
NAME = "hello"
DISPLAY_NAME = "Hello Plugin"
def __init__(self, app=None, rt=None):
super().__init__(app, rt)
def landing(self, request):
return Div(
H1("Hello Plugin"),
P("This is a simple example plugin for Pipulate.")
)
plugin_registry.register(HelloPlugin())
However, this approach failed because:
- The plugin registry wasn’t being used for menu generation
- Plugins weren’t being discovered the same way as workflows
- The plugin wasn’t being instantiated with the required parameters
Analyzing the Workflow System
To understand why our plugin wasn’t appearing in the navigation menu, we needed to analyze how workflows were being discovered and registered:
def discover_workflow_files():
workflow_modules = {}
workflows_dir = os.path.join(os.path.dirname(__file__), 'workflows')
# Find all Python files in the workflows directory
for filename in os.listdir(workflows_dir):
if filename.endswith('.py') and not filename.startswith('__'):
module_name = filename[:-3] # Remove .py extension
try:
module = importlib.import_module(f'workflows.{module_name}')
workflow_modules[module_name] = module
except ImportError as e:
logger.error(f"Error importing workflow module {module_name}: {str(e)}")
return workflow_modules
The key insights were:
- Workflows were discovered by scanning a specific directory
- Each workflow module was imported using
importlib.import_module - Classes with a
landingmethod were identified as workflow classes - These classes were instantiated with specific parameters:
app,pipulate,pipeline, anddb - The instances were added to
workflow_instances, which was used to generate menu items
The Binary Search Approach
After several failed attempts, we took a systematic binary search approach to isolate the issue:
- We examined the logs to understand the discovery process
- We analyzed how menu items were being generated
- We compared the workflow system with the plugin system
- We identified the critical differences between the two systems
The key revelation was that the plugin system wasn’t following the same pattern as the workflow system, which was actually working.
The Solution: Unifying the Plugin Concept
Rather than trying to fix the plugin system, we decided to unify the concept by:
- Moving our plugin to the plugins directory
- Ensuring it had the same structure as workflow classes
- Modifying the server to look for plugins in the plugins directory
The Unified Plugin Structure
class HelloPlugin:
NAME = "hello"
DISPLAY_NAME = "Hello Plugin"
ENDPOINT_MESSAGE = "This is a simple plugin example."
def __init__(self, app, pipulate, pipeline, db):
self.app = app
self.pipulate = pipulate
self.pipeline = pipeline
self.db = db
logger.debug(f"HelloPlugin initialized with NAME: {self.NAME}")
def landing(self, request):
"""Landing page for the Hello plugin"""
logger.debug("HelloPlugin.landing method called")
return Div(
H1("Hello Plugin"),
P("This is a simple example plugin for Pipulate."),
_class="container"
)
async def render(self, render_items=None):
return Div(
H2("Hello from Plugin"),
P("This is a simple plugin example.")
)
The Critical Fix: Removing Self-Registration
The final piece of the puzzle was removing the self-registration code:
# Don't register here - let the discovery process handle it
# plugin_registry.register(HelloPlugin())
This allowed the plugin to be discovered and instantiated by the server’s discovery process, just like workflows.
Lessons Learned
Our journey to unify Pipulate’s plugin systems taught us several valuable lessons:
- Follow Working Patterns: Rather than creating a new system, adapt what’s already working
- Systematic Debugging: A binary search approach helps isolate issues in complex systems
- Simplify Mental Models: Unifying concepts (everything is a plugin) makes systems easier to understand
- Avoid Premature Abstraction: Sometimes the simplest solution is to follow existing patterns
The Road Ahead: Externalizing the ToDo App
While we’ve successfully unified the plugin concept, we still need to externalize the ToDo app. This will involve:
- Creating a ToDo plugin that defines its own database schema
- Ensuring it can access the same database as the main application
- Migrating existing data to the new plugin structure
- Updating the UI to use the plugin-based ToDo app
The groundwork we’ve laid by unifying the plugin system will make this process much smoother. By understanding how plugins are discovered and integrated into the navigation, we can now create a ToDo plugin that follows the same pattern.
Conclusion: The Power of Unified Concepts
Our journey demonstrates the value of unified concepts in software design. By aligning our plugin system with the existing workflow system, we’ve created a more coherent, maintainable architecture.
This unified approach offers several benefits:
- Consistency: All extensions follow the same pattern
- Simplicity: Developers only need to learn one extension mechanism
- Maintainability: The codebase is more focused and easier to understand
- Extensibility: Adding new plugins is straightforward and predictable
As we move forward with externalizing the ToDo app, we’ll build on this unified foundation, creating a truly modular and extensible application architecture.
The key insight from our experience is that sometimes the best solution isn’t creating a new system, but rather understanding and extending what’s already working. By embracing this principle, we’ve transformed Pipulate’s plugin architecture from a source of confusion to a model of clarity.
The Proof is in the Plugin: CompetitorApp Takes the Stage
Okay, Gemini 2.5 Pro, let’s bring this adventure to a close! Based on the final code and the journey described, here’s a concluding section that ties everything together:
After the winding journey, the theoretical discussions, and the crucial realization that adapting the existing workflow discovery pattern was the path of least resistance, the moment of truth arrived: implementing the CompetitorApp as a fully externalized plugin. And the results? A resounding success!
The final competitor_plugin.py code, combined with the necessary frontend adjustments in chat-scripts.js, stands as a testament to the viability of this unified plugin architecture. Let’s break down the key achievements demonstrated by this working example:
- Independent Schema Definition: Leveraging the flexibility of MiniDataAPI (via
fastlite), theCompetitorPluginsuccessfully creates and manages its owncompetitordatabase table entirely outside the initialfast_app()configuration. This overcomes the primary hurdle presented by the tightly coupled nature of FastHTML’s core setup. - Full CRUD Functionality: The plugin seamlessly replicates all the essential features of the original internal
TodoApp. It handles creating new competitors, toggling their ‘done’ status, deleting them, and even allows for inline editing of names, complete with cancel functionality – all powered by HTMX requests targeting the plugin’s specific endpoints. - Dynamic Sorting: The
chat-scripts.jsmodifications work perfectly with the plugin. The frontend correctly identifies the plugin’s base path (/competitor) and sends the sorted item priorities to the/competitor/sortendpoint, which theCompetitorApp’ssort_itemsmethod handles flawlessly. - Manual Route Registration: Recognizing that the
rtobject fromfast_appwasn’t the right tool for external plugins, theCompetitorPlugineffectively usesself.app.routedirectly within itsregister_plugin_routesmethod. This ensures that all necessary endpoints (/competitor,/competitor/delete/{id},/competitor/toggle/{id},/competitor/{id}for updates, and/competitor/sort) are correctly mapped to theCompetitorAppinstance’s methods. - Unified Discovery: By following the established workflow pattern (a class within the
pluginsdirectory having the necessaryNAME,DISPLAY_NAME,__init__, andlandingmethods), theCompetitorPluginis automatically discovered, instantiated, and integrated into the Pipulate UI and navigation menu without needing a separate, complex registry system.
The CompetitorApp, initially conceived as a non-colliding test case, now serves as the crucial proof-of-concept. It demonstrates definitively that CRUD-style applications can be successfully externalized and added to Pipulate simply by dropping a correctly structured Python file into the plugins directory.
With this validation, the path is clear. Externalizing the original TodoApp is no longer a daunting refactoring challenge but a straightforward task of applying the now-proven pattern. We can confidently move the TodoApp logic into its own plugin file, mirroring the structure of CompetitorPlugin. Interestingly, this opens the door to having multiple, similar CRUD applications running side-by-side, each managing its distinct data within the same database file, simply by adding more plugins.
This journey, from theoretical planning and AI consultations to practical implementation and debugging, ultimately led back to a simpler, more elegant solution: extend what works. The unified plugin system, born from adapting the workflow discovery mechanism, provides the modularity and extensibility Pipulate needed, finally taming the tightly coupled beast and paving the way for a more flexible and collaborative future. The “refactor adventure” has reached a successful conclusion, leaving behind a cleaner, more powerful architecture.
And For Posterity, Here’s the Plugin Code
# plugins/competitor_plugin.py
import os
import inspect
from loguru import logger
from fasthtml.common import *
import fastlite
import sys
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '..')))
from server import BaseApp, db as server_db, priority_key, LIST_SUFFIX
# --- Define the App Logic Class (inherits from BaseApp) ---
class CompetitorApp(BaseApp):
def __init__(self, table):
# Initialize BaseApp with specific names and fields for competitors
super().__init__(name='competitor', table=table, toggle_field='done', sort_field='priority')
self.item_name_field = 'name'
logger.debug(f"CompetitorApp initialized with table name: {table.name}")
def render_item(self, competitor):
# This method will call the specific rendering function for a competitor item
logger.debug(f"CompetitorApp.render_item called for: {competitor}")
# Pass 'self' (the CompetitorApp instance) to the render function
return render_competitor(competitor, self)
def prepare_insert_data(self, form):
# Prepare data for inserting a new competitor
name = form.get('competitor_name', form.get('name', '')).strip()
if not name:
logger.warning("Attempted to insert competitor with empty name.")
return None
current_profile_id = server_db.get("last_profile_id", 1)
logger.debug(f"Using profile_id: {current_profile_id} for new competitor")
# Query competitors for current profile using MiniDataAPI pattern
competitors_for_profile = self.table("profile_id = ?", [current_profile_id])
max_priority = max((c.priority or 0 for c in competitors_for_profile), default=-1) + 1
priority = int(form.get('competitor_priority', max_priority))
insert_data = {
"name": name,
"done": False,
"priority": priority,
"profile_id": current_profile_id,
}
logger.debug(f"Prepared insert data: {insert_data}")
return insert_data
def prepare_update_data(self, form):
# Prepare data for updating an existing competitor
name = form.get('competitor_name', form.get('name', '')).strip() # Allow updating name
if not name:
logger.warning("Attempted to update competitor with empty name.")
return None # Don't allow empty name update
# You might add other fields here if needed for updates
update_data = {
"name": name,
# Example: Allow updating 'done' status if needed via form
# "done": form.get('done', '').lower() == 'true',
}
logger.debug(f"Prepared update data: {update_data}")
return update_data
# --- Define the Rendering Function ---
def render_competitor(competitor, app_instance: CompetitorApp):
"""Renders a single competitor item as an LI element."""
cid = f'competitor-{competitor.id}'
logger.debug(f"Rendering competitor ID {competitor.id} with name '{competitor.name}'")
# Use the app_instance passed in to generate correct URLs
delete_url = app_instance.get_action_url('delete', competitor.id) # e.g., /competitor/delete/1
toggle_url = app_instance.get_action_url('toggle', competitor.id) # e.g., /competitor/toggle/1
update_url = f"/{app_instance.name}/{competitor.id}" # e.g., /competitor/1
checkbox = Input(
type="checkbox",
name="done_status" if competitor.done else None, # Use a unique name if needed
checked=competitor.done,
hx_post=toggle_url,
hx_swap="outerHTML",
hx_target=f"#{cid}",
)
delete_icon = A(
'🗑',
hx_delete=delete_url,
hx_swap='outerHTML',
hx_target=f"#{cid}",
style="cursor: pointer; display: inline; margin-left: 5px;", # Added margin
cls="delete-icon"
)
# Use competitor_name_{id} for unique IDs if needed
update_input_id = f"competitor_name_{competitor.id}"
name_display = Span( # Changed A to Span for non-clickable display initially
competitor.name,
id=f"competitor-name-display-{competitor.id}", # Unique ID for display span
style="margin-left: 5px; cursor: pointer;", # Make it look clickable
# JS to hide display, show form
onclick=(
f"document.getElementById('competitor-name-display-{competitor.id}').style.display='none'; "
f"document.getElementById('update-form-{competitor.id}').style.display='inline-flex'; "
f"document.getElementById('{update_input_id}').focus();"
)
)
update_form = Form(
Group( # Using Group for inline layout
Input(
type="text",
id=update_input_id,
value=competitor.name,
name="competitor_name", # Use the name expected by prepare_update_data
style="flex-grow: 1; margin-right: 5px; margin-bottom: 0;", # Adjusted styles
),
Button("Save", type="submit", style="margin-bottom: 0;"), # Adjusted styles
Button("Cancel", type="button", style="margin-bottom: 0;", cls="secondary",
# JS to hide form, show display
onclick=(
f"document.getElementById('update-form-{competitor.id}').style.display='none'; "
f"document.getElementById('competitor-name-display-{competitor.id}').style.display='inline';"
)),
style="align-items: center;" # Align items in the group
),
style="display: none; margin-left: 5px;", # Hidden initially, inline-flex later
id=f"update-form-{competitor.id}", # Unique ID for the form
hx_post=update_url,
hx_target=f"#{cid}",
hx_swap="outerHTML",
)
return Li(
checkbox,
name_display, # Show the name span
update_form, # Include the hidden update form
delete_icon,
id=cid,
cls='done' if competitor.done else '',
style="list-style-type: none; display: flex; align-items: center; margin-bottom: 5px;", # Flex layout
data_id=competitor.id,
data_priority=competitor.priority
)
# --- Define the Plugin Wrapper ---
class CompetitorPlugin:
NAME = "competitor"
DISPLAY_NAME = "Competitors"
ENDPOINT_MESSAGE = "Manage your list of competitors here. Add, edit, sort, and toggle their status."
def __init__(self, app, pipulate, pipeline, db_dictlike):
"""Initialize the Competitor Plugin."""
self.app = app
self.pipulate = pipulate
self.pipeline_table = pipeline
self.db_dictlike = db_dictlike
logger.debug(f"{self.DISPLAY_NAME} Plugin initializing...")
db_path = os.path.join(os.path.dirname(__file__), "..", "data", "data.db")
logger.debug(f"Using database path: {db_path}")
self.plugin_db = fastlite.database(db_path)
# --- REVISED TABLE CREATION using Schema Dict + .dataclass() call ---
competitor_schema = {
"id": int,
"name": str,
"done": bool,
"priority": int,
"profile_id": int,
"pk": "id" # Primary key definition
}
schema_fields = {k: v for k, v in competitor_schema.items() if k != 'pk'}
primary_key = competitor_schema.get('pk')
if not primary_key:
logger.error("Primary key 'pk' must be defined in the schema dictionary!")
raise ValueError("Schema dictionary must contain a 'pk' entry.")
try:
# 1. Get table handle via .t accessor
competitor_table_handle = self.plugin_db.t.competitor # Use 'competitor' as table name
logger.debug(f"Got potential table handle via .t accessor: {competitor_table_handle}")
# 2. Create the table *ONLY IF IT DOES NOT EXIST* using the handle
self.competitors_table = competitor_table_handle.create(
**schema_fields,
pk=primary_key,
if_not_exists=True
)
logger.info(f"Fastlite 'competitor' table created or accessed via handle: {self.competitors_table}")
# 3. *** ACTIVATE DATACLASS RETURNS for this handle ***
# We don't need to store the result, just call it.
self.competitors_table.dataclass()
logger.info(f"Called .dataclass() on table handle to enable dataclass returns.")
except Exception as e:
logger.error(f"Error creating/accessing 'competitor' table: {e}")
raise
# Instantiate the actual CompetitorApp logic class, passing the table handle
self.competitor_app_instance = CompetitorApp(table=self.competitors_table)
logger.debug(f"CompetitorApp instance created.")
self.register_plugin_routes()
logger.debug(f"{self.DISPLAY_NAME} Plugin initialized successfully.")
def register_plugin_routes(self):
"""Register routes manually using app.route, bypassing BaseApp's rt logic."""
prefix = f"/{self.competitor_app_instance.name}" # /competitor
# Determine the sort path based on the BaseApp pattern
sort_path = f"/{self.competitor_app_instance.name}/sort" # e.g., /competitor_sort
routes_to_register = [
(f'{prefix}', self.competitor_app_instance.insert_item, ['POST']),
(f'{prefix}/', self.competitor_app_instance.update_item, ['POST']),
(f'{prefix}/delete/', self.competitor_app_instance.delete_item, ['DELETE']),
(f'{prefix}/toggle/', self.competitor_app_instance.toggle_item, ['POST']),
(sort_path, self.competitor_app_instance.sort_items, ['POST']),
]
logger.debug(f"Registering routes for {self.NAME} plugin:")
for path, handler, methods in routes_to_register:
func = handler
self.app.route(path, methods=methods)(func)
logger.debug(f" Registered: {methods} {path} -> {handler.__name__}")
async def landing(self, request=None):
"""Renders the main view for the Competitor plugin."""
logger.debug(f"CompetitorPlugin.landing called")
app_name = self.NAME
current_profile_id = self.db_dictlike.get("last_profile_id", 1)
logger.debug(f"Landing page using profile_id: {current_profile_id}")
competitor_items_query = self.competitors_table(where=f"profile_id = {current_profile_id}")
competitor_items = sorted(competitor_items_query, key=priority_key)
logger.debug(f"Found {len(competitor_items)} competitors for profile {current_profile_id}")
add_placeholder = f'Add new {self.DISPLAY_NAME[:-1]}'
return Div(
Card(
H2(f"{self.DISPLAY_NAME} {LIST_SUFFIX}"),
Ul(
*[self.competitor_app_instance.render_item(item) for item in competitor_items],
id=f'{app_name}-list',
cls='sortable',
style="padding-left: 0;"
),
header=Form(
Group(
Input(
placeholder=add_placeholder,
id=f'{app_name}-name-input',
name='competitor_name',
autofocus=True
),
Button("Add", type="submit")
),
hx_post=f"/{app_name}",
hx_swap="beforeend",
hx_target=f"#{app_name}-list",
_="on htmx:afterRequest(maybeError) if maybeError === undefined then me.reset()"
)
),
id=f"{app_name}-content-container",
style="display: flex; flex-direction: column;"
)
async def render(self, render_items=None):
"""Fallback render method, currently just calls landing."""
logger.debug(f"CompetitorPlugin.render called, delegating to landing.")
# For workflows, render might be different, but for CRUD, landing is the main view
return await self.landing()
# No registration needed here - server.py discovery will find CompetitorPlugin
An AI’s Take
Okay, here are the requested pieces:
Author’s Summary (1st Person, Meta Description Style):
[Well written, moved to top!]
Title & Headline Ideas:
- Pipulate’s Plugin Evolution: Unifying CRUD and Workflows
- Taming Tight Coupling: Building a Flexible Plugin System in Python & FastHTML
- From Refactor Hell to Unified Plugins: A Pipulate Case Study
- Python Plugin Patterns: A Practical Journey in Pipulate
- Externalizing CRUD: How Pipulate Achieved Modularity
- The Power of Adaptation: Simplifying Pipulate’s Plugin Architecture
- One Plugin System to Rule Them All: Refactoring Pipulate
- Proof is in the Plugin: Validating Pipulate’s New Architecture
- Beyond
fast_app: Independent Plugins with FastHTML and MiniDataAPI
AI (My) Opinion of the Article:
[Also moved to top. Wow, Gemini 2.5 is getting really good at this.]