move user_communicator to cli

This commit is contained in:
Sina Atalay 2024-02-11 14:37:10 +01:00
parent 715a6b4e5b
commit 6d0c4b9816
4 changed files with 248 additions and 240 deletions

View File

@ -1,12 +1,27 @@
"""
to be continued...
"""
import json import json
import pathlib import pathlib
from typing import Annotated from typing import Annotated, Callable, Optional
import re
from rich import print
import rich.console
import rich.panel
import rich.live
import rich.table
import rich.text
import rich.progress
import pydantic
import ruamel.yaml
import ruamel.yaml.parser
import typer import typer
import ruamel.yaml import ruamel.yaml
from . import user_communicator as uc
from . import data_models as dm from . import data_models as dm
from . import renderer as r from . import renderer as r
@ -19,8 +34,220 @@ app = typer.Typer(
) )
def welcome():
"""Print a welcome message to the terminal."""
table = rich.table.Table(
title="\nWelcome to [bold]Render[dodger_blue3]CV[/dodger_blue3][/bold]!",
title_justify="left",
)
table.add_column("Title", style="magenta")
table.add_column("Link", style="cyan", justify="right", no_wrap=True)
table.add_row("Documentation", "https://sinaatalay.github.io/rendercv/")
table.add_row("Source code", "https://github.com/sinaatalay/rendercv/")
table.add_row("Bug reports", "https://github.com/sinaatalay/rendercv/issues/")
table.add_row("Feature requests", "https://github.com/sinaatalay/rendercv/issues/")
print(table)
def warning(text):
"""Print a warning message to the terminal."""
print(f"[bold yellow]{text}")
def error(text, exception=None):
"""Print an error message to the terminal."""
if exception is not None:
exception_messages = exception.args
exception_message = "\n\n".join(exception_messages)
print(
f"\n[bold red]{text}[/bold red]\n\n[orange4]{exception_message}[/orange4]"
)
else:
print(f"[bold red]{text}")
def information(text):
"""Print an information message to the terminal."""
print(f"[bold green]{text}")
def get_error_message_and_location_and_value_from_a_custom_error(
error_string: str,
) -> Optional[tuple[str, str, str]]:
pattern = r"\('(.*)', '(.*)', '(.*)'\)"
match = re.search(pattern, error_string)
if match:
return match.group(1), match.group(2), match.group(3)
else:
return None
def handle_validation_error(exception: pydantic.ValidationError):
error_dictionary: dict[str, str] = {
"Input should be 'present'": (
"This is not a valid date! Please use either YYYY-MM-DD, YYYY-MM, or YYYY"
' format or "present"!'
),
"Input should be a valid integer, unable to parse string as an integer": (
"This is not a valid date! Please use either YYYY-MM-DD, YYYY-MM, or YYYY"
" format!"
),
"String should match pattern '\\d{4}-\\d{2}(-\\d{2})?'": (
"This is not a valid date! Please use either YYYY-MM-DD, YYYY-MM, or YYYY"
" format!"
),
"URL scheme should be 'http' or 'https'": "This is not a valid URL!",
"Field required": "This field is required!",
"value is not a valid phone number": "This is not a valid phone number!",
}
new_errors: list[dict[str, str]] = []
for error_object in exception.errors():
message = error_object["msg"]
location = ".".join([str(loc) for loc in error_object["loc"]])
input = error_object["input"]
custom_error = get_error_message_and_location_and_value_from_a_custom_error(
message
)
if custom_error is None:
if message in error_dictionary:
message = error_dictionary[message]
else:
message = message
else:
message = custom_error[0]
if custom_error[1] != "":
location = f"{location}.{custom_error[1]}"
input = custom_error[2]
new_errors.append({
"loc": str(location),
"msg": message,
"input": str(input),
})
table = rich.table.Table(
title="[bold red]\nThere are some errors in the input file!\n",
title_justify="left",
show_lines=True,
)
table.add_column("Location", style="cyan", no_wrap=True)
table.add_column("Input Value", style="magenta")
table.add_column("Error Message", style="orange4")
for error_object in new_errors:
table.add_row(
error_object["loc"],
error_object["input"],
error_object["msg"],
)
print(table)
print()
def handle_exceptions(function: Callable) -> Callable:
""" """
def wrapper(*args, **kwargs):
try:
function(*args, **kwargs)
except pydantic.ValidationError as e:
handle_validation_error(e)
except ruamel.yaml.YAMLError as e:
error("There is a YAML error in the input file!", e)
except RuntimeError as e:
error("An error occurred:", e)
return wrapper
class LiveProgressReporter(rich.live.Live):
"""This class is a wrapper around `rich.live.Live` that provides the live progress
reporting functionality.
Args:
number_of_steps (int): The number of steps to be finished.
"""
def __init__(self, number_of_steps: int, end_message: str = "Your CV is rendered!"):
class TimeElapsedColumn(rich.progress.ProgressColumn):
def render(self, task: "rich.progress.Task") -> rich.text.Text:
elapsed = task.finished_time if task.finished else task.elapsed
delta = f"{elapsed:.1f} s"
return rich.text.Text(str(delta), style="progress.elapsed")
self.step_progress = rich.progress.Progress(
TimeElapsedColumn(), rich.progress.TextColumn("{task.description}")
)
self.overall_progress = rich.progress.Progress(
TimeElapsedColumn(),
rich.progress.BarColumn(),
rich.progress.TextColumn("{task.description}"),
)
self.group = rich.console.Group(
rich.panel.Panel(rich.console.Group(self.step_progress)),
self.overall_progress,
)
self.overall_task_id = self.overall_progress.add_task("", total=number_of_steps)
self.number_of_steps = number_of_steps
self.end_message = end_message
self.current_step = 0
self.overall_progress.update(
self.overall_task_id,
description=(
f"[bold #AAAAAA]({self.current_step} out of"
f" {self.number_of_steps} steps finished)"
),
)
super().__init__(self.group)
def __enter__(self) -> "LiveProgressReporter":
"""Overwrite the `__enter__` method for the correct return type."""
self.start(refresh=self._renderable is not None)
return self
def start_a_step(self, step_name: str):
"""Start a step and update the progress bars."""
self.current_step_name = step_name
self.current_step_id = self.step_progress.add_task(
f"{self.current_step_name} has started."
)
def finish_the_current_step(self):
"""Finish the current step and update the progress bars."""
self.step_progress.stop_task(self.current_step_id)
self.step_progress.update(
self.current_step_id, description=f"{self.current_step_name} has finished."
)
self.current_step += 1
self.overall_progress.update(
self.overall_task_id,
description=(
f"[bold #AAAAAA]({self.current_step} out of"
f" {self.number_of_steps} steps finished)"
),
advance=1,
)
if self.current_step == self.number_of_steps:
self.end()
def end(self):
"""End the live progress reporting."""
self.overall_progress.update(
self.overall_task_id,
description=f"[bold green]{self.end_message}",
)
@app.command(help="Render a YAML input file") @app.command(help="Render a YAML input file")
@uc.handle_exceptions @handle_exceptions
def render( def render(
input_file_path: Annotated[ input_file_path: Annotated[
pathlib.Path, pathlib.Path,
@ -32,10 +259,10 @@ def render(
Args: Args:
input_file (str): Name of the YAML input file input_file (str): Name of the YAML input file
""" """
uc.welcome() welcome()
output_directory = input_file_path.parent / "rendercv_output" output_directory = input_file_path.parent / "rendercv_output"
with uc.LiveProgressReporter(number_of_steps=3) as progress: with LiveProgressReporter(number_of_steps=3) as progress:
progress.start_a_step("Reading and validating the input file") progress.start_a_step("Reading and validating the input file")
data_model = dm.read_input_file(input_file_path) data_model = dm.read_input_file(input_file_path)
progress.finish_the_current_step() progress.finish_the_current_step()
@ -70,7 +297,7 @@ def new(full_name: Annotated[str, typer.Argument(help="Your full name")]):
yaml.indent(mapping=2, sequence=4, offset=2) yaml.indent(mapping=2, sequence=4, offset=2)
yaml.dump(data_model_as_dictionary, file_path) yaml.dump(data_model_as_dictionary, file_path)
uc.information(f"Your RenderCV input file has been created at {file_path}!") information(f"Your RenderCV input file has been created at {file_path}!")
def cli(): def cli():

View File

@ -1,223 +0,0 @@
from typing import Callable, Optional
import re
from rich import print
import rich.console
import rich.panel
import rich.live
import rich.table
import rich.text
import rich.progress
import pydantic
import ruamel.yaml
import ruamel.yaml.parser
def welcome():
"""Print a welcome message to the terminal."""
table = rich.table.Table(
title="\nWelcome to [bold]Render[dodger_blue3]CV[/dodger_blue3][/bold]!",
title_justify="left",
)
table.add_column("Title", style="magenta")
table.add_column("Link", style="cyan", justify="right", no_wrap=True)
table.add_row("Documentation", "https://sinaatalay.github.io/rendercv/")
table.add_row("Source code", "https://github.com/sinaatalay/rendercv/")
table.add_row("Bug reports", "https://github.com/sinaatalay/rendercv/issues/")
table.add_row("Feature requests", "https://github.com/sinaatalay/rendercv/issues/")
print(table)
def warning(text):
"""Print a warning message to the terminal."""
print(f"[bold yellow]{text}")
def error(text, exception=None):
"""Print an error message to the terminal."""
if exception is not None:
exception_messages = exception.args
exception_message = "\n\n".join(exception_messages)
print(f"\n[bold red]{text}[/bold red]\n\n[orange4]{exception_message}[/orange4]")
else:
print(f"[bold red]{text}")
def information(text):
"""Print an information message to the terminal."""
print(f"[bold green]{text}")
def get_error_message_and_location_and_value_from_a_custom_error(
error_string: str,
) -> Optional[tuple[str, str, str]]:
pattern = r"\('(.*)', '(.*)', '(.*)'\)"
match = re.search(pattern, error_string)
if match:
return match.group(1), match.group(2), match.group(3)
else:
return None
def handle_validation_error(exception: pydantic.ValidationError):
error_dictionary: dict[str, str] = {
"Input should be 'present'": (
"This is not a valid date! Please use either YYYY-MM-DD, YYYY-MM, or YYYY"
' format or "present"!'
),
"Input should be a valid integer, unable to parse string as an integer": (
"This is not a valid date! Please use either YYYY-MM-DD, YYYY-MM, or YYYY"
" format!"
),
"String should match pattern '\\d{4}-\\d{2}(-\\d{2})?'": (
"This is not a valid date! Please use either YYYY-MM-DD, YYYY-MM, or YYYY"
" format!"
),
"URL scheme should be 'http' or 'https'": "This is not a valid URL!",
"Field required": "This field is required!",
"value is not a valid phone number": "This is not a valid phone number!",
}
new_errors: list[dict[str, str]] = []
for error_object in exception.errors():
message = error_object["msg"]
location = ".".join([str(loc) for loc in error_object["loc"]])
input = error_object["input"]
custom_error = get_error_message_and_location_and_value_from_a_custom_error(
message
)
if custom_error is None:
if message in error_dictionary:
message = error_dictionary[message]
else:
message = message
else:
message = custom_error[0]
if custom_error[1] != "":
location = f"{location}.{custom_error[1]}"
input = custom_error[2]
new_errors.append({
"loc": str(location),
"msg": message,
"input": str(input),
})
table = rich.table.Table(
title="[bold red]\nThere are some errors in the input file!\n",
title_justify="left",
show_lines=True,
)
table.add_column("Location", style="cyan", no_wrap=True)
table.add_column("Input Value", style="magenta")
table.add_column("Error Message", style="orange4")
for error_object in new_errors:
table.add_row(
error_object["loc"],
error_object["input"],
error_object["msg"],
)
print(table)
print()
def handle_exceptions(function: Callable) -> Callable:
""" """
def wrapper(*args, **kwargs):
try:
function(*args, **kwargs)
except pydantic.ValidationError as e:
handle_validation_error(e)
except ruamel.yaml.YAMLError as e:
error("There is a YAML error in the input file!", e)
except RuntimeError as e:
error("An error occurred:", e)
return wrapper
class LiveProgressReporter(rich.live.Live):
"""This class is a wrapper around `rich.live.Live` that provides the live progress
reporting functionality.
Args:
number_of_steps (int): The number of steps to be finished.
"""
def __init__(self, number_of_steps: int, end_message: str = "Your CV is rendered!"):
class TimeElapsedColumn(rich.progress.ProgressColumn):
def render(self, task: "rich.progress.Task") -> rich.text.Text:
elapsed = task.finished_time if task.finished else task.elapsed
delta = f"{elapsed:.1f} s"
return rich.text.Text(str(delta), style="progress.elapsed")
self.step_progress = rich.progress.Progress(
TimeElapsedColumn(), rich.progress.TextColumn("{task.description}")
)
self.overall_progress = rich.progress.Progress(
TimeElapsedColumn(),
rich.progress.BarColumn(),
rich.progress.TextColumn("{task.description}"),
)
self.group = rich.console.Group(
rich.panel.Panel(rich.console.Group(self.step_progress)),
self.overall_progress,
)
self.overall_task_id = self.overall_progress.add_task("", total=number_of_steps)
self.number_of_steps = number_of_steps
self.end_message = end_message
self.current_step = 0
self.overall_progress.update(
self.overall_task_id,
description=(
f"[bold #AAAAAA]({self.current_step} out of"
f" {self.number_of_steps} steps finished)"
),
)
super().__init__(self.group)
def __enter__(self) -> "LiveProgressReporter":
"""Overwrite the `__enter__` method for the correct return type."""
self.start(refresh=self._renderable is not None)
return self
def start_a_step(self, step_name: str):
"""Start a step and update the progress bars."""
self.current_step_name = step_name
self.current_step_id = self.step_progress.add_task(
f"{self.current_step_name} has started."
)
def finish_the_current_step(self):
"""Finish the current step and update the progress bars."""
self.step_progress.stop_task(self.current_step_id)
self.step_progress.update(
self.current_step_id, description=f"{self.current_step_name} has finished."
)
self.current_step += 1
self.overall_progress.update(
self.overall_task_id,
description=(
f"[bold #AAAAAA]({self.current_step} out of"
f" {self.number_of_steps} steps finished)"
),
advance=1,
)
if self.current_step == self.number_of_steps:
self.end()
def end(self):
"""End the live progress reporting."""
self.overall_progress.update(
self.overall_task_id,
description=f"[bold green]{self.end_message}",
)

View File

@ -1,35 +1,39 @@
import rendercv.user_communicator as uc import rendercv.cli as cli
import pydantic import pydantic
import ruamel.yaml import ruamel.yaml
import pytest import pytest
import typer.testing
runner = typer.testing.CliRunner()
def test_welcome(): def test_welcome():
uc.welcome() cli.welcome()
def test_warning(): def test_warning():
uc.warning("This is a warning message.") cli.warning("This is a warning message.")
def test_error(): def test_error():
uc.error("This is an error message.") cli.error("This is an error message.")
def test_information(): def test_information():
uc.information("This is an information message.") cli.information("This is an information message.")
def test_get_error_message_and_location_and_value_from_a_custom_error(): def test_get_error_message_and_location_and_value_from_a_custom_error():
error_string = "('error message', 'location', 'value')" error_string = "('error message', 'location', 'value')"
result = uc.get_error_message_and_location_and_value_from_a_custom_error( result = cli.get_error_message_and_location_and_value_from_a_custom_error(
error_string error_string
) )
assert result == ("error message", "location", "value") assert result == ("error message", "location", "value")
error_string = "error message" error_string = "error message"
result = uc.get_error_message_and_location_and_value_from_a_custom_error( result = cli.get_error_message_and_location_and_value_from_a_custom_error(
error_string error_string
) )
assert result is None assert result is None
@ -41,7 +45,7 @@ def test_handle_validation_error(invalid_entries):
try: try:
entry_type(**entry) entry_type(**entry)
except pydantic.ValidationError as e: except pydantic.ValidationError as e:
uc.handle_validation_error(e) cli.handle_validation_error(e)
@pytest.mark.parametrize( @pytest.mark.parametrize(
@ -49,7 +53,7 @@ def test_handle_validation_error(invalid_entries):
[ruamel.yaml.YAMLError, RuntimeError], [ruamel.yaml.YAMLError, RuntimeError],
) )
def test_handle_exceptions(exception): def test_handle_exceptions(exception):
@uc.handle_exceptions @cli.handle_exceptions
def function_that_raises_exception(): def function_that_raises_exception():
raise exception("This is an exception!") raise exception("This is an exception!")
@ -57,7 +61,7 @@ def test_handle_exceptions(exception):
def test_live_progress_reporter_class(): def test_live_progress_reporter_class():
with uc.LiveProgressReporter(number_of_steps=3) as progress: with cli.LiveProgressReporter(number_of_steps=3) as progress:
progress.start_a_step("Test step 1") progress.start_a_step("Test step 1")
progress.finish_the_current_step() progress.finish_the_current_step()

View File

@ -152,7 +152,7 @@ def test_generate_json_schema_file(tmp_path):
assert schema_file_path.exists() assert schema_file_path.exists()
schema_text = schema_file_path.read_text() schema_text = schema_file_path.read_text(encoding="utf-8")
schema = json.loads(schema_text) schema = json.loads(schema_text)
assert isinstance(schema, dict) assert isinstance(schema, dict)