refactor: add if_present utility (#128)

opvious/common.py

@@ -2,7 +2,7 @@
 import functools
 from importlib import metadata
 import math
-from typing import Any, Callable, Iterable, Optional, Union
+from typing import Any, Callable, Iterable, Optional, TypeVar, Union
 import urllib.parse
 import weakref
 
@@ -19,6 +19,14 @@
 Uuid = str
 
 
+_V = TypeVar("_V", contravariant=True)
+_R = TypeVar("_R", covariant=True)
+
+
+def if_present(arg: Optional[_V], fn: Callable[[_V], _R]) -> Optional[_R]:
+    return None if arg is None else fn(arg)
+
+
 # Formatting
 
 
@@ -100,7 +108,7 @@ def encode_annotations(annots: list[Annotation]) -> Json:
     return [
         json_dict(key=annot)
         if isinstance(annot, str)
-        else json_dict(key=annot[0], value=annot[1])
+        else json_dict(key=annot[0], value=str(annot[1]))
         for annot in annots
     ]
 
@@ -114,9 +122,9 @@ def decode_annotations(elems: Json) -> list[Annotation]:
     ]
 
 
-def decode_datetime(iso: Optional[str]) -> Optional[datetime]:
+def decode_datetime(iso: str) -> datetime:
     """Parses a datetime from an ISO-formatted string"""
-    return datetime.fromisoformat(iso) if iso else None
+    return datetime.fromisoformat(iso)
 
 
 # Async

opvious/data/queued_solves.py

@@ -1,7 +1,7 @@
 from __future__ import annotations
 
 import dataclasses
-from datetime import datetime
+from datetime import datetime, timedelta
 from typing import Any, Optional, cast
 
 from ..common import (
@@ -10,6 +10,7 @@
     Uuid,
     decode_datetime,
     decode_annotations,
+    if_present,
 )
 from .outcomes import (
     SolveOutcome,
@@ -31,6 +32,12 @@ class QueuedSolve:
     uuid: Uuid
     """The solve's unique identifier"""
 
+    annotations: list[Annotation]
+    """Annotation metadata"""
+
+    outcome: Optional[SolveOutcome]
+    """Final solve outcome, if available"""
+
     enqueued_at: datetime
     """The time the solve was created"""
 
@@ -40,19 +47,20 @@ class QueuedSolve:
     completed_at: Optional[datetime]
     """The time the solve completed"""
 
-    annotations: list[Annotation]
-    """Annotation metadata"""
-
-    outcome: Optional[SolveOutcome]
-    """Final solve outcome, if available"""
-
     problem_summary: Optional[ProblemSummary] = dataclasses.field(repr=False)
     """Summary information about the solved problem"""
 
     options: Json = dataclasses.field(repr=False)
     transformations: Json = dataclasses.field(repr=False)
     strategy: Json = dataclasses.field(repr=False)
 
+    @property
+    def duration(self) -> Optional[timedelta]:
+        """The solve's runtime, if it is complete"""
+        return (
+            self.completed_at - self.enqueued_at if self.completed_at else None
+        )
+
 
 def queued_solve_from_graphql(
     data: Json, attempt_data: Optional[Json] = None
@@ -68,22 +76,19 @@ def queued_solve_from_graphql(
     else:
         outcome = None
 
-    enqueued_at = decode_datetime(attempt_data["startedAt"])
-    assert enqueued_at
-
     return QueuedSolve(
         uuid=data["uuid"],
-        enqueued_at=enqueued_at,
-        dequeued_at=decode_datetime(data["dequeuedAt"]),
-        completed_at=decode_datetime(attempt_data["endedAt"]),
+        enqueued_at=decode_datetime(attempt_data["startedAt"]),
+        dequeued_at=if_present(data["dequeuedAt"], decode_datetime),
+        completed_at=if_present(attempt_data["endedAt"], decode_datetime),
         annotations=decode_annotations(attempt_data["annotations"]),
         options=data["options"],
         transformations=data["transformations"],
         strategy=data["strategy"],
         outcome=outcome,
-        problem_summary=problem_summary_from_json(data["problemSummary"])
-        if data["problemSummary"]
-        else None,
+        problem_summary=if_present(
+            data["problemSummary"], problem_summary_from_json
+        ),
     )
 
 

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "poetry.core.masonry.api"
 
 [tool.poetry]
 name = "opvious"
-version = "0.19.0rc1"
+version = "0.19.0rc2"
 description = "Opvious Python SDK"
 authors = ["Opvious Engineering <oss@opvious.io>"]
 readme = "README.md"

tests/test_client.py

@@ -461,6 +461,5 @@ async def test_paginate_formulation_solves(self):
 
     @pytest.mark.asyncio
     async def test_fetch_unknown_solve_outputs(self):
-        with pytest.raises(opvious.executors.ExecutorError) as info:
-            await client.fetch_solve_outputs("abc")
-        assert info.value.status == 404
+        with pytest.raises(opvious.executors.ExecutorError):
+            await client.fetch_solve_outputs("invalid-uuid")