Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions .gitignore
Original file line number Diff line number Diff line change
Expand Up @@ -318,3 +318,5 @@ http-client.private.env.json
# Apifox Helper cache
.idea/.cache/.Apifox_Helper
.idea/ApifoxUploaderProjectSetting.xml

.zed
2 changes: 1 addition & 1 deletion docs/root/api/app/index.rst
Original file line number Diff line number Diff line change
Expand Up @@ -30,7 +30,7 @@ App
system_router_title: str | None = "System points:",
ignore_command_register: bool = True,
dividing_line: AVAILABLE_DIVIDING_LINES = DEFAULT_DIVIDING_LINE,
repeat_command_groups_printing: bool = True,
repeat_command_groups_printing: bool = False,
override_system_messages: bool = False,
autocompleter: AutoCompleter = DEFAULT_AUTOCOMPLETER,
print_func: Printer = DEFAULT_PRINT_FUNC) -> None
Expand Down
4 changes: 4 additions & 0 deletions justfile
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,10 @@ tests:
tests-cov:
python -m pytest --cov=argenta tests

# Запустить тесты с отчетом о покрытии с html репортом
tests-cov-html:
python -m pytest --cov=argenta tests --cov-report=html

# Отформатировать код (Ruff + isort)
format:
python -m ruff format ./src
Expand Down
21 changes: 19 additions & 2 deletions mock/local_test.py
Original file line number Diff line number Diff line change
@@ -1,2 +1,19 @@
import sys
print(sys.version_info >= (3, 13))
from argenta import App, DataBridge, Response, Router
from argenta.di import FromDishka
from argenta.di.integration import setup_dishka, _auto_inject_handlers
from argenta.di.providers import SystemProvider
from dishka import make_container

container = make_container()

Response.patch_by_container(container)

app = App()
router = Router()

@router.command('command')
def handler(res: Response, data_bridge: FromDishka[DataBridge]):
print(data_bridge)

_auto_inject_handlers(app)
_auto_inject_handlers(app)
13 changes: 13 additions & 0 deletions pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -50,6 +50,19 @@ line-length=90
[tool.pyright]
typeCheckingMode = "strict"

[[tool.pyright.executionEnvironments]]
root = "tests/"
reportPrivateUsage = false
reportUnusedFunction = false

[tool.coverage.run]
branch = true
omit = [
"src/argenta/app/protocols.py",
"src/argenta/*/exceptions.py",
"src/argenta/metrics/*"
]

[tool.mypy]
disable_error_code = "import-untyped"

Expand Down
74 changes: 32 additions & 42 deletions src/argenta/app/models.py
Original file line number Diff line number Diff line change
Expand Up @@ -19,7 +19,6 @@
)
from argenta.app.registered_routers.entity import RegisteredRouters
from argenta.command.exceptions import (
EmptyInputCommandException,
InputCommandException,
RepeatedInputFlagsException,
UnprocessedInputFlagException,
Expand All @@ -40,7 +39,7 @@ def __init__(
initial_message: str,
farewell_message: str,
exit_command: Command,
system_router_title: str | None,
system_router_title: str,
ignore_command_register: bool,
dividing_line: StaticDividingLine | DynamicDividingLine,
repeat_command_groups_printing: bool,
Expand All @@ -51,7 +50,7 @@ def __init__(
self._prompt: str = prompt
self._print_func: Printer = print_func
self._exit_command: Command = exit_command
self._system_router_title: str | None = system_router_title
self._system_router_title: str = system_router_title
self._dividing_line: StaticDividingLine | DynamicDividingLine = dividing_line
self._ignore_command_register: bool = ignore_command_register
self._repeat_command_groups_printing_description: bool = repeat_command_groups_printing
Expand Down Expand Up @@ -144,8 +143,7 @@ def _print_command_group_description(self) -> None:
:return: None
"""
for registered_router in self.registered_routers:
if registered_router.title:
self._print_func(registered_router.title)
self._print_func(registered_router.title)
for command_handler in registered_router.command_handlers:
handled_command = command_handler.handled_command
self._print_func(
Expand Down Expand Up @@ -239,7 +237,7 @@ def _error_handler(self, error: InputCommandException, raw_command: str) -> None
self._incorrect_input_syntax_handler(raw_command)
elif isinstance(error, RepeatedInputFlagsException):
self._repeated_input_flags_handler(raw_command)
elif isinstance(error, EmptyInputCommandException):
else:
self._empty_input_command_handler()

def _setup_system_router(self) -> None:
Expand All @@ -253,9 +251,8 @@ def _setup_system_router(self) -> None:
def _(response: Response) -> None:
self._exit_command_handler(response)

if system_router not in self.registered_routers.registered_routers:
system_router.command_register_ignore = self._ignore_command_register
self.registered_routers.add_registered_router(system_router)
system_router.command_register_ignore = self._ignore_command_register
self.registered_routers.add_registered_router(system_router)

def _most_similar_command(self, unknown_command: str) -> str | None:
all_commands = list(self._current_matching_triggers_with_routers.keys())
Expand Down Expand Up @@ -323,23 +320,14 @@ def _pre_cycle_setup(self) -> None:
for router_entity in self.registered_routers:
router_triggers = router_entity.triggers
router_aliases = router_entity.aliases
combined = router_triggers + router_aliases
combined = router_triggers | router_aliases

for trigger in combined:
self._matching_default_triggers_with_routers[trigger] = router_entity
self._matching_lower_triggers_with_routers[trigger.lower()] = router_entity

self._autocompleter.initial_setup(list(self._current_matching_triggers_with_routers.keys()))

seen = {}
for item in list(self._current_matching_triggers_with_routers.keys()):
if item in seen:
Console().print(
f"\n[b red]WARNING:[/b red] Overlapping trigger or alias: [b blue]{item}[/b blue]"
)
else:
seen[item] = True

if not self._override_system_messages:
self._setup_default_view()

Expand All @@ -351,6 +339,28 @@ def _pre_cycle_setup(self) -> None:
print("\n")
if not self._repeat_command_groups_printing_description:
self._print_command_group_description()

def _process_exist_and_valid_command(self, input_command: InputCommand) -> None:
processing_router = self._current_matching_triggers_with_routers[input_command.trigger.lower()]

if processing_router.disable_redirect_stdout:
dividing_line_unit_part: str = self._dividing_line.get_unit_part()
self._print_func(
StaticDividingLine(dividing_line_unit_part).get_full_static_line(
is_override=self._override_system_messages
)
)
processing_router.finds_appropriate_handler(input_command)
self._print_func(
StaticDividingLine(dividing_line_unit_part).get_full_static_line(
is_override=self._override_system_messages
)
)
else:
with redirect_stdout(io.StringIO()) as stdout:
processing_router.finds_appropriate_handler(input_command)
stdout_result: str = stdout.getvalue()
self._print_framed_text(stdout_result)


AVAILABLE_DIVIDING_LINES: TypeAlias = StaticDividingLine | DynamicDividingLine
Expand All @@ -369,10 +379,10 @@ def __init__(
initial_message: str = "Argenta\n",
farewell_message: str = "\nSee you\n",
exit_command: Command = DEFAULT_EXIT_COMMAND,
system_router_title: str | None = "System points:",
system_router_title: str = "System points:",
ignore_command_register: bool = True,
dividing_line: AVAILABLE_DIVIDING_LINES = DEFAULT_DIVIDING_LINE,
repeat_command_groups_printing: bool = True,
repeat_command_groups_printing: bool = False,
override_system_messages: bool = False,
autocompleter: AutoCompleter = DEFAULT_AUTOCOMPLETER,
print_func: Printer = DEFAULT_PRINT_FUNC,
Expand Down Expand Up @@ -442,27 +452,7 @@ def run_polling(self) -> None:
self._print_framed_text(stdout_res)
continue

processing_router = self._current_matching_triggers_with_routers[input_command.trigger.lower()]

if processing_router.disable_redirect_stdout:
dividing_line_unit_part: str = self._dividing_line.get_unit_part()
self._print_func(
StaticDividingLine(dividing_line_unit_part).get_full_static_line(
is_override=self._override_system_messages
)
)
processing_router.finds_appropriate_handler(input_command)
self._print_func(
StaticDividingLine(dividing_line_unit_part).get_full_static_line(
is_override=self._override_system_messages
)
)
else:
with redirect_stdout(io.StringIO()) as stdout:
processing_router.finds_appropriate_handler(input_command)
stdout_result: str = stdout.getvalue()
if stdout_result:
self._print_framed_text(stdout_result)
self._process_exist_and_valid_command(input_command)

def include_router(self, router: Router) -> None:
"""
Expand Down
7 changes: 4 additions & 3 deletions src/argenta/app/protocols.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,11 +2,12 @@

from typing import Protocol, TypeVar


T = TypeVar("T", contravariant=True) # noqa: WPS111


class NonStandardBehaviorHandler(Protocol[T]):
def __call__(self, __param: T) -> None:
def __call__(self, _param: T, /) -> None:
raise NotImplementedError


Expand All @@ -16,10 +17,10 @@ def __call__(self) -> None:


class Printer(Protocol):
def __call__(self, __text: str) -> None:
def __call__(self, _text: str, /) -> None:
raise NotImplementedError


class DescriptionMessageGenerator(Protocol):
def __call__(self, __first_param: str, __second_param: str) -> str:
def __call__(self, _command: str, _description: str, /) -> str:
raise NotImplementedError
3 changes: 0 additions & 3 deletions src/argenta/app/registered_routers/entity.py
Original file line number Diff line number Diff line change
Expand Up @@ -24,6 +24,3 @@ def add_registered_router(self, router: Router, /) -> None:

def __iter__(self) -> Iterator[Router]:
return iter(self.registered_routers)

def __next__(self) -> Router:
return next(iter(self.registered_routers))
7 changes: 2 additions & 5 deletions src/argenta/command/flag/flags/models.py
Original file line number Diff line number Diff line change
Expand Up @@ -39,9 +39,6 @@ def __len__(self) -> int:
def __iter__(self) -> Iterator[FlagType]:
return iter(self.flags)

def __next__(self) -> FlagType:
return next(iter(self))

def __getitem__(self, flag_index: int) -> FlagType:
return self.flags[flag_index]

Expand All @@ -61,7 +58,7 @@ def get_flag_by_name(self, name: str) -> Flag | None:
@override
def __eq__(self, other: object) -> bool:
if not isinstance(other, Flags):
return NotImplemented
return False

if len(self.flags) != len(other.flags):
return False
Expand Down Expand Up @@ -91,7 +88,7 @@ def get_flag_by_name(self, name: str) -> InputFlag | None:
@override
def __eq__(self, other: object) -> bool:
if not isinstance(other, InputFlags):
raise NotImplementedError
return False

if len(self.flags) != len(other.flags):
return False
Expand Down
18 changes: 7 additions & 11 deletions src/argenta/command/flag/models.py
Original file line number Diff line number Diff line change
Expand Up @@ -43,20 +43,16 @@ def validate_input_flag_value(self, input_flag_value: str) -> bool:
Private. Validates the input flag value
:param input_flag_value: The input flag value to validate
:return: whether the entered flag is valid as bool
"""
if self.possible_values == PossibleValues.NEITHER:
return input_flag_value == ''

if self.possible_values == PossibleValues.ALL:
"""
if isinstance(self.possible_values, PossibleValues):
if self.possible_values == PossibleValues.NEITHER:
return input_flag_value == ''
return input_flag_value != ''

if isinstance(self.possible_values, Pattern):
return bool(self.possible_values.match(input_flag_value))

if isinstance(self.possible_values, list):
return input_flag_value in self.possible_values

return False
return input_flag_value in self.possible_values

@property
def string_entity(self) -> str:
Expand Down Expand Up @@ -88,9 +84,9 @@ def __init__(
self,
name: str,
*,
prefix: PREFIX_TYPE = "--",
input_value: str,
status: ValidationStatus | None,
prefix: PREFIX_TYPE = "--",
status: ValidationStatus | None = None,
):
"""
Public. The entity of the flag of the entered command
Expand Down
9 changes: 3 additions & 6 deletions src/argenta/command/models.py
Original file line number Diff line number Diff line change
Expand Up @@ -17,7 +17,7 @@
MIN_FLAG_PREFIX: str = "-"
PREFIX_TYPE = Literal["-", "--", "---"]
DEFAULT_WITHOUT_FLAGS: Flags = Flags()
DEFAULT_WITHOUT_ALIASES: list[Never] = []
DEFAULT_WITHOUT_ALIASES: set[Never] = set()

DEFAULT_WITHOUT_INPUT_FLAGS: InputFlags = InputFlags()

Expand All @@ -29,7 +29,7 @@ def __init__(
*,
description: str = "Some useful command",
flags: Flag | Flags = DEFAULT_WITHOUT_FLAGS,
aliases: list[str] | list[Never] = DEFAULT_WITHOUT_ALIASES,
aliases: set[str] | set[Never] = DEFAULT_WITHOUT_ALIASES,
):
"""
Public. The command that can and should be registered in the Router
Expand All @@ -41,7 +41,7 @@ def __init__(
self.registered_flags: Flags = flags if isinstance(flags, Flags) else Flags([flags])
self.trigger: str = trigger
self.description: str = description
self.aliases: list[str] | list[Never] = aliases
self.aliases: set[str] | set[Never] = aliases

def validate_input_flag(self, flag: InputFlag) -> ValidationStatus:
"""
Expand Down Expand Up @@ -104,9 +104,6 @@ def parse(cls, raw_command: str) -> Self:
else:
raise UnprocessedInputFlagException

if not name:
raise UnprocessedInputFlagException

if i + 1 < len(tokens) and not tokens[i + 1].startswith("-"):
input_value = tokens[i + 1]
i += 2
Expand Down
6 changes: 3 additions & 3 deletions src/argenta/di/integration.py
Original file line number Diff line number Diff line change
Expand Up @@ -20,16 +20,16 @@ def inject(func: Callable[..., T]) -> Callable[..., T]:


def setup_dishka(app: App, container: Container, *, auto_inject: bool = False) -> None:
Response.patch_by_container(container)
if auto_inject:
_auto_inject_handlers(app)
Response.patch_by_container(container)


def _get_container_from_response(args: tuple[Any, ...], kwargs: dict[str, Any]) -> Container:
for arg in args:
if isinstance(arg, Response):
if hasattr(arg, "_dishka_container"):
return arg._dishka_container # pyright: ignore[reportPrivateUsage]
if hasattr(arg, "__dishka_container__"):
return arg.__dishka_container__ # pyright: ignore[reportPrivateUsage]
break
raise RuntimeError("dishka container not found in Response")

Expand Down
2 changes: 1 addition & 1 deletion src/argenta/metrics/main.py
Original file line number Diff line number Diff line change
Expand Up @@ -9,7 +9,7 @@
from argenta import App


def get_time_of_pre_cycle_setup(app: App) -> float:
def get_time_of_pre_cycle_setup(app: App) -> float:
"""
Public. Return time of pre cycle setup
:param app: app instance for testing time of pre cycle setup
Expand Down
2 changes: 1 addition & 1 deletion src/argenta/orchestrator/argparser/entity.py
Original file line number Diff line number Diff line change
Expand Up @@ -94,7 +94,7 @@ def _parse_args(self) -> None:
namespace=self._core.parse_args(), processed_args=self.processed_args
)

def _register_args(self, processed_args: list[ValueArgument | BooleanArgument]) -> None:
def _register_args(self, processed_args: list[ValueArgument | BooleanArgument]) -> None: # pragma: no cover
if sys.version_info >= (3, 13):
for arg in processed_args:
if isinstance(arg, BooleanArgument):
Expand Down
Loading