# cjm-fasthtml-workflows
<!-- WARNING: THIS FILE WAS AUTOGENERATED! DO NOT EDIT! -->
## Install
``` bash
pip install cjm_fasthtml_workflows
```
## Project Structure
nbs/
└── core/ (4)
├── deduplication.ipynb # Track one-time operations per job to prevent duplicates.
├── helpers.ipynb # Helper functions for common workflow patterns.
├── job_session.ipynb # Manage job-related metadata that needs to be accessed across multiple requests.
└── workflow_session.ipynb # Namespaced workflow state management within FastHTML sessions.
Total: 4 notebooks across 1 directory
## Module Dependencies
``` mermaid
graph LR
core_deduplication[core.deduplication<br/>DeduplicationTracker]
core_helpers[core.helpers<br/>Helpers]
core_job_session[core.job_session<br/>JobSessionManager]
core_workflow_session[core.workflow_session<br/>WorkflowSession]
core_helpers --> core_job_session
core_helpers --> core_deduplication
```
*2 cross-module dependencies detected*
## CLI Reference
No CLI commands found in this project.
## Module Overview
Detailed documentation for each module in the project:
### DeduplicationTracker (`deduplication.ipynb`)
> Track one-time operations per job to prevent duplicates.
#### Import
``` python
from cjm_fasthtml_workflows.core.deduplication import (
DeduplicationTracker
)
```
#### Classes
``` python
class DeduplicationTracker:
def __init__(
self,
sess:Any, # FastHTML session object
tracker_id:str # Unique identifier for this tracker (e.g., "saved_jobs", "sent_notifications")
)
"Track one-time operations per job to prevent duplicates."
def __init__(
self,
sess:Any, # FastHTML session object
tracker_id:str # Unique identifier for this tracker (e.g., "saved_jobs", "sent_notifications")
)
"Initialize deduplication tracker."
def is_processed(
self,
job_id:str # Unique job identifier
) -> bool: # True if job has been marked as processed, False otherwise
"Check if a job has been processed."
def mark_processed(
self,
job_id:str # Unique job identifier
) -> None
"Mark a job as processed."
def unmark_processed(
self,
job_id:str # Unique job identifier
) -> None
"Remove processed mark from a job (allows reprocessing)."
def clear_all(self) -> None:
"""Clear all processed job IDs."""
if self._key in self.sess
"Clear all processed job IDs."
def get_all_processed(self) -> Set[str]: # Set of job IDs that have been marked as processed
"Get all processed job IDs."
```
### Helpers (`helpers.ipynb`)
> Helper functions for common workflow patterns.
#### Import
``` python
from cjm_fasthtml_workflows.core.helpers import (
get_job_session_info,
once_per_job
)
```
#### Functions
``` python
def get_job_session_info(
job_id:str, # Unique job identifier
job:Any, # Job object (should have attributes like file_name, file_path, plugin_id)
sess:Any, # FastHTML session object
fallback_fields:Optional[Dict[str, Any]]=None # Optional additional fallback fields
) -> Tuple[Dict[str, Any], Dict[str, Any]]: # (file_info, plugin_info) dictionaries
"Retrieve job-related display information from session with fallbacks."
```
``` python
def once_per_job(
sess:Any, # FastHTML session object
job_id:str, # Unique job identifier
operation:Callable[[], Any], # Callable to execute (should take no arguments)
tracker_id:str="default_operation" # Unique identifier for this operation type
) -> Optional[Any]: # Result of operation if executed, None if already processed
"Execute an operation once per job using automatic deduplication."
```
### JobSessionManager (`job_session.ipynb`)
> Manage job-related metadata that needs to be accessed across multiple
> requests.
#### Import
``` python
from cjm_fasthtml_workflows.core.job_session import (
JobSessionManager
)
```
#### Classes
``` python
class JobSessionManager:
def __init__(
self,
sess:Any, # FastHTML session object
prefix:str="job" # Prefix for job keys in session
)
"Manage job-related metadata that needs to be accessed across multiple requests."
def __init__(
self,
sess:Any, # FastHTML session object
prefix:str="job" # Prefix for job keys in session
)
"Initialize job session manager."
def store_job_metadata(
self,
job_id:str, # Unique job identifier
metadata:Dict[str, Any] # Dictionary of metadata to store
) -> None
"Store metadata for a job."
def get_job_metadata(
self,
job_id:str, # Unique job identifier
default:Optional[Dict[str, Any]]=None # Default value if metadata not found
) -> Dict[str, Any]: # Job metadata dictionary
"Retrieve metadata for a job."
def delete_job_metadata(
self,
job_id:str # Unique job identifier
) -> None
"Remove metadata for a job."
def clear_all_jobs(self) -> None:
"""Clear all job metadata from session."""
keys_to_delete = [k for k in self.sess.keys() if k.startswith(f"{self.prefix}_")]
for key in keys_to_delete
"Clear all job metadata from session."
```
### WorkflowSession (`workflow_session.ipynb`)
> Namespaced workflow state management within FastHTML sessions.
#### Import
``` python
from cjm_fasthtml_workflows.core.workflow_session import (
WorkflowSession
)
```
#### Classes
``` python
class WorkflowSession:
def __init__(
self,
sess:Any, # FastHTML session object (from route handler parameter)
workflow_id:str # Unique identifier for this workflow (e.g., "registration", "checkout")
)
"Manage workflow state within FastHTML sessions with proper namespacing."
def __init__(
self,
sess:Any, # FastHTML session object (from route handler parameter)
workflow_id:str # Unique identifier for this workflow (e.g., "registration", "checkout")
)
"Initialize workflow session wrapper."
def set(
self,
key:str, # Key name within this workflow namespace
value:Any # Value to store (must be JSON-serializable for session storage)
) -> None
"Store value in workflow session."
def get(
self,
key:str, # Key name within this workflow namespace
default:Any=None # Default value if key not found
) -> Any: # Stored value or default
"Retrieve value from workflow session."
def delete(
self,
key:str # Key name within this workflow namespace
) -> None
"Remove value from workflow session."
def clear(self) -> None:
"""Clear all data for this workflow from session."""
keys_to_delete = [k for k in self.sess.keys() if k.startswith(self._prefix)]
for key in keys_to_delete
"Clear all data for this workflow from session."
def get_all(self) -> Dict[str, Any]: # Dictionary mapping original keys to their values
"""Get all workflow data as a dictionary with original (unprefixed) keys."""
result = {}
prefix_len = len(self._prefix)
for key, value in self.sess.items()
"Get all workflow data as a dictionary with original (unprefixed) keys."
def has(
self,
key:str # Key name within this workflow namespace
) -> bool: # True if key exists, False otherwise
"Check if key exists in workflow session."
```
Raw data
{
"_id": null,
"home_page": "https://github.com/cj-mills/cjm-fasthtml-workflows",
"name": "cjm-fasthtml-workflows",
"maintainer": null,
"docs_url": null,
"requires_python": ">=3.11",
"maintainer_email": null,
"keywords": "nbdev jupyter notebook python",
"author": "Christian J. Mills",
"author_email": "9126128+cj-mills@users.noreply.github.com",
"download_url": "https://files.pythonhosted.org/packages/32/45/abc5e82321ee76975da04703bdff6aed8645da6f8387597ef2c5f960a273/cjm_fasthtml_workflows-0.0.1.tar.gz",
"platform": null,
"description": "# cjm-fasthtml-workflows\n\n\n<!-- WARNING: THIS FILE WAS AUTOGENERATED! DO NOT EDIT! -->\n\n## Install\n\n``` bash\npip install cjm_fasthtml_workflows\n```\n\n## Project Structure\n\n nbs/\n \u2514\u2500\u2500 core/ (4)\n \u251c\u2500\u2500 deduplication.ipynb # Track one-time operations per job to prevent duplicates.\n \u251c\u2500\u2500 helpers.ipynb # Helper functions for common workflow patterns.\n \u251c\u2500\u2500 job_session.ipynb # Manage job-related metadata that needs to be accessed across multiple requests.\n \u2514\u2500\u2500 workflow_session.ipynb # Namespaced workflow state management within FastHTML sessions.\n\nTotal: 4 notebooks across 1 directory\n\n## Module Dependencies\n\n``` mermaid\ngraph LR\n core_deduplication[core.deduplication<br/>DeduplicationTracker]\n core_helpers[core.helpers<br/>Helpers]\n core_job_session[core.job_session<br/>JobSessionManager]\n core_workflow_session[core.workflow_session<br/>WorkflowSession]\n\n core_helpers --> core_job_session\n core_helpers --> core_deduplication\n```\n\n*2 cross-module dependencies detected*\n\n## CLI Reference\n\nNo CLI commands found in this project.\n\n## Module Overview\n\nDetailed documentation for each module in the project:\n\n### DeduplicationTracker (`deduplication.ipynb`)\n\n> Track one-time operations per job to prevent duplicates.\n\n#### Import\n\n``` python\nfrom cjm_fasthtml_workflows.core.deduplication import (\n DeduplicationTracker\n)\n```\n\n#### Classes\n\n``` python\nclass DeduplicationTracker:\n def __init__(\n self, \n sess:Any, # FastHTML session object\n tracker_id:str # Unique identifier for this tracker (e.g., \"saved_jobs\", \"sent_notifications\")\n )\n \"Track one-time operations per job to prevent duplicates.\"\n \n def __init__(\n self, \n sess:Any, # FastHTML session object\n tracker_id:str # Unique identifier for this tracker (e.g., \"saved_jobs\", \"sent_notifications\")\n )\n \"Initialize deduplication tracker.\"\n \n def is_processed(\n self, \n job_id:str # Unique job identifier\n ) -> bool: # True if job has been marked as processed, False otherwise\n \"Check if a job has been processed.\"\n \n def mark_processed(\n self, \n job_id:str # Unique job identifier\n ) -> None\n \"Mark a job as processed.\"\n \n def unmark_processed(\n self, \n job_id:str # Unique job identifier\n ) -> None\n \"Remove processed mark from a job (allows reprocessing).\"\n \n def clear_all(self) -> None:\n \"\"\"Clear all processed job IDs.\"\"\"\n if self._key in self.sess\n \"Clear all processed job IDs.\"\n \n def get_all_processed(self) -> Set[str]: # Set of job IDs that have been marked as processed\n \"Get all processed job IDs.\"\n```\n\n### Helpers (`helpers.ipynb`)\n\n> Helper functions for common workflow patterns.\n\n#### Import\n\n``` python\nfrom cjm_fasthtml_workflows.core.helpers import (\n get_job_session_info,\n once_per_job\n)\n```\n\n#### Functions\n\n``` python\ndef get_job_session_info(\n job_id:str, # Unique job identifier\n job:Any, # Job object (should have attributes like file_name, file_path, plugin_id)\n sess:Any, # FastHTML session object\n fallback_fields:Optional[Dict[str, Any]]=None # Optional additional fallback fields\n) -> Tuple[Dict[str, Any], Dict[str, Any]]: # (file_info, plugin_info) dictionaries\n \"Retrieve job-related display information from session with fallbacks.\"\n```\n\n``` python\ndef once_per_job(\n sess:Any, # FastHTML session object\n job_id:str, # Unique job identifier\n operation:Callable[[], Any], # Callable to execute (should take no arguments)\n tracker_id:str=\"default_operation\" # Unique identifier for this operation type\n) -> Optional[Any]: # Result of operation if executed, None if already processed\n \"Execute an operation once per job using automatic deduplication.\"\n```\n\n### JobSessionManager (`job_session.ipynb`)\n\n> Manage job-related metadata that needs to be accessed across multiple\n> requests.\n\n#### Import\n\n``` python\nfrom cjm_fasthtml_workflows.core.job_session import (\n JobSessionManager\n)\n```\n\n#### Classes\n\n``` python\nclass JobSessionManager:\n def __init__(\n self, \n sess:Any, # FastHTML session object\n prefix:str=\"job\" # Prefix for job keys in session\n )\n \"Manage job-related metadata that needs to be accessed across multiple requests.\"\n \n def __init__(\n self, \n sess:Any, # FastHTML session object\n prefix:str=\"job\" # Prefix for job keys in session\n )\n \"Initialize job session manager.\"\n \n def store_job_metadata(\n self, \n job_id:str, # Unique job identifier\n metadata:Dict[str, Any] # Dictionary of metadata to store\n ) -> None\n \"Store metadata for a job.\"\n \n def get_job_metadata(\n self, \n job_id:str, # Unique job identifier\n default:Optional[Dict[str, Any]]=None # Default value if metadata not found\n ) -> Dict[str, Any]: # Job metadata dictionary\n \"Retrieve metadata for a job.\"\n \n def delete_job_metadata(\n self, \n job_id:str # Unique job identifier\n ) -> None\n \"Remove metadata for a job.\"\n \n def clear_all_jobs(self) -> None:\n \"\"\"Clear all job metadata from session.\"\"\"\n keys_to_delete = [k for k in self.sess.keys() if k.startswith(f\"{self.prefix}_\")]\n for key in keys_to_delete\n \"Clear all job metadata from session.\"\n```\n\n### WorkflowSession (`workflow_session.ipynb`)\n\n> Namespaced workflow state management within FastHTML sessions.\n\n#### Import\n\n``` python\nfrom cjm_fasthtml_workflows.core.workflow_session import (\n WorkflowSession\n)\n```\n\n#### Classes\n\n``` python\nclass WorkflowSession:\n def __init__(\n self, \n sess:Any, # FastHTML session object (from route handler parameter)\n workflow_id:str # Unique identifier for this workflow (e.g., \"registration\", \"checkout\")\n )\n \"Manage workflow state within FastHTML sessions with proper namespacing.\"\n \n def __init__(\n self, \n sess:Any, # FastHTML session object (from route handler parameter)\n workflow_id:str # Unique identifier for this workflow (e.g., \"registration\", \"checkout\")\n )\n \"Initialize workflow session wrapper.\"\n \n def set(\n self, \n key:str, # Key name within this workflow namespace\n value:Any # Value to store (must be JSON-serializable for session storage)\n ) -> None\n \"Store value in workflow session.\"\n \n def get(\n self, \n key:str, # Key name within this workflow namespace\n default:Any=None # Default value if key not found\n ) -> Any: # Stored value or default\n \"Retrieve value from workflow session.\"\n \n def delete(\n self, \n key:str # Key name within this workflow namespace\n ) -> None\n \"Remove value from workflow session.\"\n \n def clear(self) -> None:\n \"\"\"Clear all data for this workflow from session.\"\"\"\n keys_to_delete = [k for k in self.sess.keys() if k.startswith(self._prefix)]\n for key in keys_to_delete\n \"Clear all data for this workflow from session.\"\n \n def get_all(self) -> Dict[str, Any]: # Dictionary mapping original keys to their values\n \"\"\"Get all workflow data as a dictionary with original (unprefixed) keys.\"\"\"\n result = {}\n prefix_len = len(self._prefix)\n for key, value in self.sess.items()\n \"Get all workflow data as a dictionary with original (unprefixed) keys.\"\n \n def has(\n self, \n key:str # Key name within this workflow namespace\n ) -> bool: # True if key exists, False otherwise\n \"Check if key exists in workflow session.\"\n```\n",
"bugtrack_url": null,
"license": "Apache Software License 2.0",
"summary": "Session-based workflow management for FastHTML applications with multi-step state tracking, job metadata storage, and operation deduplication.",
"version": "0.0.1",
"project_urls": {
"Homepage": "https://github.com/cj-mills/cjm-fasthtml-workflows"
},
"split_keywords": [
"nbdev",
"jupyter",
"notebook",
"python"
],
"urls": [
{
"comment_text": null,
"digests": {
"blake2b_256": "4d5aa160c4e9ca3d5b56bf131bf3b286309565eafd48abadb5c3ba2bc539ef41",
"md5": "3ddde8fb123a0fc5e5d40363a78c1d3a",
"sha256": "a9cf3d2a3ea6104fcc6f10c4575611a5464a72bf3bc004c7c0266b639d3bd03c"
},
"downloads": -1,
"filename": "cjm_fasthtml_workflows-0.0.1-py3-none-any.whl",
"has_sig": false,
"md5_digest": "3ddde8fb123a0fc5e5d40363a78c1d3a",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.11",
"size": 13513,
"upload_time": "2025-10-24T04:50:14",
"upload_time_iso_8601": "2025-10-24T04:50:14.511357Z",
"url": "https://files.pythonhosted.org/packages/4d/5a/a160c4e9ca3d5b56bf131bf3b286309565eafd48abadb5c3ba2bc539ef41/cjm_fasthtml_workflows-0.0.1-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": null,
"digests": {
"blake2b_256": "3245abc5e82321ee76975da04703bdff6aed8645da6f8387597ef2c5f960a273",
"md5": "467853fa51f39330cf5602fc244f85fe",
"sha256": "987eb8d0b9bdb7f10b64c100c7e6a1b578b20278276ee41b1d8e6cc4eb8958d3"
},
"downloads": -1,
"filename": "cjm_fasthtml_workflows-0.0.1.tar.gz",
"has_sig": false,
"md5_digest": "467853fa51f39330cf5602fc244f85fe",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.11",
"size": 13165,
"upload_time": "2025-10-24T04:50:15",
"upload_time_iso_8601": "2025-10-24T04:50:15.685946Z",
"url": "https://files.pythonhosted.org/packages/32/45/abc5e82321ee76975da04703bdff6aed8645da6f8387597ef2c5f960a273/cjm_fasthtml_workflows-0.0.1.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2025-10-24 04:50:15",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "cj-mills",
"github_project": "cjm-fasthtml-workflows",
"travis_ci": false,
"coveralls": false,
"github_actions": true,
"lcname": "cjm-fasthtml-workflows"
}
JSON
