FastAPI: Add DTOs and pipeline run endpoint

.pre-commit-config.yaml

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

README.MD

@@ -1 +1,16 @@
-# Pyris V2
+# Pyris V2
+## With local environment
+
+### Setup
+ - Check python version: `python --version` (should be 3.12)
+ - Install packages: `pip install -r requirements.txt`
+
+### Run server
+ - Run server: 
+   ```[bash]
+     APPLICATION_YML_PATH=<path-to-your-application-yml-file> LLM_CONFIG_PATH=<path-to-your-llm-config-yml> uvicorn app.main:app --reload
+    ```
+ - Access API docs: http://localhost:8000/docs
+
+## With docker
+TBD

app/common/__init__.py

@@ -1,5 +1,5 @@
-from common.singleton import Singleton
-from common.message_converters import (
+from ..common.singleton import Singleton
+from ..common.message_converters import (
     convert_iris_message_to_langchain_message,
     convert_langchain_message_to_iris_message,
 )

app/common/custom_exceptions.py

@@ -0,0 +1,45 @@
+from fastapi import HTTPException, status
+
+
+class RequiresAuthenticationException(HTTPException):
+    def __init__(self):
+        super().__init__(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail={
+                "type": "not_authenticated",
+                "errorMessage": "Requires authentication",
+            },
+        )
+
+
+class PermissionDeniedException(HTTPException):
+    def __init__(self):
+        super().__init__(
+            status_code=status.HTTP_403_FORBIDDEN,
+            detail={
+                "type": "not_authorized",
+                "errorMessage": "Permission denied",
+            },
+        )
+
+
+class PipelineInvocationError(HTTPException):
+    def __init__(self):
+        super().__init__(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail={
+                "type": "bad_request",
+                "errorMessage": "Cannot invoke pipeline",
+            },
+        )
+
+
+class PipelineNotFoundException(HTTPException):
+    def __init__(self):
+        super().__init__(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail={
+                "type": "pipeline_not_found",
+                "errorMessage": "Pipeline not found",
+            },
+        )

app/common/message_converters.py

@@ -1,6 +1,5 @@
 from langchain_core.messages import BaseMessage
-
-from domain import IrisMessage, IrisMessageRole
+from ..domain.iris_message import IrisMessage, IrisMessageRole
 
 
 def convert_iris_message_to_langchain_message(iris_message: IrisMessage) -> BaseMessage:

app/config.py

@@ -0,0 +1,36 @@
+import os
+from pathlib import Path
+from pydantic import BaseModel
+import yaml
+
+
+class APIKeyConfig(BaseModel):
+    token: str
+
+
+class Settings(BaseModel):
+    api_keys: list[APIKeyConfig]
+
+    @classmethod
+    def get_settings(cls):
+        """Get the settings from the configuration file."""
+        file_path_env = os.environ.get("APPLICATION_YML_PATH")
+        if not file_path_env:
+            raise EnvironmentError(
+                "APPLICATION_YML_PATH environment variable is not set."
+            )
+
+        file_path = Path(file_path_env)
+        try:
+            with open(file_path, "r") as file:
+                settings_file = yaml.safe_load(file)
+            return cls.parse_obj(settings_file)
+        except FileNotFoundError as e:
+            raise FileNotFoundError(
+                f"Configuration file not found at {file_path}."
+            ) from e
+        except yaml.YAMLError as e:
+            raise yaml.YAMLError(f"Error parsing YAML file at {file_path}.") from e
+
+
+settings = Settings.get_settings()

app/dependencies.py

@@ -0,0 +1,25 @@
+from fastapi import Depends
+from fastapi.requests import Request
+
+from app.common.custom_exceptions import (
+    RequiresAuthenticationException,
+    PermissionDeniedException,
+)
+from app.config import APIKeyConfig, settings
+
+
+def _get_api_key(request: Request) -> str:
+    authorization_header = request.headers.get("Authorization")
+
+    if not authorization_header:
+        raise RequiresAuthenticationException
+
+    return authorization_header
+
+
+class TokenValidator:
+    async def __call__(self, api_key: str = Depends(_get_api_key)) -> APIKeyConfig:
+        for key in settings.api_keys:
+            if key.token == api_key:
+                return key
+        raise PermissionDeniedException

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 .iris_message import IrisMessage, IrisMessageRole

app/domain/data/build_log_entry.py

@@ -0,0 +1,12 @@
+from datetime import datetime
+from typing import Optional
+
+from pydantic import BaseModel
+
+
+class BuildLogEntryDTO(BaseModel):
+    timestamp: Optional[datetime] = None
+    message: Optional[str] = None
+
+    def __str__(self):
+        return f"{self.timestamp}: {self.message}"

app/domain/data/course_dto.py

@@ -0,0 +1,9 @@
+from typing import Optional
+
+from pydantic import BaseModel
+
+
+class CourseDTO(BaseModel):
+    id: int
+    name: Optional[str] = None
+    description: Optional[str] = None

app/domain/data/feedback_dto.py

@@ -0,0 +1,12 @@
+from typing import Optional
+
+from pydantic import BaseModel, Field
+
+
+class FeedbackDTO(BaseModel):
+    text: Optional[str] = None
+    test_case_name: str = Field(alias="testCaseName")
+    credits: float
+
+    def __str__(self):
+        return f"{self.test_case_name}: {self.text} ({self.credits} credits)"

app/domain/data/image_message_content_dto.py

@@ -0,0 +1,7 @@
+from typing import Optional
+
+from pydantic import BaseModel, Field
+
+
+class ImageMessageContentDTO(BaseModel):
+    image_data: Optional[str] = Field(alias="imageData", default=None)

app/domain/data/json_message_content_dto.py

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

app/domain/data/lecture_unit_dto.py

@@ -0,0 +1,12 @@
+from datetime import datetime
+from typing import Optional
+
+from pydantic import BaseModel, Field
+
+
+class LectureUnitDTO(BaseModel):
+    id: int
+    lecture_id: int = Field(alias="lectureId")
+    release_date: Optional[datetime] = Field(alias="releaseDate", default=None)
+    name: Optional[str] = None
+    attachment_version: int = Field(alias="attachmentVersion")

app/domain/data/message_content_dto.py

@@ -0,0 +1,9 @@
+from typing import Union
+
+from ...domain.data.image_message_content_dto import ImageMessageContentDTO
+from ...domain.data.json_message_content_dto import JsonMessageContentDTO
+from ...domain.data.text_message_content_dto import TextMessageContentDTO
+
+MessageContentDTO = Union[
+    TextMessageContentDTO, ImageMessageContentDTO, JsonMessageContentDTO
+]

app/domain/data/message_dto.py

@@ -0,0 +1,51 @@
+from datetime import datetime
+from enum import Enum
+from typing import List, Literal
+
+from langchain_core.messages import HumanMessage, AIMessage
+
+from .message_content_dto import MessageContentDTO
+from ...domain.iris_message import IrisMessage
+
+from pydantic import BaseModel, Field
+
+
+class IrisMessageSender(str, Enum):
+    USER = "USER"
+    LLM = "LLM"
+
+
+class MessageDTO(BaseModel):
+    sent_at: datetime | None = Field(alias="sentAt", default=None)
+    sender: Literal[IrisMessageSender.USER, IrisMessageSender.LLM]
+    contents: List[MessageContentDTO] = []
+
+    def __str__(self):
+        match self.sender:
+            case IrisMessageSender.USER:
+                sender = "user"
+            case IrisMessageSender.LLM:
+                sender = "assistant"
+            case _:
+                raise ValueError(f"Unknown message sender: {self.sender}")
+        return f"{sender}: {self.contents[0].text_content}"
+
+    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].text_content, role=sender)
+
+    def convert_to_langchain_message(self):
+        match self.sender:
+            case IrisMessageSender.USER:
+                return HumanMessage(content=self.contents[0].text_content)
+            case IrisMessageSender.LLM:
+                return AIMessage(content=self.contents[0].text_content)
+            case _:
+                raise ValueError(f"Unknown message sender: {self.sender}")

app/domain/data/programming_exercise_dto.py

@@ -0,0 +1,30 @@
+from typing import Dict, Optional
+
+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: Optional[datetime] = Field(alias="startDate", default=None)
+    end_date: Optional[datetime] = Field(alias="endDate", default=None)

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,19 @@
+from typing import List, Dict, Optional
+
+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: Optional[datetime] = None
+    repository: Dict[str, str]
+    is_practice: bool = Field(alias="isPractice")
+    build_failed: bool = Field(alias="buildFailed")
+    build_log_entries: List[BuildLogEntryDTO] = Field(
+        alias="buildLogEntries", default=[]
+    )
+    latest_result: Optional[ResultDTO] = Field(alias="latestResult", default=None)