Models

models

Data models for beads-ClickUp integration.

This module provides type-safe dataclasses for the core domain objects used throughout the integration.

ClickUpTask dataclass

Represents a ClickUp task.

This is the canonical representation of a task as returned by the ClickUp API.

Source code in beads_clickup/models.py

@dataclass
class ClickUpTask:
    """Represents a ClickUp task.

    This is the canonical representation of a task as returned by the ClickUp API.
    """
    id: str
    name: str
    status: str
    priority: Optional[int] = None  # 1=Urgent, 2=High, 3=Normal, 4=Low
    description: Optional[str] = None
    tags: list[str] = field(default_factory=list)
    assignees: list[str] = field(default_factory=list)
    parent_id: Optional[str] = None
    list_id: Optional[str] = None
    folder_id: Optional[str] = None
    space_id: Optional[str] = None
    url: Optional[str] = None
    start_date: Optional[datetime] = None
    due_date: Optional[datetime] = None
    time_estimate: Optional[int] = None  # milliseconds
    date_created: Optional[datetime] = None
    date_updated: Optional[datetime] = None
    date_closed: Optional[datetime] = None
    custom_fields: list[dict[str, Any]] = field(default_factory=list)

    @classmethod
    def from_api_response(cls, data: dict[str, Any]) -> "ClickUpTask":
        """Create a ClickUpTask from API response data.

        Args:
            data: Dictionary from ClickUp API

        Returns:
            ClickUpTask instance
        """
        # Extract status
        status_obj = data.get("status", {})
        status = status_obj.get("status", "") if isinstance(status_obj, dict) else str(status_obj)

        # Extract priority
        priority_obj = data.get("priority")
        priority = None
        if priority_obj:
            if isinstance(priority_obj, dict):
                priority = int(priority_obj.get("id", 0)) if priority_obj.get("id") else None
            else:
                priority = int(priority_obj) if priority_obj else None

        # Extract tags
        tags_raw = data.get("tags", [])
        tags = []
        for tag in tags_raw:
            if isinstance(tag, dict):
                tags.append(tag.get("name", ""))
            else:
                tags.append(str(tag))

        # Extract assignees
        assignees_raw = data.get("assignees", [])
        assignees = []
        for assignee in assignees_raw:
            if isinstance(assignee, dict):
                assignees.append(assignee.get("id", ""))
            else:
                assignees.append(str(assignee))

        # Extract parent ID
        parent = data.get("parent")
        parent_id = None
        if parent:
            parent_id = parent.get("id") if isinstance(parent, dict) else str(parent)

        # Parse timestamps (ClickUp uses milliseconds)
        def parse_ms_timestamp(value: Any) -> Optional[datetime]:
            if value is None:
                return None
            try:
                if isinstance(value, str):
                    value = int(value)
                return datetime.utcfromtimestamp(value / 1000)
            except (ValueError, TypeError):
                return None

        return cls(
            id=data.get("id", ""),
            name=data.get("name", ""),
            status=status,
            priority=priority,
            description=data.get("description"),
            tags=tags,
            assignees=assignees,
            parent_id=parent_id,
            list_id=data.get("list", {}).get("id") if isinstance(data.get("list"), dict) else data.get("list"),
            folder_id=data.get("folder", {}).get("id") if isinstance(data.get("folder"), dict) else data.get("folder"),
            space_id=data.get("space", {}).get("id") if isinstance(data.get("space"), dict) else data.get("space"),
            url=data.get("url"),
            start_date=parse_ms_timestamp(data.get("start_date")),
            due_date=parse_ms_timestamp(data.get("due_date")),
            time_estimate=data.get("time_estimate"),
            date_created=parse_ms_timestamp(data.get("date_created")),
            date_updated=parse_ms_timestamp(data.get("date_updated")),
            date_closed=parse_ms_timestamp(data.get("date_closed")),
            custom_fields=data.get("custom_fields", []),
        )

    @property
    def task_url(self) -> str:
        """Get the ClickUp URL for this task."""
        if self.url:
            return self.url
        return f"https://app.clickup.com/t/{self.id}"

task_url property

Get the ClickUp URL for this task.

from_api_response(data) classmethod

Create a ClickUpTask from API response data.

Parameters:

Name	Type	Description	Default
data	dict[str, Any]	Dictionary from ClickUp API	required

Returns:

Type	Description
ClickUpTask	ClickUpTask instance

Source code in beads_clickup/models.py

@classmethod
def from_api_response(cls, data: dict[str, Any]) -> "ClickUpTask":
    """Create a ClickUpTask from API response data.

    Args:
        data: Dictionary from ClickUp API

    Returns:
        ClickUpTask instance
    """
    # Extract status
    status_obj = data.get("status", {})
    status = status_obj.get("status", "") if isinstance(status_obj, dict) else str(status_obj)

    # Extract priority
    priority_obj = data.get("priority")
    priority = None
    if priority_obj:
        if isinstance(priority_obj, dict):
            priority = int(priority_obj.get("id", 0)) if priority_obj.get("id") else None
        else:
            priority = int(priority_obj) if priority_obj else None

    # Extract tags
    tags_raw = data.get("tags", [])
    tags = []
    for tag in tags_raw:
        if isinstance(tag, dict):
            tags.append(tag.get("name", ""))
        else:
            tags.append(str(tag))

    # Extract assignees
    assignees_raw = data.get("assignees", [])
    assignees = []
    for assignee in assignees_raw:
        if isinstance(assignee, dict):
            assignees.append(assignee.get("id", ""))
        else:
            assignees.append(str(assignee))

    # Extract parent ID
    parent = data.get("parent")
    parent_id = None
    if parent:
        parent_id = parent.get("id") if isinstance(parent, dict) else str(parent)

    # Parse timestamps (ClickUp uses milliseconds)
    def parse_ms_timestamp(value: Any) -> Optional[datetime]:
        if value is None:
            return None
        try:
            if isinstance(value, str):
                value = int(value)
            return datetime.utcfromtimestamp(value / 1000)
        except (ValueError, TypeError):
            return None

    return cls(
        id=data.get("id", ""),
        name=data.get("name", ""),
        status=status,
        priority=priority,
        description=data.get("description"),
        tags=tags,
        assignees=assignees,
        parent_id=parent_id,
        list_id=data.get("list", {}).get("id") if isinstance(data.get("list"), dict) else data.get("list"),
        folder_id=data.get("folder", {}).get("id") if isinstance(data.get("folder"), dict) else data.get("folder"),
        space_id=data.get("space", {}).get("id") if isinstance(data.get("space"), dict) else data.get("space"),
        url=data.get("url"),
        start_date=parse_ms_timestamp(data.get("start_date")),
        due_date=parse_ms_timestamp(data.get("due_date")),
        time_estimate=data.get("time_estimate"),
        date_created=parse_ms_timestamp(data.get("date_created")),
        date_updated=parse_ms_timestamp(data.get("date_updated")),
        date_closed=parse_ms_timestamp(data.get("date_closed")),
        custom_fields=data.get("custom_fields", []),
    )

Comment dataclass

Represents a comment on an issue/task.

Source code in beads_clickup/models.py

@dataclass
class Comment:
    """Represents a comment on an issue/task."""
    id: str
    text: str
    author: str
    source: str  # "beads" or "clickup"
    created_at: datetime
    synced_to_clickup: bool = False
    clickup_comment_id: Optional[str] = None

    @classmethod
    def from_dict(cls, data: dict[str, Any]) -> "Comment":
        """Create Comment from a dictionary."""
        created_at = data.get("created_at")
        if isinstance(created_at, str):
            created_at = datetime.fromisoformat(created_at.replace("Z", "+00:00"))
        elif created_at is None:
            created_at = datetime.utcnow()

        return cls(
            id=data.get("id", ""),
            text=data.get("text", ""),
            author=data.get("author", "system"),
            source=data.get("source", "beads"),
            created_at=created_at,
            synced_to_clickup=data.get("synced_to_clickup", False),
            clickup_comment_id=data.get("clickup_comment_id"),
        )

    def to_dict(self) -> dict[str, Any]:
        """Convert to dictionary for JSON serialization."""
        result = {
            "id": self.id,
            "text": self.text,
            "author": self.author,
            "source": self.source,
            "created_at": self.created_at.isoformat(),
            "synced_to_clickup": self.synced_to_clickup,
        }
        if self.clickup_comment_id:
            result["clickup_comment_id"] = self.clickup_comment_id
        return result

from_dict(data) classmethod

Create Comment from a dictionary.

Source code in beads_clickup/models.py

@classmethod
def from_dict(cls, data: dict[str, Any]) -> "Comment":
    """Create Comment from a dictionary."""
    created_at = data.get("created_at")
    if isinstance(created_at, str):
        created_at = datetime.fromisoformat(created_at.replace("Z", "+00:00"))
    elif created_at is None:
        created_at = datetime.utcnow()

    return cls(
        id=data.get("id", ""),
        text=data.get("text", ""),
        author=data.get("author", "system"),
        source=data.get("source", "beads"),
        created_at=created_at,
        synced_to_clickup=data.get("synced_to_clickup", False),
        clickup_comment_id=data.get("clickup_comment_id"),
    )

to_dict()

Convert to dictionary for JSON serialization.

Source code in beads_clickup/models.py

def to_dict(self) -> dict[str, Any]:
    """Convert to dictionary for JSON serialization."""
    result = {
        "id": self.id,
        "text": self.text,
        "author": self.author,
        "source": self.source,
        "created_at": self.created_at.isoformat(),
        "synced_to_clickup": self.synced_to_clickup,
    }
    if self.clickup_comment_id:
        result["clickup_comment_id"] = self.clickup_comment_id
    return result

CustomFieldConfig dataclass

Configuration for a custom field mapping.

Source code in beads_clickup/models.py

@dataclass
class CustomFieldConfig:
    """Configuration for a custom field mapping."""
    name: str
    clickup_field_id: str
    field_type: str
    required: bool = False
    default: Optional[Any] = None
    options: dict[str, str] = field(default_factory=dict)  # For dropdowns: name -> id

    @classmethod
    def from_dict(cls, name: str, data: dict[str, Any]) -> "CustomFieldConfig":
        """Create CustomFieldConfig from configuration dictionary."""
        return cls(
            name=name,
            clickup_field_id=data.get("clickup_field_id", ""),
            field_type=data.get("type", "text"),
            required=data.get("required", False),
            default=data.get("default"),
            options=data.get("options", {}),
        )

from_dict(name, data) classmethod

Create CustomFieldConfig from configuration dictionary.

Source code in beads_clickup/models.py

@classmethod
def from_dict(cls, name: str, data: dict[str, Any]) -> "CustomFieldConfig":
    """Create CustomFieldConfig from configuration dictionary."""
    return cls(
        name=name,
        clickup_field_id=data.get("clickup_field_id", ""),
        field_type=data.get("type", "text"),
        required=data.get("required", False),
        default=data.get("default"),
        options=data.get("options", {}),
    )

Issue dataclass

Represents a beads issue.

This is the canonical representation of an issue in the beads system.

Source code in beads_clickup/models.py

@dataclass
class Issue:
    """Represents a beads issue.

    This is the canonical representation of an issue in the beads system.
    """
    id: str
    title: str
    status: IssueStatus = IssueStatus.OPEN
    priority: Priority = Priority.P2
    issue_type: IssueType = IssueType.TASK
    description: Optional[str] = None
    labels: list[str] = field(default_factory=list)
    created_at: Optional[datetime] = None
    updated_at: Optional[datetime] = None
    closed_at: Optional[datetime] = None
    close_reason: Optional[str] = None
    external_ref: Optional[str] = None
    custom_fields: dict[str, Any] = field(default_factory=dict)

    @classmethod
    def from_dict(cls, data: dict[str, Any]) -> "Issue":
        """Create an Issue from a dictionary (e.g., from bd CLI JSON output).

        Args:
            data: Dictionary with issue data

        Returns:
            Issue instance
        """
        # Parse status
        status_str = data.get("status", "open")
        try:
            status = IssueStatus.from_string(status_str)
        except ValueError:
            status = IssueStatus.OPEN

        # Parse priority
        priority_str = data.get("priority", "p2")
        try:
            priority = Priority.from_string(str(priority_str))
        except ValueError:
            priority = Priority.P2

        # Parse issue type
        type_str = data.get("type", "task")
        try:
            issue_type = IssueType.from_string(type_str)
        except ValueError:
            issue_type = IssueType.TASK

        # Parse timestamps
        created_at = cls._parse_datetime(data.get("created_at"))
        updated_at = cls._parse_datetime(data.get("updated_at"))
        closed_at = cls._parse_datetime(data.get("closed_at"))

        return cls(
            id=data.get("id", ""),
            title=data.get("title", "Untitled"),
            status=status,
            priority=priority,
            issue_type=issue_type,
            description=data.get("description"),
            labels=data.get("labels", []),
            created_at=created_at,
            updated_at=updated_at,
            closed_at=closed_at,
            close_reason=data.get("close_reason"),
            external_ref=data.get("external_ref"),
            custom_fields=data.get("custom_fields", {}),
        )

    @staticmethod
    def _parse_datetime(value: Any) -> Optional[datetime]:
        """Parse a datetime value from various formats."""
        if value is None:
            return None
        if isinstance(value, datetime):
            return value
        if isinstance(value, str):
            # Try ISO format
            try:
                # Handle 'Z' suffix
                value = value.replace("Z", "+00:00")
                return datetime.fromisoformat(value)
            except ValueError:
                pass
            # Try other common formats
            for fmt in ["%Y-%m-%d %H:%M:%S", "%Y-%m-%d"]:
                try:
                    return datetime.strptime(value, fmt)
                except ValueError:
                    continue
        return None

    def to_dict(self) -> dict[str, Any]:
        """Convert to dictionary for JSON serialization."""
        result = {
            "id": self.id,
            "title": self.title,
            "status": self.status.value,
            "priority": self.priority.value,
            "type": self.issue_type.value,
        }
        if self.description:
            result["description"] = self.description
        if self.labels:
            result["labels"] = self.labels
        if self.created_at:
            result["created_at"] = self.created_at.isoformat()
        if self.updated_at:
            result["updated_at"] = self.updated_at.isoformat()
        if self.closed_at:
            result["closed_at"] = self.closed_at.isoformat()
        if self.close_reason:
            result["close_reason"] = self.close_reason
        if self.external_ref:
            result["external_ref"] = self.external_ref
        if self.custom_fields:
            result["custom_fields"] = self.custom_fields
        return result

    @property
    def clickup_task_id(self) -> Optional[str]:
        """Extract ClickUp task ID from external_ref if present."""
        if self.external_ref and self.external_ref.startswith("clickup-"):
            return self.external_ref.replace("clickup-", "")
        return None

clickup_task_id property

Extract ClickUp task ID from external_ref if present.

from_dict(data) classmethod

Create an Issue from a dictionary (e.g., from bd CLI JSON output).

Parameters:

Name	Type	Description	Default
data	dict[str, Any]	Dictionary with issue data	required

Returns:

Type	Description
Issue	Issue instance

Source code in beads_clickup/models.py

@classmethod
def from_dict(cls, data: dict[str, Any]) -> "Issue":
    """Create an Issue from a dictionary (e.g., from bd CLI JSON output).

    Args:
        data: Dictionary with issue data

    Returns:
        Issue instance
    """
    # Parse status
    status_str = data.get("status", "open")
    try:
        status = IssueStatus.from_string(status_str)
    except ValueError:
        status = IssueStatus.OPEN

    # Parse priority
    priority_str = data.get("priority", "p2")
    try:
        priority = Priority.from_string(str(priority_str))
    except ValueError:
        priority = Priority.P2

    # Parse issue type
    type_str = data.get("type", "task")
    try:
        issue_type = IssueType.from_string(type_str)
    except ValueError:
        issue_type = IssueType.TASK

    # Parse timestamps
    created_at = cls._parse_datetime(data.get("created_at"))
    updated_at = cls._parse_datetime(data.get("updated_at"))
    closed_at = cls._parse_datetime(data.get("closed_at"))

    return cls(
        id=data.get("id", ""),
        title=data.get("title", "Untitled"),
        status=status,
        priority=priority,
        issue_type=issue_type,
        description=data.get("description"),
        labels=data.get("labels", []),
        created_at=created_at,
        updated_at=updated_at,
        closed_at=closed_at,
        close_reason=data.get("close_reason"),
        external_ref=data.get("external_ref"),
        custom_fields=data.get("custom_fields", {}),
    )

to_dict()

Convert to dictionary for JSON serialization.

Source code in beads_clickup/models.py

def to_dict(self) -> dict[str, Any]:
    """Convert to dictionary for JSON serialization."""
    result = {
        "id": self.id,
        "title": self.title,
        "status": self.status.value,
        "priority": self.priority.value,
        "type": self.issue_type.value,
    }
    if self.description:
        result["description"] = self.description
    if self.labels:
        result["labels"] = self.labels
    if self.created_at:
        result["created_at"] = self.created_at.isoformat()
    if self.updated_at:
        result["updated_at"] = self.updated_at.isoformat()
    if self.closed_at:
        result["closed_at"] = self.closed_at.isoformat()
    if self.close_reason:
        result["close_reason"] = self.close_reason
    if self.external_ref:
        result["external_ref"] = self.external_ref
    if self.custom_fields:
        result["custom_fields"] = self.custom_fields
    return result

SyncState dataclass

Represents the sync state for a single issue.

Tracks the mapping between beads issues and ClickUp tasks, along with timestamps for conflict detection.

Source code in beads_clickup/models.py

@dataclass
class SyncState:
    """Represents the sync state for a single issue.

    Tracks the mapping between beads issues and ClickUp tasks,
    along with timestamps for conflict detection.
    """
    issue_id: str
    clickup_task_id: Optional[str] = None
    last_synced: Optional[datetime] = None
    last_beads_update: Optional[datetime] = None
    last_clickup_update: Optional[datetime] = None
    sync_direction: SyncDirection = SyncDirection.BIDIRECTIONAL
    parent_task_id: Optional[str] = None

    @classmethod
    def from_dict(cls, issue_id: str, data: dict[str, Any]) -> "SyncState":
        """Create SyncState from a dictionary.

        Args:
            issue_id: The beads issue ID (key in sync_state.json)
            data: Dictionary with sync state data

        Returns:
            SyncState instance
        """
        # Parse timestamps
        def parse_dt(value: Any) -> Optional[datetime]:
            if value is None:
                return None
            if isinstance(value, datetime):
                return value
            if isinstance(value, str):
                try:
                    return datetime.fromisoformat(value.replace("Z", "+00:00"))
                except ValueError:
                    return None
            return None

        # Parse sync direction
        direction_str = data.get("sync_direction", "bidirectional")
        try:
            direction = SyncDirection(direction_str)
        except ValueError:
            direction = SyncDirection.BIDIRECTIONAL

        return cls(
            issue_id=issue_id,
            clickup_task_id=data.get("clickup_task_id"),
            last_synced=parse_dt(data.get("last_synced")),
            last_beads_update=parse_dt(data.get("last_beads_update")),
            last_clickup_update=parse_dt(data.get("last_clickup_update")),
            sync_direction=direction,
            parent_task_id=data.get("parent_task_id"),
        )

    def to_dict(self) -> dict[str, Any]:
        """Convert to dictionary for JSON serialization."""
        result = {}
        if self.clickup_task_id:
            result["clickup_task_id"] = self.clickup_task_id
        if self.last_synced:
            result["last_synced"] = self.last_synced.isoformat()
        if self.last_beads_update:
            result["last_beads_update"] = self.last_beads_update.isoformat()
        if self.last_clickup_update:
            result["last_clickup_update"] = self.last_clickup_update.isoformat()
        result["sync_direction"] = self.sync_direction.value
        if self.parent_task_id:
            result["parent_task_id"] = self.parent_task_id
        return result

from_dict(issue_id, data) classmethod

Create SyncState from a dictionary.

Parameters:

Name	Type	Description	Default
issue_id	str	The beads issue ID (key in sync_state.json)	required
data	dict[str, Any]	Dictionary with sync state data	required

Returns:

Type	Description
SyncState	SyncState instance

Source code in beads_clickup/models.py

@classmethod
def from_dict(cls, issue_id: str, data: dict[str, Any]) -> "SyncState":
    """Create SyncState from a dictionary.

    Args:
        issue_id: The beads issue ID (key in sync_state.json)
        data: Dictionary with sync state data

    Returns:
        SyncState instance
    """
    # Parse timestamps
    def parse_dt(value: Any) -> Optional[datetime]:
        if value is None:
            return None
        if isinstance(value, datetime):
            return value
        if isinstance(value, str):
            try:
                return datetime.fromisoformat(value.replace("Z", "+00:00"))
            except ValueError:
                return None
        return None

    # Parse sync direction
    direction_str = data.get("sync_direction", "bidirectional")
    try:
        direction = SyncDirection(direction_str)
    except ValueError:
        direction = SyncDirection.BIDIRECTIONAL

    return cls(
        issue_id=issue_id,
        clickup_task_id=data.get("clickup_task_id"),
        last_synced=parse_dt(data.get("last_synced")),
        last_beads_update=parse_dt(data.get("last_beads_update")),
        last_clickup_update=parse_dt(data.get("last_clickup_update")),
        sync_direction=direction,
        parent_task_id=data.get("parent_task_id"),
    )

to_dict()

Convert to dictionary for JSON serialization.

Source code in beads_clickup/models.py

def to_dict(self) -> dict[str, Any]:
    """Convert to dictionary for JSON serialization."""
    result = {}
    if self.clickup_task_id:
        result["clickup_task_id"] = self.clickup_task_id
    if self.last_synced:
        result["last_synced"] = self.last_synced.isoformat()
    if self.last_beads_update:
        result["last_beads_update"] = self.last_beads_update.isoformat()
    if self.last_clickup_update:
        result["last_clickup_update"] = self.last_clickup_update.isoformat()
    result["sync_direction"] = self.sync_direction.value
    if self.parent_task_id:
        result["parent_task_id"] = self.parent_task_id
    return result

SyncStats dataclass

Statistics from a sync operation.

Source code in beads_clickup/models.py

@dataclass
class SyncStats:
    """Statistics from a sync operation."""
    created: int = 0
    updated: int = 0
    failed: int = 0
    skipped: int = 0
    unchanged: int = 0
    closed: int = 0

    def to_dict(self) -> dict[str, int]:
        """Convert to dictionary."""
        return {
            "created": self.created,
            "updated": self.updated,
            "failed": self.failed,
            "skipped": self.skipped,
            "unchanged": self.unchanged,
            "closed": self.closed,
        }

    def __add__(self, other: "SyncStats") -> "SyncStats":
        """Add two SyncStats together."""
        return SyncStats(
            created=self.created + other.created,
            updated=self.updated + other.updated,
            failed=self.failed + other.failed,
            skipped=self.skipped + other.skipped,
            unchanged=self.unchanged + other.unchanged,
            closed=self.closed + other.closed,
        )

__add__(other)

Add two SyncStats together.

Source code in beads_clickup/models.py

def __add__(self, other: "SyncStats") -> "SyncStats":
    """Add two SyncStats together."""
    return SyncStats(
        created=self.created + other.created,
        updated=self.updated + other.updated,
        failed=self.failed + other.failed,
        skipped=self.skipped + other.skipped,
        unchanged=self.unchanged + other.unchanged,
        closed=self.closed + other.closed,
    )

to_dict()

Convert to dictionary.

Source code in beads_clickup/models.py

def to_dict(self) -> dict[str, int]:
    """Convert to dictionary."""
    return {
        "created": self.created,
        "updated": self.updated,
        "failed": self.failed,
        "skipped": self.skipped,
        "unchanged": self.unchanged,
        "closed": self.closed,
    }

WebhookEvent dataclass

Represents a ClickUp webhook event.

Source code in beads_clickup/models.py

@dataclass
class WebhookEvent:
    """Represents a ClickUp webhook event."""
    event_type: str
    task_id: str
    workspace_id: str
    timestamp: datetime
    task_data: dict[str, Any] = field(default_factory=dict)
    raw_payload: dict[str, Any] = field(default_factory=dict)

    @property
    def event_id(self) -> str:
        """Generate a unique event ID for deduplication."""
        return f"{self.event_type}:{self.task_id}:{self.timestamp.isoformat()}"

event_id property

Generate a unique event ID for deduplication.