-
Notifications
You must be signed in to change notification settings - Fork 11
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Add: Introduce a release Command "interface"
Add abstract base classes representing sync and async release related commands. This will allow to abstract the current sign and release commands. The base class also adds an error terminal and convenience methods for printing errors and warnings to the error terminal.
- Loading branch information
1 parent
d530abc
commit 434f596
Showing
1 changed file
with
63 additions
and
0 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,63 @@ | ||
# SPDX-FileCopyrightText: 2023 Greenbone AG | ||
# | ||
# SPDX-License-Identifier: GPL-3.0-or-later | ||
|
||
import asyncio | ||
from abc import ABC, abstractmethod | ||
from typing import Any | ||
|
||
from pontos.terminal import Terminal | ||
|
||
|
||
class Command(ABC): | ||
"""Base class for release related command""" | ||
|
||
def __init__(self, *, terminal: Terminal, error_terminal: Terminal) -> None: | ||
self.terminal = terminal | ||
self.error_terminal = error_terminal | ||
|
||
def print_error(self, *messages: Any, **kwargs: Any) -> None: | ||
"""Print an error to the error terminal""" | ||
self.error_terminal.error(*messages, **kwargs) | ||
|
||
def print_warning(self, *messages: Any, **kwargs: Any) -> None: | ||
"""Print a warning to the error console""" | ||
self.error_terminal.warning(*messages, **kwargs) | ||
|
||
@abstractmethod | ||
def run(self, **kwargs) -> int: | ||
"""Run the command""" | ||
|
||
|
||
class AsyncCommand(Command): | ||
"""Base class for release related commands using asyncio""" | ||
|
||
def run(self, **kwargs) -> int: | ||
""" | ||
Run the command using asyncio | ||
MUST NOT be called when an asyncio event loop is already running. | ||
""" | ||
return asyncio.run(self.async_run(**kwargs)) | ||
|
||
@abstractmethod | ||
async def async_run(self, **kwargs) -> int: | ||
""" | ||
Run the async command | ||
Gets called via run. Alternatively use similar code as the example. | ||
Example: | ||
.. code-block:: python | ||
import asyncio | ||
async def main(): | ||
cmd = MyAsyncCommand() | ||
task = asyncio.create_task(cmd.async_run(arg1, arg2)) | ||
... | ||
await task | ||
asyncio.run(main()) | ||
""" |