flowrra.ui.base package

Submodules

flowrra.ui.base.adapter module

class flowrra.ui.base.adapter.BaseUIAdapter(flowrra_app)[source]

Bases: ABC

Base adapter for framework-specific UI implementations.

Framework-specific adapters (FastAPI, Flask, Django) inherit from this class and implement get_routes() to provide their routing logic.

__init__(flowrra_app)[source]

Initialize adapter with services.

Parameters:: flowrra_app (Flowrra) – Flowrra application instance

abstractmethod get_routes()[source]

Return framework-specific routes.

Returns:

FastAPI: APIRouter
Flask: Blueprint
Django: List[URLPattern]

Return type:

Framework-specific router/blueprint/urls object

property templates_dir: Path

Get templates directory path.

Returns:: Path to templates directory

property static_dir: Path

Get static files directory path.

Returns:: Path to static files directory

async get_dashboard_data()[source]

Get data for dashboard page.

Return type:: Dict[str, Any]
Returns:: Dictionary with dashboard data

async get_tasks_page_data(status=None, limit=50)[source]

Get data for tasks page.

Parameters:

status (Optional[str]) – Filter by status (pending/running/success/failed)
limit (int) – Maximum tasks to return

Return type:

Dict[str, Any]

Returns:

Dictionary with tasks data

async get_schedules_page_data(enabled_only=False)[source]

Get data for schedules page.

Parameters:: enabled_only (bool) – If True, only return enabled schedules
Return type:: Dict[str, Any]
Returns:: Dictionary with schedules data

static format_datetime(dt)[source]

Format datetime for display.

Parameters:: dt (Any) – Datetime object or ISO string
Return type:: str
Returns:: Formatted datetime string

static format_duration(seconds)[source]

Format duration in seconds to human-readable string.

Parameters:: seconds (float) – Duration in seconds
Return type:: str
Returns:: Formatted duration (e.g., “2h 30m”, “45s”)

static get_status_color(status)[source]

Get color class for task status.

Parameters:: status (str) – Task status string
Return type:: str
Returns:: CSS color class name

async get_stats()[source]

API endpoint: Get system statistics.

Return type:: Dict[str, Any]
Returns:: System statistics dictionary

async health_check()[source]

API endpoint: Health check.

Return type:: Dict[str, Any]
Returns:: Health status dictionary

async list_tasks()[source]

API endpoint: List registered tasks.

Return type:: List[Dict[str, Any]]
Returns:: List of registered tasks

async get_task(task_name)[source]

API endpoint: Get specific task info.

Parameters:: task_name (str) – Task name
Return type:: Optional[Dict[str, Any]]
Returns:: Task information or None

async list_schedules(enabled_only=False)[source]

API endpoint: List schedules.

Parameters:: enabled_only (bool) – If True, only return enabled schedules
Return type:: List[Dict[str, Any]]
Returns:: List of schedules

async get_schedule(schedule_id)[source]

API endpoint: Get specific schedule.

Parameters:: schedule_id (str) – Schedule ID
Return type:: Optional[Dict[str, Any]]
Returns:: Schedule details or None

async create_schedule_cron(task_name, cron, args=(), kwargs=None, enabled=True, description=None, priority=0)[source]

API endpoint: Create cron schedule.

Parameters:

task_name (str) – Task name
cron (str) – Cron expression
args (tuple) – Task arguments
kwargs (Optional[dict]) – Task keyword arguments
enabled (bool) – Start enabled
description (Optional[str]) – Optional description
priority (int) – Task priority

Return type:

Dict[str, str]

Returns:

Dictionary with schedule_id

async enable_schedule(schedule_id)[source]

API endpoint: Enable schedule.

Parameters:: schedule_id (str) – Schedule ID
Return type:: Dict[str, str]
Returns:: Success message

async disable_schedule(schedule_id)[source]

API endpoint: Disable schedule.

Parameters:: schedule_id (str) – Schedule ID
Return type:: Dict[str, str]
Returns:: Success message

async delete_schedule(schedule_id)[source]

API endpoint: Delete schedule.

Parameters:: schedule_id (str) – Schedule ID
Return type:: Dict[str, Any]
Returns:: Success status

flowrra.ui.base.formatter module

Formatting utilities for Flowrra UI.

class flowrra.ui.base.formatter.Formatter[source]

Bases: object

Static utility class for formatting values in the UI.

Provides consistent formatting across all UI adapters: - Datetime formatting - Duration formatting - Status color mapping

static format_datetime(dt)[source]

Format datetime for display.

Parameters:: dt (Any) – Datetime object or ISO string
Return type:: str
Returns:: Formatted datetime string

static format_duration(seconds)[source]

Format duration in seconds to human-readable string.

Parameters:: seconds (float) – Duration in seconds
Return type:: str
Returns:: Formatted duration (e.g., “2h 30m”, “45s”)

static get_status_color(status)[source]

Get color class for task status.

Parameters:: status (str) – Task status string
Return type:: str
Returns:: CSS color class name

flowrra.ui.base.schedule_service module

Service for schedule management operations.

class flowrra.ui.base.schedule_service.ScheduleService(manager)[source]

Bases: object

Service for managing schedules via the UI.

Provides schedule CRUD operations: - List schedules - Get schedule details - Create schedules - Enable/disable schedules - Delete schedules

This service wraps FlowrraManager schedule operations and provides a consistent API for UI adapters.

__init__(manager)[source]

Initialize schedule service.

Parameters:: manager (FlowrraManager) – FlowrraManager instance for schedule operations

async list_schedules(enabled_only=False)[source]

List schedules.

Parameters:: enabled_only (bool) – If True, only return enabled schedules
Return type:: List[Dict[str, Any]]
Returns:: List of schedules

async get_schedule(schedule_id)[source]

Get specific schedule.

Parameters:: schedule_id (str) – Schedule ID
Return type:: Optional[Dict[str, Any]]
Returns:: Schedule details or None

async create_schedule_cron(task_name, cron, args=(), kwargs=None, enabled=True, description=None, priority=0)[source]

Create cron schedule.

Parameters:

task_name (str) – Task name
cron (str) – Cron expression
args (tuple) – Task arguments
kwargs (Optional[dict]) – Task keyword arguments
enabled (bool) – Start enabled
description (Optional[str]) – Optional description
priority (int) – Task priority

Return type:

Dict[str, str]

Returns:

Dictionary with schedule_id

async enable_schedule(schedule_id)[source]

Enable schedule.

Parameters:: schedule_id (str) – Schedule ID
Return type:: Dict[str, str]
Returns:: Success message

async disable_schedule(schedule_id)[source]

Disable schedule.

Parameters:: schedule_id (str) – Schedule ID
Return type:: Dict[str, str]
Returns:: Success message

async delete_schedule(schedule_id)[source]

Delete schedule.

Parameters:: schedule_id (str) – Schedule ID
Return type:: Dict[str, Any]
Returns:: Success status

flowrra.ui.base.ui_service module

Service for retrieving UI page data.

class flowrra.ui.base.ui_service.UIService(manager)[source]

Bases: object

Service for preparing data for UI pages.

Handles data aggregation and transformation for: - Dashboard page - Tasks page - Schedules page

This service acts as a bridge between FlowrraManager (raw data) and UI templates (formatted data).

__init__(manager)[source]

Initialize UI service.

Parameters:: manager (FlowrraManager) – FlowrraManager instance for data access

async get_dashboard_data()[source]

Get data for dashboard page.

Returns:

{: “stats”: {…}, “recent_failed_tasks”: […], “schedules”: […], “total_schedules”: int

}

Return type:

Dictionary with dashboard data

async get_tasks_page_data(status=None, limit=200)[source]

Get data for tasks page.

Parameters:

status (Optional[str]) – Filter by status (pending/running/success/failed)
limit (int) – Maximum tasks to return (default: 200)

Return type:

Dict[str, Any]

Returns:

Dictionary with tasks data

async get_schedules_page_data(enabled_only=False)[source]

Get data for schedules page.

Parameters:: enabled_only (bool) – If True, only return enabled schedules
Return type:: Dict[str, Any]
Returns:: Dictionary with schedules data

Module contents

Base classes for Flowrra UI adapters.

This module provides the building blocks for creating framework-specific UI adapters for Flowrra. It includes:

BaseUIAdapter: Abstract adapter class for framework integration
UIService: Service for UI page data aggregation
ScheduleService: Service for schedule management operations
Formatter: Static utility class for formatting values

These classes follow the Single Responsibility Principle and can be composed to create adapters for different web frameworks (FastAPI, Flask, Django, etc.).

class flowrra.ui.base.BaseUIAdapter(flowrra_app)[source]

Bases: ABC

Base adapter for framework-specific UI implementations.

Framework-specific adapters (FastAPI, Flask, Django) inherit from this class and implement get_routes() to provide their routing logic.

__init__(flowrra_app)[source]

Initialize adapter with services.

Parameters:: flowrra_app (Flowrra) – Flowrra application instance

abstractmethod get_routes()[source]

Return framework-specific routes.

Returns:

FastAPI: APIRouter
Flask: Blueprint
Django: List[URLPattern]

Return type:

Framework-specific router/blueprint/urls object

property templates_dir: Path

Get templates directory path.

Returns:: Path to templates directory

property static_dir: Path

Get static files directory path.

Returns:: Path to static files directory

async get_dashboard_data()[source]

Get data for dashboard page.

Return type:: Dict[str, Any]
Returns:: Dictionary with dashboard data

async get_tasks_page_data(status=None, limit=50)[source]

Get data for tasks page.

Parameters:

status (Optional[str]) – Filter by status (pending/running/success/failed)
limit (int) – Maximum tasks to return

Return type:

Dict[str, Any]

Returns:

Dictionary with tasks data

async get_schedules_page_data(enabled_only=False)[source]

Get data for schedules page.

Parameters:: enabled_only (bool) – If True, only return enabled schedules
Return type:: Dict[str, Any]
Returns:: Dictionary with schedules data

static format_datetime(dt)[source]

Format datetime for display.

Parameters:: dt (Any) – Datetime object or ISO string
Return type:: str
Returns:: Formatted datetime string

static format_duration(seconds)[source]

Format duration in seconds to human-readable string.

Parameters:: seconds (float) – Duration in seconds
Return type:: str
Returns:: Formatted duration (e.g., “2h 30m”, “45s”)

static get_status_color(status)[source]

Get color class for task status.

Parameters:: status (str) – Task status string
Return type:: str
Returns:: CSS color class name

async get_stats()[source]

API endpoint: Get system statistics.

Return type:: Dict[str, Any]
Returns:: System statistics dictionary

async health_check()[source]

API endpoint: Health check.

Return type:: Dict[str, Any]
Returns:: Health status dictionary

async list_tasks()[source]

API endpoint: List registered tasks.

Return type:: List[Dict[str, Any]]
Returns:: List of registered tasks

async get_task(task_name)[source]

API endpoint: Get specific task info.

Parameters:: task_name (str) – Task name
Return type:: Optional[Dict[str, Any]]
Returns:: Task information or None

async list_schedules(enabled_only=False)[source]

API endpoint: List schedules.

Parameters:: enabled_only (bool) – If True, only return enabled schedules
Return type:: List[Dict[str, Any]]
Returns:: List of schedules

async get_schedule(schedule_id)[source]

API endpoint: Get specific schedule.

Parameters:: schedule_id (str) – Schedule ID
Return type:: Optional[Dict[str, Any]]
Returns:: Schedule details or None

async create_schedule_cron(task_name, cron, args=(), kwargs=None, enabled=True, description=None, priority=0)[source]

API endpoint: Create cron schedule.

Parameters:

task_name (str) – Task name
cron (str) – Cron expression
args (tuple) – Task arguments
kwargs (Optional[dict]) – Task keyword arguments
enabled (bool) – Start enabled
description (Optional[str]) – Optional description
priority (int) – Task priority

Return type:

Dict[str, str]

Returns:

Dictionary with schedule_id

async enable_schedule(schedule_id)[source]

API endpoint: Enable schedule.

Parameters:: schedule_id (str) – Schedule ID
Return type:: Dict[str, str]
Returns:: Success message

async disable_schedule(schedule_id)[source]

API endpoint: Disable schedule.

Parameters:: schedule_id (str) – Schedule ID
Return type:: Dict[str, str]
Returns:: Success message

async delete_schedule(schedule_id)[source]

API endpoint: Delete schedule.

Parameters:: schedule_id (str) – Schedule ID
Return type:: Dict[str, Any]
Returns:: Success status

class flowrra.ui.base.Formatter[source]

Bases: object

Static utility class for formatting values in the UI.

Provides consistent formatting across all UI adapters: - Datetime formatting - Duration formatting - Status color mapping

static format_datetime(dt)[source]

Format datetime for display.

Parameters:: dt (Any) – Datetime object or ISO string
Return type:: str
Returns:: Formatted datetime string

static format_duration(seconds)[source]

Format duration in seconds to human-readable string.

Parameters:: seconds (float) – Duration in seconds
Return type:: str
Returns:: Formatted duration (e.g., “2h 30m”, “45s”)

static get_status_color(status)[source]

Get color class for task status.

Parameters:: status (str) – Task status string
Return type:: str
Returns:: CSS color class name

class flowrra.ui.base.UIService(manager)[source]

Bases: object

Service for preparing data for UI pages.

Handles data aggregation and transformation for: - Dashboard page - Tasks page - Schedules page

This service acts as a bridge between FlowrraManager (raw data) and UI templates (formatted data).

__init__(manager)[source]

Initialize UI service.

Parameters:: manager (FlowrraManager) – FlowrraManager instance for data access

async get_dashboard_data()[source]

Get data for dashboard page.

Returns:

{: “stats”: {…}, “recent_failed_tasks”: […], “schedules”: […], “total_schedules”: int

}

Return type:

Dictionary with dashboard data

async get_tasks_page_data(status=None, limit=200)[source]

Get data for tasks page.

Parameters:

status (Optional[str]) – Filter by status (pending/running/success/failed)
limit (int) – Maximum tasks to return (default: 200)

Return type:

Dict[str, Any]

Returns:

Dictionary with tasks data

async get_schedules_page_data(enabled_only=False)[source]

Get data for schedules page.

Parameters:: enabled_only (bool) – If True, only return enabled schedules
Return type:: Dict[str, Any]
Returns:: Dictionary with schedules data

class flowrra.ui.base.ScheduleService(manager)[source]

Bases: object

Service for managing schedules via the UI.

Provides schedule CRUD operations: - List schedules - Get schedule details - Create schedules - Enable/disable schedules - Delete schedules

This service wraps FlowrraManager schedule operations and provides a consistent API for UI adapters.

__init__(manager)[source]

Initialize schedule service.

Parameters:: manager (FlowrraManager) – FlowrraManager instance for schedule operations

async list_schedules(enabled_only=False)[source]

List schedules.

Parameters:: enabled_only (bool) – If True, only return enabled schedules
Return type:: List[Dict[str, Any]]
Returns:: List of schedules

async get_schedule(schedule_id)[source]

Get specific schedule.

Parameters:: schedule_id (str) – Schedule ID
Return type:: Optional[Dict[str, Any]]
Returns:: Schedule details or None

async create_schedule_cron(task_name, cron, args=(), kwargs=None, enabled=True, description=None, priority=0)[source]

Create cron schedule.

Parameters:

task_name (str) – Task name
cron (str) – Cron expression
args (tuple) – Task arguments
kwargs (Optional[dict]) – Task keyword arguments
enabled (bool) – Start enabled
description (Optional[str]) – Optional description
priority (int) – Task priority

Return type:

Dict[str, str]

Returns:

Dictionary with schedule_id

async enable_schedule(schedule_id)[source]

Enable schedule.

Parameters:: schedule_id (str) – Schedule ID
Return type:: Dict[str, str]
Returns:: Success message

async disable_schedule(schedule_id)[source]

Disable schedule.

Parameters:: schedule_id (str) – Schedule ID
Return type:: Dict[str, str]
Returns:: Success message

async delete_schedule(schedule_id)[source]

Delete schedule.

Parameters:: schedule_id (str) – Schedule ID
Return type:: Dict[str, Any]
Returns:: Success status