Maharshi Gor
Updated workflow APIs, code clean up and minor functions for hf pipeline support
f064c62
| from datetime import datetime | |
| from typing import Dict, List, Literal, Optional | |
| from pydantic import BaseModel, Field | |
| from shared.workflows.structs import TossupWorkflow, Workflow | |
| CompetitionType = Literal["tossup", "bonus"] | |
| SubmissionType = Literal["python_file", "simple_workflow", "complex_workflow", "hf_pipeline"] | |
| SubmissionStatus = Literal["submitted", "in_progress", "completed", "failed"] | |
| class User(BaseModel): | |
| """ | |
| Represents a user in the competition system, formatted for HuggingFace datasets. | |
| """ | |
| username: str = Field(description="HuggingFace username of the user") | |
| name: str = Field(description="Full name of the user") | |
| email: str = Field(description="Contact email of the user") | |
| affiliation: str = Field(description="Affiliation of the user") | |
| class Submission(BaseModel): | |
| """ | |
| Represents a submission in the competition system, formatted for HuggingFace datasets. | |
| This model is designed to be easily serializable to/from HuggingFace dataset format | |
| while maintaining type safety and validation through Pydantic. | |
| Attributes: | |
| id: Unique identifier for the submission | |
| name: Display name of the submission | |
| description: Detailed description of what the submission does | |
| user_email: Email of the user who created the submission | |
| competition_type: Type of competition (Tossup or Bonus) | |
| submission_type: Format of the submission (python file or workflow) | |
| workflow: Optional workflow definition for workflow submissions, stored as JSON | |
| code: Optional code content for python file submissions | |
| status: Current status of the submission | |
| created_at: ISO format timestamp of creation | |
| updated_at: ISO format timestamp of last update | |
| """ | |
| id: str = Field(description="Unique identifier for the submission") | |
| username: str = Field(description="HuggingFace username of the user who created the submission") | |
| model_name: str = Field(description="Display name of the submission") | |
| description: str = Field(description="Detailed description of what the submission does") | |
| competition_type: CompetitionType = Field(description="Type of competition (tossup or bonus)") | |
| submission_type: SubmissionType = Field(description="Format of the submission (python file or workflow)") | |
| # TODO: Make workflow as json / yaml string instead of Workflow object | |
| workflow: Optional[Workflow] = Field(default=None, description="Optional workflow definition stored as JSON dict") | |
| code: Optional[str] = Field(default=None, description="Optional code content for python file submissions") | |
| status: SubmissionStatus = Field(description="Current status of the submission") | |
| created_at: str = Field(description="ISO format timestamp of creation") | |
| updated_at: str = Field(description="ISO format timestamp of last update") | |
| version: int = Field(description="Version number of the submission", default=1) # For backwards compatibility | |
| def to_dict(self) -> Dict: | |
| """Convert to dictionary format suitable for HF datasets""" | |
| data = self.model_dump() | |
| if self.workflow: | |
| data["workflow"] = self.workflow.model_dump(exclude_defaults=True) | |
| return data | |
| def from_dict(cls, data: Dict) -> "Submission": | |
| """Create instance from dictionary format used in HF datasets""" | |
| if data.get("workflow"): | |
| if data["competition_type"] == "tossup": | |
| data["workflow"] = TossupWorkflow.model_validate(data["workflow"]) | |
| else: | |
| data["workflow"] = Workflow.model_validate(data["workflow"]) | |
| return cls.model_validate(data) | |