FastAPI: Add DTOs and pipeline run endpoint

.gitignore

@@ -158,3 +158,4 @@ cython_debug/
 #  and can be added to the global gitignore or merged into this file.  For a more nuclear
 #  option (not recommended) you can uncomment the following to ignore the entire idea folder.
 .idea/
+/app/pipeline_playground.py

.pre-commit-config.yaml

@@ -10,4 +10,4 @@
         rev: v2.0.0
         hooks:
           - id: flake8
-            language_version: python3.12
+            language_version: python3.12

app/common/message_converters.py

@@ -1,6 +1,25 @@
+from datetime import datetime
+
 from langchain_core.messages import BaseMessage
 
-from domain import IrisMessage, IrisMessageRole
+from domain.iris_message import IrisMessage, IrisMessageRole
+from domain.data.message_dto import MessageDTO, MessageContentDTO, IrisMessageSender
+
+
+def convert_iris_message_to_message_dto(iris_message: IrisMessage) -> MessageDTO:
+    match iris_message.role:
+        case "user":
+            sender = IrisMessageSender.USER
+        case "assistant":
+            sender = IrisMessageSender.LLM
+        case _:
+            raise ValueError(f"Unknown message role: {iris_message.role}")
+
+    return MessageDTO(
+        sent_at=datetime.now(),
+        sender=sender,
+        contents=[MessageContentDTO(textContent=iris_message.text)],
+    )
 
 
 def convert_iris_message_to_langchain_message(iris_message: IrisMessage) -> BaseMessage:

app/domain/__init__.py

@@ -1,5 +1,7 @@
-from domain.message import IrisMessage, IrisMessageRole
-from domain.course import Course
-from domain.exercise import ProgrammingExercise
-from domain.submission import ProgrammingSubmission
-from domain.codehint import CodeHint
+from error_response_dto import IrisErrorResponseDTO
+from pipeline_execution_dto import PipelineExecutionDTO
+from pipeline_execution_settings_dto import PipelineExecutionSettingsDTO
+from domain.tutor_chat.tutor_chat_pipeline_execution_dto import (
+    TutorChatPipelineExecutionDTO,
+)
+from result_dto import PyrisResultDTO

app/domain/data/build_log_entry.py

@@ -0,0 +1,10 @@
+from datetime import datetime
+from pydantic import BaseModel
+
+
+class BuildLogEntryDTO(BaseModel):
+    timestamp: datetime
+    message: str
+
+    def __str__(self):
+        return f"{self.timestamp}: {self.message}"

app/domain/data/course_dto.py

@@ -0,0 +1,7 @@
+from pydantic import BaseModel
+
+
+class CourseDTO(BaseModel):
+    id: int
+    name: str
+    description: str

app/domain/data/feedback_dto.py

@@ -0,0 +1,7 @@
+from pydantic import BaseModel, Field
+
+
+class FeedbackDTO(BaseModel):
+    text: str
+    test_case_name: str = Field(alias="testCaseName")
+    credits: float

app/domain/data/image_message_content_dto.py

@@ -0,0 +1,5 @@
+from pydantic import BaseModel, Field
+
+
+class ImageMessageContentDTO(BaseModel):
+    image_data: str = Field(alias="imageData")

app/domain/data/json_message_content_dto.py

@@ -0,0 +1,6 @@
+from pydantic import BaseModel, Field, Json
+from typing import Any
+
+
+class JsonMessageContentDTO(BaseModel):
+    json_content: Json[Any] = Field(alias="jsonContent")

app/domain/data/lecture_unit_dto.py

@@ -0,0 +1,11 @@
+from datetime import datetime
+
+from pydantic import BaseModel, Field
+
+
+class LectureUnitDTO(BaseModel):
+    id: int
+    lecture_id: int = Field(alias="lectureId")
+    release_date: datetime = Field(alias="releaseDate")
+    name: str
+    attachment_version: int = Field(alias="attachmentVersion")

app/domain/data/message_content_dto.py

@@ -0,0 +1,5 @@
+from pydantic import BaseModel
+
+
+class MessageContentDTO(BaseModel):
+    pass

app/domain/data/message_dto.py

@@ -0,0 +1,40 @@
+from datetime import datetime
+from enum import Enum
+from typing import List, Literal
+
+from domain.iris_message import IrisMessage
+from message_content_dto import MessageContentDTO
+
+from pydantic import BaseModel, Field
+
+
+class IrisMessageSender(str, Enum):
+    USER = "USER"
+    LLM = "LLM"
+
+
+class MessageDTO(BaseModel):
+    sent_at: datetime = Field(alias="sentAt")
+    sender: Literal[IrisMessageSender.USER, IrisMessageSender.LLM]
+    contents: List[MessageContentDTO]
+
+    def __str__(self):
+        match self.sender:
+            case IrisMessageSender.USER:
+                sender = "user"
+            case IrisMessageSender.LLM:
+                sender = "ai"
+            case _:
+                raise ValueError(f"Unknown message sender: {self.sender}")
+        return f"{sender}: {self.contents[0].textContent}"
+
+    def convert_to_iris_message(self):
+        match self.sender:
+            case IrisMessageSender.USER:
+                sender = "user"
+            case IrisMessageSender.LLM:
+                sender = "assistant"
+            case _:
+                raise ValueError(f"Unknown message sender: {self.sender}")
+
+        return IrisMessage(text=self.contents[0].textContent, role=sender)

app/domain/data/programming_exercise_dto.py

@@ -0,0 +1,30 @@
+from typing import Dict
+
+from pydantic import BaseModel, Field
+from datetime import datetime
+from enum import Enum
+
+
+class ProgrammingLanguage(str, Enum):
+    JAVA = "JAVA"
+    PYTHON = "PYTHON"
+    C = "C"
+    HASKELL = "HASKELL"
+    KOTLIN = "KOTLIN"
+    VHDL = "VHDL"
+    ASSEMBLER = "ASSEMBLER"
+    SWIFT = "SWIFT"
+    OCAML = "OCAML"
+    EMPTY = "EMPTY"
+
+
+class ProgrammingExerciseDTO(BaseModel):
+    id: int
+    name: str
+    programming_language: ProgrammingLanguage = Field(alias="programmingLanguage")
+    template_repository: Dict[str, str] = Field(alias="templateRepository")
+    solution_repository: Dict[str, str] = Field(alias="solutionRepository")
+    test_repository: Dict[str, str] = Field(alias="testRepository")
+    problem_statement: str = Field(alias="problemStatement")
+    start_date: datetime = Field(alias="startDate")
+    end_date: datetime = Field(alias="endDate")

app/domain/data/result_dto.py

@@ -0,0 +1,12 @@
+from typing import List
+
+from pydantic import BaseModel, Field
+from datetime import datetime
+
+from domain.data.feedback_dto import FeedbackDTO
+
+
+class ResultDTO(BaseModel):
+    completion_date: datetime = Field(alias="completionDate")
+    successful: bool
+    feedbacks: List[FeedbackDTO]

app/domain/data/submission_dto.py

@@ -0,0 +1,17 @@
+from typing import List, Dict
+
+from pydantic import BaseModel, Field
+
+from datetime import datetime
+from domain.data.build_log_entry import BuildLogEntryDTO
+from domain.data.result_dto import ResultDTO
+
+
+class SubmissionDTO(BaseModel):
+    id: int
+    date: datetime
+    repository: Dict[str, str]
+    is_practice: bool = Field(alias="isPractice")
+    build_failed: bool = Field(alias="buildFailed")
+    build_log_entries: List[BuildLogEntryDTO] = Field(alias="buildLogEntries")
+    latest_result: ResultDTO = Field(alias="latestResult")

app/domain/data/text_message_content_dto.py

@@ -0,0 +1,5 @@
+from pydantic import BaseModel, Field
+
+
+class TextMessageContentDTO(BaseModel):
+    text_content: str = Field(alias="textContent")

app/domain/data/user_dto.py

@@ -0,0 +1,7 @@
+from pydantic import BaseModel, Field
+
+
+class UserDTO(BaseModel):
+    id: int
+    first_name: str = Field(alias="firstName")
+    last_name: str = Field(alias="lastName")

app/domain/error_response_dto.py

@@ -0,0 +1,5 @@
+from pydantic import BaseModel, Field
+
+
+class IrisErrorResponseDTO(BaseModel):
+    error_message: str = Field(alias="errorMessage")

app/domain/health_status_dto.py

@@ -0,0 +1,13 @@
+from pydantic import BaseModel
+from enum import Enum
+
+
+class ModelStatus(str, Enum):
+    UP = "UP"
+    DOWN = "DOWN"
+    NOT_AVAILABLE = "NOT_AVAILABLE"
+
+
+class HealthStatusDTO(BaseModel):
+    model: str
+    status: ModelStatus

app/domain/iris_message.py

@@ -0,0 +1,20 @@
+from enum import Enum
+from typing import Literal
+
+from pydantic import BaseModel, Field
+
+
+class IrisMessageRole(str, Enum):
+    USER = "USER"
+    ASSISTANT = "ASSISTANT"
+    SYSTEM = "SYSTEM"
+
+
+class IrisMessage(BaseModel):
+    text: str
+    role: Literal[
+        IrisMessageRole.USER, IrisMessageRole.ASSISTANT, IrisMessageRole.SYSTEM
+    ]
+
+    def __str__(self):
+        return f"{self.role.lower()}: {self.text}"