Skip to content

Commit

Permalink
refactor: define protocols for started and scored attempts
Browse files Browse the repository at this point in the history
  • Loading branch information
MHajoha committed Sep 23, 2024
1 parent 2951478 commit 89deaa0
Show file tree
Hide file tree
Showing 3 changed files with 128 additions and 26 deletions.
89 changes: 87 additions & 2 deletions questionpy/_attempt.py
Original file line number Diff line number Diff line change
@@ -1,7 +1,7 @@
from abc import ABC, abstractmethod
from collections.abc import Sequence
from collections.abc import Mapping, Sequence
from functools import cached_property
from typing import TYPE_CHECKING, ClassVar
from typing import TYPE_CHECKING, ClassVar, Protocol

import jinja2
from pydantic import BaseModel, JsonValue
Expand Down Expand Up @@ -60,6 +60,91 @@ def _merge_uis(
)


class AttemptProtocol(Protocol):
"""Defines the properties and methods an attempt must always contain."""

@property
def cache_control(self) -> CacheControl:
pass

@property
def placeholders(self) -> dict[str, str]:
pass

@property
def css_files(self) -> list[str]:
pass

@property
def files(self) -> dict[str, AttemptFile]:
pass

@property
def variant(self) -> int:
pass

@property
def formulation(self) -> str:
pass

@property
def general_feedback(self) -> str | None:
pass

@property
def specific_feedback(self) -> str | None:
pass

@property
def right_answer_description(self) -> str | None:
pass

@classmethod
@abstractmethod
def make_attempt_state(cls, question: "Question", variant: int) -> BaseAttemptState:
"""Create your attempt state."""

@abstractmethod
def score_response(self, *, try_scoring_with_countback: bool = False, try_giving_hint: bool = False) -> None:
pass


class AttemptStartedProtocol(AttemptProtocol, Protocol):
"""In addition to [AttemptProtocol][], defines that a newly started attempt must provide its attempt state.
The attempt state is only generated at attempt start and immutable afterwards, so it must only be defined on the
object returned by [Question.start_attempt][].
"""

@abstractmethod
def to_plain_attempt_state(self) -> dict[str, JsonValue]:
"""Return a jsonable representation of this attempt's state."""


class AttemptScoredProtocol(AttemptProtocol, Protocol):
"""In addition to [AttemptProtocol][], defines properties and methods which must be set after scoring."""

@property
def scoring_code(self) -> ScoringCode:
pass

@property
def scored_inputs(self) -> Mapping[str, ScoredInputModel]:
pass

@property
def score(self) -> float | None:
pass

@property
def score_final(self) -> float | None:
pass

@abstractmethod
def to_plain_scoring_state(self) -> Mapping[str, JsonValue] | None:
"""Return a jsonable representation of this attempt's scoring state, if any."""


class Attempt(ABC):
attempt_state: BaseAttemptState
scoring_state: BaseScoringState | None
Expand Down
21 changes: 17 additions & 4 deletions questionpy/_qtype.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,15 +2,15 @@
# The QuestionPy SDK is free software released under terms of the MIT license. See LICENSE.md.
# (c) Technische Universität Berlin, innoCampus <[email protected]>
from abc import ABC
from typing import ClassVar, Generic, Self, TypeVar
from typing import ClassVar, Generic, Self, TypeVar, cast

from pydantic import BaseModel, JsonValue, ValidationError

from questionpy_common.api.qtype import InvalidQuestionStateError, OptionsFormValidationError
from questionpy_common.api.question import ScoringMethod, SubquestionModel
from questionpy_common.environment import get_qpy_environment

from ._attempt import Attempt
from ._attempt import Attempt, AttemptProtocol, AttemptScoredProtocol, AttemptStartedProtocol
from ._util import get_mro_type_hint
from .form import FormModel, OptionsFormDefinition

Expand Down Expand Up @@ -126,7 +126,7 @@ def get_options_form(self) -> tuple[OptionsFormDefinition, dict[str, JsonValue]]
"""Return the options form and field values for viewing or editing this question."""
return self.options_class.qpy_form, self.options.model_dump(mode="json")

def start_attempt(self, variant: int) -> Attempt:
def start_attempt(self, variant: int) -> AttemptStartedProtocol:
attempt_state = self.attempt_class.make_attempt_state(self, variant)
return self.attempt_class(self, attempt_state)

Expand All @@ -135,14 +135,27 @@ def get_attempt(
attempt_state: dict[str, JsonValue],
scoring_state: dict[str, JsonValue] | None = None,
response: dict[str, JsonValue] | None = None,
) -> Attempt:
) -> AttemptProtocol:
parsed_attempt_state = self.attempt_class.attempt_state_class.model_validate(attempt_state)
parsed_scoring_state = None
if scoring_state is not None:
parsed_scoring_state = self.attempt_class.scoring_state_class.model_validate(scoring_state)

return self.attempt_class(self, parsed_attempt_state, parsed_scoring_state, response)

def score_attempt(
self,
attempt_state: dict[str, JsonValue],
scoring_state: dict[str, JsonValue] | None,
response: dict[str, JsonValue] | None,
*,
try_scoring_with_countback: bool,
try_giving_hint: bool,
) -> AttemptScoredProtocol:
attempt = self.get_attempt(attempt_state, scoring_state, response)
attempt.score_response(try_scoring_with_countback=try_scoring_with_countback, try_giving_hint=try_giving_hint)
return cast(AttemptScoredProtocol, attempt)

def __init_subclass__(cls, *args: object, **kwargs: object) -> None:
super().__init_subclass__(*args, **kwargs)

Expand Down
44 changes: 24 additions & 20 deletions questionpy/_wrappers/_question.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,7 +5,8 @@

from pydantic import JsonValue

from questionpy import Attempt, Question
from questionpy import Question
from questionpy._attempt import AttemptProtocol, AttemptScoredProtocol
from questionpy_common.api.attempt import AttemptModel, AttemptScoredModel, AttemptStartedModel, AttemptUi
from questionpy_common.api.question import QuestionInterface, QuestionModel
from questionpy_common.environment import get_qpy_environment
Expand Down Expand Up @@ -43,10 +44,10 @@ def _export_question(question: Question) -> QuestionModel:
)


def _export_attempt(attempt: Attempt) -> dict:
def _export_attempt(attempt: AttemptProtocol) -> dict:
return {
"lang": _get_output_lang(),
"variant": attempt.attempt_state.variant,
"variant": attempt.variant,
"ui": AttemptUi(
formulation=attempt.formulation,
general_feedback=attempt.general_feedback,
Expand All @@ -60,7 +61,7 @@ def _export_attempt(attempt: Attempt) -> dict:
}


def _export_score(attempt: Attempt) -> dict:
def _export_score(attempt: AttemptScoredProtocol) -> dict:
plain_scoring_state = attempt.to_plain_scoring_state()
return {
"scoring_state": None if plain_scoring_state is None else json.dumps(plain_scoring_state),
Expand All @@ -81,23 +82,16 @@ def start_attempt(self, variant: int) -> AttemptStartedModel:
plain_attempt_state = attempt.to_plain_attempt_state()
return AttemptStartedModel(**_export_attempt(attempt), attempt_state=json.dumps(plain_attempt_state))

def _get_attempt_internal(
self,
attempt_state: str,
scoring_state: str | None = None,
response: dict[str, JsonValue] | None = None,
) -> Attempt:
plain_attempt_state = json.loads(attempt_state)
plain_scoring_state = None
if scoring_state:
plain_scoring_state = json.loads(scoring_state)

return self._question.get_attempt(plain_attempt_state, plain_scoring_state, response)

def get_attempt(
self, attempt_state: str, scoring_state: str | None = None, response: dict[str, JsonValue] | None = None
) -> AttemptModel:
return AttemptModel(**_export_attempt(self._get_attempt_internal(attempt_state, scoring_state, response)))
parsed_attempt_state = json.loads(attempt_state)
parsed_scoring_state = None
if scoring_state:
parsed_scoring_state = json.loads(scoring_state)

attempt = self._question.get_attempt(parsed_attempt_state, parsed_scoring_state, response)
return AttemptModel(**_export_attempt(attempt))

def score_attempt(
self,
Expand All @@ -108,8 +102,18 @@ def score_attempt(
try_scoring_with_countback: bool = False,
try_giving_hint: bool = False,
) -> AttemptScoredModel:
attempt = self._get_attempt_internal(attempt_state, scoring_state, response)
attempt.score_response(try_scoring_with_countback=try_scoring_with_countback, try_giving_hint=try_giving_hint)
parsed_attempt_state = json.loads(attempt_state)
parsed_scoring_state = None
if scoring_state:
parsed_scoring_state = json.loads(scoring_state)

attempt = self._question.score_attempt(
parsed_attempt_state,
parsed_scoring_state,
response,
try_scoring_with_countback=try_scoring_with_countback,
try_giving_hint=try_giving_hint,
)
return AttemptScoredModel(**_export_attempt(attempt), **_export_score(attempt))

def export_question_state(self) -> str:
Expand Down

0 comments on commit 89deaa0

Please sign in to comment.