src.integrations.kanban_interface module#

Common interface for all Kanban board integrations.

This abstract base class defines the standard interface that all kanban integrations (Planka, Linear, GitHub Projects) must implement.

class src.integrations.kanban_interface.KanbanProvider[source]#

Bases: Enum

Supported kanban providers.

PLANKA = 'planka'#
LINEAR = 'linear'#
GITHUB = 'github'#
SQLITE = 'sqlite'#
class src.integrations.kanban_interface.KanbanInterface[source]#

Bases: ABC

Abstract base class for kanban board integrations.

All kanban providers must implement these methods to ensure consistent behavior across different platforms.

__init__(config)[source]#

Initialize kanban provider with configuration.

Parameters:

config (Dict[str, Any]) – Provider-specific configuration - For Planka: url, username, password - For Linear: api_key, team_id - For GitHub: token, owner, repo, project_number

provider: KanbanProvider | None#
abstractmethod async connect()[source]#

Establish connection to the kanban service.

Returns:

True if connection successful

Return type:

bool

abstractmethod async disconnect()[source]#

Close connection to the kanban service.

Return type:

None

abstractmethod async get_available_tasks()[source]#

Get all unassigned tasks from backlog/ready columns.

Returns:

List of Task objects that are available for assignment

Return type:

List[Task]

abstractmethod async get_all_tasks()[source]#

Get all tasks from the board regardless of status or assignment.

Returns:

List of all Task objects on the board

Return type:

List[Task]

abstractmethod async get_task_by_id(task_id)[source]#

Get a specific task by its ID.

Parameters:

task_id (str) – The task identifier

Returns:

Task object or None if not found

Return type:

Optional[Task]

abstractmethod async create_task(task_data)[source]#

Create a new task on the board.

Parameters:

task_data (Dict[str, Any]) –

Dictionary containing:

  • name: Task title

  • description: Task description

  • priority: Priority level

  • labels: List of labels/tags

  • estimated_hours: Time estimate

Returns:

Created Task object

Return type:

Task

abstractmethod async update_task(task_id, updates)[source]#

Update an existing task.

Parameters:

  • task_id (str) – The task identifier

  • updates (Dict[str, Any]) – Dictionary of fields to update

Returns:

Updated Task object

Return type:

Optional[Task]

abstractmethod async assign_task(task_id, assignee_id)[source]#

Assign a task to a worker.

Parameters:

  • task_id (str) – The task identifier

  • assignee_id (str) – The worker/user identifier

Returns:

True if assignment successful

Return type:

bool

abstractmethod async move_task_to_column(task_id, column_name)[source]#

Move task to a specific column/status.

Parameters:

  • task_id (str) – The task identifier

  • column_name (str) – Target column (e.g., “In Progress”, “Done”)

Returns:

True if move successful

Return type:

bool

abstractmethod async add_comment(task_id, comment)[source]#

Add a comment to a task.

Parameters:

  • task_id (str) – The task identifier

  • comment (str) – Comment text

Returns:

True if comment added successfully

Return type:

bool

abstractmethod async get_project_metrics()[source]#

Get project metrics and statistics.

Returns:

Dictionary containing:

  • total_tasks: Total number of tasks

  • backlog_tasks: Tasks in backlog

  • in_progress_tasks: Tasks being worked on

  • completed_tasks: Completed tasks

  • blocked_tasks: Blocked tasks

Return type:

Dict[str, Any]

abstractmethod async report_blocker(task_id, blocker_description, severity='medium')[source]#

Report a blocker on a task.

Parameters:

  • task_id (str) – The task identifier

  • blocker_description (str) – Description of the blocker

  • severity (str) – Blocker severity (low, medium, high)

Returns:

True if blocker reported successfully

Return type:

bool

abstractmethod async update_task_progress(task_id, progress_data)[source]#

Update task progress.

Parameters:

  • task_id (str) – The task identifier

  • progress_data (Dict[str, Any]) –

    Dictionary containing:

    • progress: Percentage complete (0-100)

    • status: Current status

    • message: Progress message

Returns:

True if update successful

Return type:

bool

normalize_priority(provider_priority)[source]#

Normalize provider-specific priority to standard Priority enum.

Parameters:

provider_priority (Any) – Provider’s priority representation

Returns:

Standardized Priority enum value

Return type:

Priority

normalize_status(provider_status)[source]#

Normalize provider-specific status to standard TaskStatus enum.

Parameters:

provider_status (Any) – Provider’s status representation

Returns:

Standardized TaskStatus enum value

Return type:

TaskStatus

abstractmethod async upload_attachment(task_id, filename, content, content_type=None)[source]#

Upload an attachment to a task.

This method provides a generic interface for uploading design artifacts, documentation, and other files to tasks across different kanban providers.

Parameters:

  • task_id (str) – The task identifier (provider-specific format)

  • filename (str) – Name for the attachment

  • content (Union[str, bytes]) – File content (base64 string or bytes)

  • content_type (Optional[str]) – MIME type (e.g., ‘application/json’, ‘text/markdown’)

Returns:

Dict with attachment details including: - success: bool indicating if upload was successful - data: Dict containing:

  • id: Attachment identifier

  • filename: The filename

  • url: URL to access the attachment (if available)

  • size: File size in bytes

  • error: Error message if success is False

Return type:

Dict[str, Any]

abstractmethod async get_attachments(task_id)[source]#

Get all attachments for a task.

Parameters:

task_id (str) – The task identifier

Returns:

Dict containing: - success: bool - data: List of attachment objects with:

  • id: Attachment identifier

  • filename: The filename

  • url: URL to access the attachment

  • created_at: Creation timestamp (ISO format)

  • created_by: User who uploaded it (if available)

  • error: Error message if success is False

Return type:

Dict[str, Any]

abstractmethod async download_attachment(attachment_id, filename, task_id=None)[source]#

Download an attachment.

Parameters:

  • attachment_id (str) – The attachment ID

  • filename (str) – The filename (required by some providers)

  • task_id (Optional[str]) – Optional task ID (required by some providers)

Returns:

Dict containing: - success: bool - data: Dict with:

  • content: Base64 encoded file content

  • filename: The filename

  • content_type: MIME type if available

  • error: Error message if success is False

Return type:

Dict[str, Any]

async delete_attachment(attachment_id, task_id=None)[source]#

Delete an attachment (optional - not all providers support this).

Parameters:

  • attachment_id (str) – The attachment ID

  • task_id (Optional[str]) – Optional task ID (required by some providers)

Returns:

Dict containing: - success: bool - error: Error message if not supported or failed

Return type:

Dict[str, Any]

async update_attachment(attachment_id, filename=None, task_id=None)[source]#

Update attachment metadata (optional - not all providers support this).

Parameters:

  • attachment_id (str) – The attachment ID

  • filename (Optional[str]) – New filename

  • task_id (Optional[str]) – Optional task ID (required by some providers)

Returns:

Dict containing: - success: bool - data: Updated attachment info if successful - error: Error message if not supported or failed

Return type:

Dict[str, Any]