feat(mma): Define TrackState and Metadata schema for track-scoped state
This commit is contained in:
92
models.py
92
models.py
@@ -1,5 +1,6 @@
|
||||
from dataclasses import dataclass, field
|
||||
from typing import List, Optional
|
||||
from typing import List, Optional, Dict, Any
|
||||
from datetime import datetime
|
||||
|
||||
@dataclass
|
||||
class Ticket:
|
||||
@@ -25,6 +26,33 @@ class Ticket:
|
||||
"""Sets the ticket status to 'completed'."""
|
||||
self.status = "completed"
|
||||
|
||||
def to_dict(self) -> Dict[str, Any]:
|
||||
return {
|
||||
"id": self.id,
|
||||
"description": self.description,
|
||||
"status": self.status,
|
||||
"assigned_to": self.assigned_to,
|
||||
"target_file": self.target_file,
|
||||
"context_requirements": self.context_requirements,
|
||||
"depends_on": self.depends_on,
|
||||
"blocked_reason": self.blocked_reason,
|
||||
"step_mode": self.step_mode,
|
||||
}
|
||||
|
||||
@classmethod
|
||||
def from_dict(cls, data: Dict[str, Any]) -> "Ticket":
|
||||
return cls(
|
||||
id=data["id"],
|
||||
description=data.get("description"),
|
||||
status=data.get("status"),
|
||||
assigned_to=data.get("assigned_to"),
|
||||
target_file=data.get("target_file"),
|
||||
context_requirements=data.get("context_requirements", []),
|
||||
depends_on=data.get("depends_on", []),
|
||||
blocked_reason=data.get("blocked_reason"),
|
||||
step_mode=data.get("step_mode", False),
|
||||
)
|
||||
|
||||
@dataclass
|
||||
class Track:
|
||||
"""
|
||||
@@ -67,3 +95,65 @@ class WorkerContext:
|
||||
ticket_id: str
|
||||
model_name: str
|
||||
messages: List[dict]
|
||||
|
||||
@dataclass
|
||||
class Metadata:
|
||||
id: str
|
||||
name: str
|
||||
status: str
|
||||
created_at: datetime
|
||||
updated_at: datetime
|
||||
|
||||
def to_dict(self) -> Dict[str, Any]:
|
||||
return {
|
||||
"id": self.id,
|
||||
"name": self.name,
|
||||
"status": self.status,
|
||||
"created_at": self.created_at.isoformat() if self.created_at else None,
|
||||
"updated_at": self.updated_at.isoformat() if self.updated_at else None,
|
||||
}
|
||||
|
||||
@classmethod
|
||||
def from_dict(cls, data: Dict[str, Any]) -> "Metadata":
|
||||
return cls(
|
||||
id=data["id"],
|
||||
name=data["name"],
|
||||
status=data.get("status"),
|
||||
created_at=datetime.fromisoformat(data['created_at']) if data.get('created_at') else None,
|
||||
updated_at=datetime.fromisoformat(data['updated_at']) if data.get('updated_at') else None,
|
||||
)
|
||||
|
||||
@dataclass
|
||||
class TrackState:
|
||||
metadata: Metadata
|
||||
discussion: List[Dict[str, Any]]
|
||||
tasks: List[Ticket]
|
||||
|
||||
def to_dict(self) -> Dict[str, Any]:
|
||||
return {
|
||||
"metadata": self.metadata.to_dict(),
|
||||
"discussion": [
|
||||
{
|
||||
k: v.isoformat() if isinstance(v, datetime) else v
|
||||
for k, v in item.items()
|
||||
}
|
||||
for item in self.discussion
|
||||
],
|
||||
"tasks": [task.to_dict() for task in self.tasks],
|
||||
}
|
||||
|
||||
@classmethod
|
||||
def from_dict(cls, data: Dict[str, Any]) -> "TrackState":
|
||||
metadata = Metadata.from_dict(data["metadata"])
|
||||
tasks = [Ticket.from_dict(task_data) for task_data in data["tasks"]]
|
||||
return cls(
|
||||
metadata=metadata,
|
||||
discussion=[
|
||||
{
|
||||
k: datetime.fromisoformat(v) if isinstance(v, str) and 'T' in v else v # Basic check for ISO format
|
||||
for k, v in item.items()
|
||||
}
|
||||
for item in data["discussion"]
|
||||
],
|
||||
tasks=tasks,
|
||||
)
|
||||
|
||||
Reference in New Issue
Block a user