"""Pydantic models for task aggregation.
Provides typed models for todos, tags, and external metadata with
validation and serialization support.
"""
from __future__ import annotations
from datetime import datetime
from enum import Enum
from typing import Any
from uuid import UUID, uuid4
from pydantic import BaseModel, Field
class TodoStatus(str, Enum):
"""Task status with extended values for external systems."""
TODO = "todo"
IN_PROGRESS = "in_progress"
DONE = "done"
BLOCKED = "blocked"
IN_REVIEW = "in_review"
CANCELLED = "cancelled"
class TodoSource(str, Enum):
"""Source of task creation."""
LOCAL = "local"
JIRA = "jira"
GITHUB = "github"
LINEAR = "linear"
MARKDOWN = "markdown"
STM = "stm"
class Tag(BaseModel):
"""Tag for categorizing todos."""
id: int | None = None
name: str
created_at: datetime = Field(default_factory=datetime.now)
model_config = {"from_attributes": True}
class JiraMetadata(BaseModel):
"""Jira-specific metadata for a task."""
key: str
project_key: str
issue_type: str
reporter: str | None = None
assignee: str | None = None
labels: list[str] = Field(default_factory=list)
components: list[str] = Field(default_factory=list)
sprint: str | None = None
parent_key: str | None = None # For subtask flattening
model_config = {"from_attributes": True}
class GitHubMetadata(BaseModel):
"""GitHub-specific metadata for a task."""
repo_full_name: str
number: int
issue_type: str # "issue" or "pr"
author: str | None = None
assignees: list[str] = Field(default_factory=list)
labels: list[str] = Field(default_factory=list)
milestone: str | None = None
review_requested: bool = False
draft: bool = False
model_config = {"from_attributes": True}
class LinearMetadata(BaseModel):
"""Linear-specific metadata for a task."""
identifier: str
team_key: str | None = None
project_name: str | None = None
creator: str | None = None
assignee: str | None = None
labels: list[str] = Field(default_factory=list)
cycle: str | None = None
model_config = {"from_attributes": True}
class MarkdownMetadata(BaseModel):
"""Markdown file-specific metadata for a task.
Stores information about the source markdown file and
the task's position within it.
"""
file_path: str
line_number: int
checkbox_state: str # "[ ]", "[x]", "[-]"
indent_level: int = 0
parent_heading: str | None = None
raw_line: str
model_config = {"from_attributes": True}
class StmMetadata(BaseModel):
"""STM (Simple Task Master) specific metadata for a task.
Stores information from the stm CLI tool.
"""
task_id: str
workspace_path: str
created_at: str | None = None
updated_at: str | None = None
completed_at: str | None = None
tags: list[str] = Field(default_factory=list)
model_config = {"from_attributes": True}
class ExternalTaskMetadata(BaseModel):
"""Container for external system metadata.
Supports optional system-specific fields and stores raw API responses
for debugging and future field mapping.
"""
jira: JiraMetadata | None = None
github: GitHubMetadata | None = None
linear: LinearMetadata | None = None
markdown: MarkdownMetadata | None = None
stm: StmMetadata | None = None
raw_response: dict[str, Any] | None = None
fetched_at: datetime | None = None
model_config = {"from_attributes": True}
class Todo(BaseModel):
"""Task item with external system integration support.
Extends the joecc Todo pattern with fields for tracking external
source systems, sync state, and external metadata.
"""
id: int | None = None
uuid: UUID = Field(default_factory=uuid4)
content: str
status: TodoStatus = TodoStatus.TODO
priority: int = 0
due_date: datetime | None = None
completed_at: datetime | None = None
created_at: datetime = Field(default_factory=datetime.now)
updated_at: datetime = Field(default_factory=datetime.now)
source_system: TodoSource = TodoSource.LOCAL
source_id: str | None = None # External system ID (e.g., Jira key)
source_url: str | None = None # Link to external task
external_metadata: ExternalTaskMetadata | None = None
last_synced_at: datetime | None = None
sync_hash: str | None = None # Hash for change detection
tags: list[Tag] = Field(default_factory=list)
model_config = {"from_attributes": True}
@classmethod
def from_row(cls, row: Any, tags: list[Tag] | None = None) -> Todo:
"""Create Todo from database row.
Args:
row: SQLite Row object or dictionary.
tags: Optional list of associated tags.
Returns:
Todo instance.
"""
data = dict(row) if hasattr(row, "keys") else row
datetime_fields = [
"due_date",
"completed_at",
"created_at",
"updated_at",
"last_synced_at",
]
for field in datetime_fields:
if data.get(field) and isinstance(data[field], str):
data[field] = datetime.fromisoformat(data[field])
if data.get("uuid") and isinstance(data["uuid"], str):
data["uuid"] = UUID(data["uuid"])
if data.get("status") and isinstance(data["status"], str):
data["status"] = TodoStatus(data["status"])
if data.get("source_system") and isinstance(data["source_system"], str):
data["source_system"] = TodoSource(data["source_system"])
if data.get("external_metadata"):
if isinstance(data["external_metadata"], str):
data["external_metadata"] = ExternalTaskMetadata.model_validate_json(data["external_metadata"])
elif isinstance(data["external_metadata"], dict):
data["external_metadata"] = ExternalTaskMetadata.model_validate(data["external_metadata"])
data["tags"] = tags or []
return cls(**data)
def to_dict(self) -> dict[str, Any]:
"""Convert to dictionary for database storage.
Returns:
Dictionary with string-serialized values suitable for SQLite.
"""
data = self.model_dump(exclude={"tags", "external_metadata"})
data["uuid"] = str(data["uuid"])
datetime_fields = [
"due_date",
"completed_at",
"created_at",
"updated_at",
"last_synced_at",
]
for field in datetime_fields:
if data.get(field):
data[field] = data[field].isoformat()
data["status"] = data["status"].value if data["status"] else None
data["source_system"] = data["source_system"].value if data["source_system"] else None
if self.external_metadata:
data["external_metadata"] = self.external_metadata.model_dump_json()
else:
data["external_metadata"] = None
return data
