Skip to content

Commit

Permalink
feat: make converters return multiple descriptors (#42)
Browse files Browse the repository at this point in the history
* feat: make converters return multiple descriptors

* fixes

* fixes

* fixes

* fixes

* fixes

* fixes
  • Loading branch information
jnicoulaud-ledger authored Oct 2, 2024
1 parent d268342 commit 5adf9b8
Show file tree
Hide file tree
Showing 17 changed files with 453 additions and 423 deletions.
178 changes: 178 additions & 0 deletions src/erc7730/common/output.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,178 @@
from abc import ABC, abstractmethod
from builtins import print as builtin_print
from enum import IntEnum, auto
from typing import assert_never, final, override

from pydantic import BaseModel, FilePath
from rich import print


class Output(BaseModel):
"""An output notice/warning/error."""

class Level(IntEnum):
"""ERC7730Linter output level."""

INFO = auto()
WARNING = auto()
ERROR = auto()

file: FilePath | None
line: int | None
title: str | None
message: str
level: Level = Level.ERROR


class OutputAdder(ABC):
"""An output notice/warning/error sink."""

@abstractmethod
def info(
self, message: str, file: FilePath | None = None, line: int | None = None, title: str | None = None
) -> None:
raise NotImplementedError()

@abstractmethod
def warning(
self, message: str, file: FilePath | None = None, line: int | None = None, title: str | None = None
) -> None:
raise NotImplementedError()

@abstractmethod
def error(
self, message: str, file: FilePath | None = None, line: int | None = None, title: str | None = None
) -> None:
raise NotImplementedError()


@final
class ListOutputAdder(OutputAdder, BaseModel):
"""An output adder that stores outputs in a list."""

outputs: list[Output] = []

@override
def info(
self, message: str, file: FilePath | None = None, line: int | None = None, title: str | None = None
) -> None:
self.outputs.append(Output(file=file, line=line, title=title, message=message, level=Output.Level.INFO))

@override
def warning(
self, message: str, file: FilePath | None = None, line: int | None = None, title: str | None = None
) -> None:
self.outputs.append(Output(file=file, line=line, title=title, message=message, level=Output.Level.WARNING))

@override
def error(
self, message: str, file: FilePath | None = None, line: int | None = None, title: str | None = None
) -> None:
self.outputs.append(Output(file=file, line=line, title=title, message=message, level=Output.Level.ERROR))


@final
class ConsoleOutputAdder(OutputAdder):
"""An output adder that prints to the console."""

@override
def info(
self, message: str, file: FilePath | None = None, line: int | None = None, title: str | None = None
) -> None:
self._log(Output.Level.INFO, message, file, line, title)

@override
def warning(
self, message: str, file: FilePath | None = None, line: int | None = None, title: str | None = None
) -> None:
self._log(Output.Level.WARNING, message, file, line, title)

@override
def error(
self, message: str, file: FilePath | None = None, line: int | None = None, title: str | None = None
) -> None:
self._log(Output.Level.ERROR, message, file, line, title)

@classmethod
def _log(
cls,
level: Output.Level,
message: str,
file: FilePath | None = None,
line: int | None = None,
title: str | None = None,
) -> None:
match level:
case Output.Level.INFO:
color = "blue"
case Output.Level.WARNING:
color = "yellow"
case Output.Level.ERROR:
color = "error"
case _:
assert_never(level)

log = f"[{color}]{level.name}"
if file is not None:
log += f": {file.name}"
if line is not None:
log += f" line {line}"
if title is not None:
log += f": {title}"
log += f"[/{color}]: {message}"

print(log)


@final
class GithubAnnotationsAdder(OutputAdder):
"""An output adder that formats errors to be parsed as Github annotations."""

@override
def info(
self, message: str, file: FilePath | None = None, line: int | None = None, title: str | None = None
) -> None:
self._log(Output.Level.INFO, message, file, line, title)

@override
def warning(
self, message: str, file: FilePath | None = None, line: int | None = None, title: str | None = None
) -> None:
self._log(Output.Level.WARNING, message, file, line, title)

@override
def error(
self, message: str, file: FilePath | None = None, line: int | None = None, title: str | None = None
) -> None:
self._log(Output.Level.ERROR, message, file, line, title)

@classmethod
def _log(
cls,
level: Output.Level,
message: str,
file: FilePath | None = None,
line: int | None = None,
title: str | None = None,
) -> None:
match level:
case Output.Level.INFO:
lvl = "notice"
case Output.Level.WARNING:
lvl = "warning"
case Output.Level.ERROR:
lvl = "error"
case _:
assert_never(level)

log = f"::{lvl} "
if file is not None:
log += f"file={file}"
if line is not None:
log += f",line={line}"
if title is not None:
log += f",title={title}"
message_formatted = message.replace("\n", "%0A")
log += f"::{message_formatted}"

builtin_print(log)
33 changes: 4 additions & 29 deletions src/erc7730/convert/__init__.py
Original file line number Diff line number Diff line change
@@ -1,9 +1,10 @@
from abc import ABC, abstractmethod
from enum import IntEnum, auto
from typing import Generic, TypeVar

from pydantic import BaseModel

from erc7730.common.output import OutputAdder

InputType = TypeVar("InputType", bound=BaseModel)
OutputType = TypeVar("OutputType", bound=BaseModel)

Expand All @@ -17,41 +18,15 @@ class ERC7730Converter(ABC, Generic[InputType, OutputType]):
"""

@abstractmethod
def convert(self, descriptor: InputType, error: "ErrorAdder") -> OutputType | None:
def convert(self, descriptor: InputType, out: OutputAdder) -> OutputType | dict[str, OutputType] | None:
"""
Convert a descriptor from/to ERC-7730.
Conversion may fail partially, in which case it should emit errors with ERROR level, or totally, in which case
it should emit errors with FATAL level.
:param descriptor: input descriptor to convert
:param error: error sink
:param out: output sink
:return: converted descriptor, or None if conversion failed
"""
raise NotImplementedError()

class Error(BaseModel):
"""ERC7730Converter output errors."""

class Level(IntEnum):
"""ERC7730Converter error level."""

WARNING = auto()
"""Indicates a non-fatal error: descriptor can be partially converted, but some parts will be lost."""

ERROR = auto()
"""Indicates a fatal error: descriptor cannot be converted."""

level: Level
message: str

class ErrorAdder(ABC):
"""ERC7730Converter output sink."""

@abstractmethod
def warning(self, message: str) -> None:
raise NotImplementedError()

@abstractmethod
def error(self, message: str) -> None:
raise NotImplementedError()
42 changes: 21 additions & 21 deletions src/erc7730/convert/convert.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,7 @@

from rich import print

from erc7730.common.output import ConsoleOutputAdder
from erc7730.common.pydantic import model_to_json_file
from erc7730.convert import ERC7730Converter, InputType, OutputType

Expand All @@ -18,40 +19,39 @@ def convert_to_file_and_print_errors(
:return: True if output file was written (if no errors, or only non-fatal errors encountered)
"""
if (output_descriptor := convert_and_print_errors(input_descriptor, converter)) is not None:
model_to_json_file(output_file, output_descriptor)
print("[green]Output descriptor file generated ✅[/green]")
if isinstance(output_descriptor, dict):
for identifier, descriptor in output_descriptor.items():
descriptor_file = output_file.with_suffix(f".{identifier}{output_file.suffix}")
model_to_json_file(descriptor_file, descriptor)
print(f"[green]generated {descriptor_file} ✅[/green]")
else:
model_to_json_file(output_file, output_descriptor)
print(f"[green]generated {output_file} ✅[/green]")
return True

print("[red]Conversion failed ❌[/red]")
print("[red]conversion failed ❌[/red]")
return False


def convert_and_print_errors(
input_descriptor: InputType, converter: ERC7730Converter[InputType, OutputType]
) -> OutputType | None:
) -> OutputType | dict[str, OutputType] | None:
"""
Convert an input descriptor using a converter, print any errors encountered, and return the result model.
:param input_descriptor: loaded, valid input descriptor
:param converter: converter to use
:return: output descriptor (if no errors, or only non-fatal errors encountered), None otherwise
"""
errors: list[ERC7730Converter.Error] = []

class ErrorAdder(ERC7730Converter.ErrorAdder):
def warning(self, message: str) -> None:
errors.append(ERC7730Converter.Error(level=ERC7730Converter.Error.Level.WARNING, message=message))

def error(self, message: str) -> None:
errors.append(ERC7730Converter.Error(level=ERC7730Converter.Error.Level.ERROR, message=message))

result = converter.convert(input_descriptor, ErrorAdder())

for error in errors:
match error.level:
case ERC7730Converter.Error.Level.WARNING:
print(f"[yellow][bold]{error.level}: [/bold]{error.message}[/yellow]")
case ERC7730Converter.Error.Level.ERROR:
print(f"[red][bold]{error.level}: [/bold]{error.message}[/red]")
result = converter.convert(input_descriptor, ConsoleOutputAdder())

if isinstance(result, dict):
match len(result):
case 0:
return None
case 1:
return next(iter(result.values()))
case _:
return result

return result
Loading

0 comments on commit 5adf9b8

Please sign in to comment.