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

split progress_report module into submodules

parent b3b976b8
from ._version import get_versions
from .progress_report import ( # noqa: F401
BaseReport,
PrintReport,
ProgressReport,
)
from .base import BaseReport # noqa: F401
from .print import PrintReport # noqa: F401
from .tree import ProgressReport # noqa: F401
__version__ = get_versions()["version"]
del get_versions
......@@ -2,31 +2,28 @@ from __future__ import annotations
import curses
from enum import Enum
from typing import TYPE_CHECKING, Any, Dict, List, Optional, Type, Union
from typing import TYPE_CHECKING, Any, Dict, Optional, Type, Union
from deprogressapi.settings import ReportSettings, ShowReportMethod
from pydantic import BaseModel, Field, create_model
class Status(str, Enum):
SUCCESS = "success"
ERROR = "error"
RUNNING = "running"
try:
from typing import Literal # type: ignore
except ImportError:
from typing_extensions import Literal # type: ignore
if TYPE_CHECKING:
from pydantic.typing import ReprArgs
from demessaging.PulsarMessageConsumer import PulsarMessageConsumer
class Status(str, Enum):
SUCCESS = "success"
ERROR = "error"
RUNNING = "running"
class BaseReport(BaseModel):
"""A base report for sending messages via pulsar."""
......@@ -263,125 +260,3 @@ class BaseReport(BaseModel):
ret.pop("report_type", None)
ret.pop("settings", None)
return list(ret.items())
class ProgressReport(BaseReport):
"""A tree-like structured progress report."""
report_type: Literal["tree"] = Field( # type: ignore
"tree", description="Selector for the report type."
)
step_message: str = Field(description="The description of the process.")
steps: int = Field(
default=0, description="The number of subprocesses in this report."
)
children: List[ProgressReport] = Field(default_factory=list)
_parent: Optional[ProgressReport] = None
def submit(self):
# reimplemented to use the parents submit method if there is a parent.
if self._parent is not None:
self._parent.submit()
else:
super().submit()
submit.__doc__ = BaseReport.submit.__doc__
def create_subreport(self, *args, **kwargs):
"""Create a child subreport.
This method creates a new subreport and registers it as a child.
Parameters are the same as for the :class:`ProgressReport`.
Parameters
----------
submit: Optional[bool]
Keyword-only argument. If True (default False), submit the child
report.
``*args, **kwargs``
The same as for the :class:`ProgressReport`
Returns
-------
ProgressReport
The child report that has been created.
"""
submit = kwargs.pop("submit", False)
child = ProgressReport(*args, **kwargs)
child._parent = self
self.children.append(child)
if submit:
self.submit()
return child
@classmethod
def get_dummy_arguments(cls) -> Dict:
# reimplemented to add a step_message
ret = super().get_dummy_arguments()
ret["step_message"] = ""
return ret
get_dummy_arguments.__doc__ = BaseReport.__doc__
ProgressReport.update_forward_refs()
class PrintReport(BaseReport):
"""A simple report to print a statement."""
report_type: Literal["print"] = Field( # type: ignore
"print", description="Selector for the report type."
)
messages: List[str] = Field(
default_factory=list, description="The message to print."
)
def report_to_string(self) -> str:
return "\n".join(self.messages).rstrip()
def show_print(self) -> None:
"""Only print the last message."""
print(self.messages[-1])
def show(self) -> None:
"""Reimplement to only show the report when it's complete."""
if self.messages:
if self.status != Status.SUCCESS or self.messages[-1]:
return super().show()
def print(self, message: str) -> None:
"""Update the report and show the message."""
self.messages.append(message)
self.status = Status.RUNNING
self.submit()
def error(self, message: str) -> None:
"""Display an error message."""
self.messages.append(message)
self.status = Status.ERROR
self.submit()
def complete(
self, status: Status = Status.SUCCESS, message: str = ""
) -> None:
"""Mark the report as complete.
This method marks the report as complete and closes eventually created
widgets or curses windows. You can submit an additional message that
will then be added to the report.
Parameters
----------
status : Status, optional
[description], by default Status.SUCCESS
"""
self.messages.append(message)
return super().complete(status=status)
def __exit__(self, exc_type, exc_value, exc_tb):
if exc_value is not None:
self.messages.append(str(exc_value))
return super().__exit__(exc_type, exc_value, exc_tb)
"""Print report to generate output."""
from __future__ import annotations
from typing import List
from deprogressapi.base import BaseReport, Status
from pydantic import Field
try:
from typing import Literal # type: ignore
except ImportError:
from typing_extensions import Literal # type: ignore
class PrintReport(BaseReport):
"""A simple report to print a statement."""
report_type: Literal["print"] = Field( # type: ignore
"print", description="Selector for the report type."
)
messages: List[str] = Field(
default_factory=list, description="The message to print."
)
def report_to_string(self) -> str:
return "\n".join(self.messages).rstrip()
def show_print(self) -> None:
"""Only print the last message."""
print(self.messages[-1])
def show(self) -> None:
"""Reimplement to only show the report when it's complete."""
if self.messages:
if self.status != Status.SUCCESS or self.messages[-1]:
return super().show()
def print(self, message: str) -> None:
"""Update the report and show the message."""
self.messages.append(message)
self.status = Status.RUNNING
self.submit()
def error(self, message: str) -> None:
"""Display an error message."""
self.messages.append(message)
self.status = Status.ERROR
self.submit()
def complete(
self, status: Status = Status.SUCCESS, message: str = ""
) -> None:
"""Mark the report as complete.
This method marks the report as complete and closes eventually created
widgets or curses windows. You can submit an additional message that
will then be added to the report.
Parameters
----------
status : Status, optional
[description], by default Status.SUCCESS
"""
self.messages.append(message)
return super().complete(status=status)
def __exit__(self, exc_type, exc_value, exc_tb):
if exc_value is not None:
self.messages.append(str(exc_value))
return super().__exit__(exc_type, exc_value, exc_tb)
"""Tree-like progress report for the DASF progress API."""
from __future__ import annotations
from typing import Dict, List, Optional
from deprogressapi.base import BaseReport
from pydantic import Field
try:
from typing import Literal # type: ignore
except ImportError:
from typing_extensions import Literal # type: ignore
class ProgressReport(BaseReport):
"""A tree-like structured progress report."""
report_type: Literal["tree"] = Field( # type: ignore
"tree", description="Selector for the report type."
)
step_message: str = Field(description="The description of the process.")
steps: int = Field(
default=0, description="The number of subprocesses in this report."
)
children: List[ProgressReport] = Field(default_factory=list)
_parent: Optional[ProgressReport] = None
def submit(self):
# reimplemented to use the parents submit method if there is a parent.
if self._parent is not None:
self._parent.submit()
else:
super().submit()
submit.__doc__ = BaseReport.submit.__doc__
def create_subreport(self, *args, **kwargs):
"""Create a child subreport.
This method creates a new subreport and registers it as a child.
Parameters are the same as for the :class:`ProgressReport`.
Parameters
----------
submit: Optional[bool]
Keyword-only argument. If True (default False), submit the child
report.
``*args, **kwargs``
The same as for the :class:`ProgressReport`
Returns
-------
ProgressReport
The child report that has been created.
"""
submit = kwargs.pop("submit", False)
child = ProgressReport(*args, **kwargs)
child._parent = self
self.children.append(child)
if submit:
self.submit()
return child
@classmethod
def get_dummy_arguments(cls) -> Dict:
# reimplemented to add a step_message
ret = super().get_dummy_arguments()
ret["step_message"] = ""
return ret
get_dummy_arguments.__doc__ = BaseReport.__doc__
ProgressReport.update_forward_refs()
Markdown is supported
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