Unverified Commit bc8f5d52 authored by Philipp Sommer's avatar Philipp Sommer
Browse files

add report_id and remove status from repr

parent 83573c2d
Pipeline #35264 failed with stage
from ._version import get_versions
from .progress_report import ProgressReport # noqa: F401
from .progress_report import BaseReport, ProgressReport # noqa: F401
__version__ = get_versions()["version"]
del get_versions
......@@ -2,20 +2,17 @@ from __future__ import annotations
import curses
from enum import Enum
from typing import TYPE_CHECKING, Dict, List, Optional
from typing import TYPE_CHECKING, Any, Dict, List, Optional, Type, Union
from deprogressapi.settings import ReportMethod, ReportSettings
from pydantic import BaseModel, Field
from pydantic import BaseModel, Field, create_model
from demessaging.PulsarMessageConstants import Status
except (ImportError, ModuleNotFoundError):
class Status(str, Enum): # type: ignore
class Status(str, Enum):
SUCCESS = "success"
ERROR = "error"
RUNNING = "running"
SUCCESS = "success"
ERROR = "error"
RUNNING = "running"
......@@ -25,6 +22,8 @@ except ImportError:
from pydantic.typing import ReprArgs
from demessaging.PulsarMessageConsumer import PulsarMessageConsumer
......@@ -43,12 +42,18 @@ class BaseReport(BaseModel):
# selector for the report type. should be changed by subclasses to make
# sure we select the correct model when deserializing
report_type: Literal["basic"] = Field(
"basic", description="Selector for the report type."
description="Selector for the report type.",
report_id: str = Field("root", description="ID for the report.")
# status of the process that we report about.
status: Status = Field(
default=Status.RUNNING, description="Status of the underlying process."
description="Status of the underlying process.",
def __init__(
......@@ -73,6 +78,21 @@ class BaseReport(BaseModel):
self._window = curses.initscr()
return self._window
def from_payload(cls, payload: str) -> BaseReport:
"""Transform a payload to a report instance.
This method takes into account all subclasses of the
:class:`BaseReport` to find the correct one.
t: Any = cls
for cls_ in cls.__subclasses__():
t = Union[t, cls_]
Model: Type[BaseModel] = create_model(
"Report", __root__=(t, Field(description="Report type"))
return Model.parse_raw(payload).__root__
def submit(self) -> None:
"""Submit the report.
......@@ -114,7 +134,13 @@ class BaseReport(BaseModel):
def print_curses(self):
"""Update the :attr:`window`."""
self.window.addstr(1, 0, self.render_report())
report = self.render_report()
height, width = self.window.getmaxyx()
lines = report.splitlines()[-height + 1 :]
report = "\n".join([line[: width - 1] for line in lines])
self.window.addstr(1, 0, report)
def render_report(self) -> str:
......@@ -135,6 +161,12 @@ class BaseReport(BaseModel):
def __exit__(self, exc_type, exc_value, exc_tb):
self.complete(Status.SUCCESS if exc_value is None else Status.ERROR)
def __repr_args__(self) -> ReprArgs:
ret = dict(super().__repr_args__())
ret.pop("status", None)
ret.pop("report_type", None)
return list(ret.items())
class ProgressReport(BaseReport):
"""A tree-like structured progress report."""
Supports Markdown
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment