Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

fix(config): update mypy configuration #117

Merged
merged 9 commits into from
Oct 7, 2024
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
8 changes: 8 additions & 0 deletions pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -67,7 +67,15 @@ testpaths = [
]

[tool.mypy]
python_version = "3.8"
check_untyped_defs = true
strict = true
ignore_missing_imports = true
warn_unused_ignores = true
warn_redundant_casts = true
warn_unused_configs = true
show_error_codes = true
exclude = ["scripts/"]


[tool.ruff]
Expand Down
46 changes: 28 additions & 18 deletions src/makim/cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -12,7 +12,8 @@

from fuzzywuzzy import process

from makim import Makim, __version__
from makim import __version__
from makim.core import Makim

CLI_ROOT_FLAGS_VALUES_COUNT = {
'--dry-run': 0,
Expand All @@ -33,7 +34,7 @@
),
)

makim = Makim()
makim: Makim = Makim()


@app.callback(invoke_without_command=True)
Expand Down Expand Up @@ -76,7 +77,7 @@ def main(
raise typer.Exit(0)


def suggest_command(user_input: str, available_commands: list) -> str:
def suggest_command(user_input: str, available_commands: list[str]) -> str:
"""
Suggest the closest command to the user input using fuzzy search.

Expand All @@ -90,10 +91,10 @@ def suggest_command(user_input: str, available_commands: list) -> str:
str: The suggested command.
"""
suggestion, _ = process.extractOne(user_input, available_commands)
return suggestion
return str(suggestion)


def map_type_from_string(type_name) -> Type:
def map_type_from_string(type_name: str) -> Type[Union[str, int, float, bool]]:
"""
Return a type object mapped from the type name.

Expand All @@ -119,7 +120,7 @@ def map_type_from_string(type_name) -> Type:
return type_mapping.get(type_name, str)


def normalize_string_type(type_name) -> str:
def normalize_string_type(type_name: str) -> str:
"""
Normalize the user type definition to the correct name.

Expand Down Expand Up @@ -151,9 +152,14 @@ def get_default_value(
) -> Optional[Union[str, int, float, bool]]:
"""Return the default value regarding its type in a string format."""
if arg_type == 'bool':
return False

return value
return False if value is None else bool(value)
elif arg_type == 'int':
return int(value) if value is not None else None
elif arg_type == 'float':
return float(value) if value is not None else None
elif arg_type == 'str':
return str(value) if value is not None else None
return None


def get_default_value_str(arg_type: str, value: Any) -> str:
Expand All @@ -167,7 +173,7 @@ def get_default_value_str(arg_type: str, value: Any) -> str:
return f'{value or 0}'


def create_args_string(args: Dict[str, str]) -> str:
def create_args_string(args: dict[str, str]) -> str:
"""Return a string for arguments for a function for typer."""
args_rendered = []

Expand Down Expand Up @@ -208,8 +214,8 @@ def create_args_string(args: Dict[str, str]) -> str:


def apply_click_options(
command_function: Callable, options: Dict[str, Any]
) -> Callable:
command_function: Callable[..., Any], options: dict[str, Any]
) -> Callable[..., Any]:
"""
Apply Click options to a Typer command function.

Expand All @@ -226,7 +232,9 @@ def apply_click_options(
The command function with options applied.
"""
for opt_name, opt_details in options.items():
opt_args: dict[str, Optional[Union[str, int, float, bool, Type]]] = {}
opt_args: dict[
str, Optional[Union[str, int, float, bool, Type[Any]]]
] = {}

opt_data = cast(Dict[str, str], opt_details)
opt_type_str = normalize_string_type(opt_data.get('type', 'str'))
Expand Down Expand Up @@ -255,7 +263,7 @@ def apply_click_options(
return command_function


def create_dynamic_command(name: str, args: Dict[str, str]) -> None:
def create_dynamic_command(name: str, args: dict[str, str]) -> None:
"""
Dynamically create a Typer command with the specified options.

Expand Down Expand Up @@ -295,7 +303,7 @@ def create_dynamic_command(name: str, args: Dict[str, str]) -> None:

function_code += f' makim.run({args_param_str})\n'

local_vars: Dict[str, Any] = {}
local_vars: dict[str, Any] = {}
exec(function_code, globals(), local_vars)
dynamic_command = decorator(local_vars['dynamic_command'])

Expand All @@ -307,7 +315,7 @@ def create_dynamic_command(name: str, args: Dict[str, str]) -> None:

def extract_root_config(
cli_list: list[str] = sys.argv,
) -> Dict[str, str | bool]:
) -> dict[str, str | bool]:
"""Extract the root configuration from the CLI."""
params = cli_list[1:]

Expand Down Expand Up @@ -399,7 +407,7 @@ def run_app() -> None:

# create tasks data
# group_names = list(makim.global_data.get('groups', {}).keys())
tasks: Dict[str, Any] = {}
tasks: dict[str, Any] = {}
for group_name, group_data in makim.global_data.get('groups', {}).items():
for task_name, task_data in group_data.get('tasks', {}).items():
tasks[f'{group_name}.{task_name}'] = task_data
Expand All @@ -417,7 +425,9 @@ def run_app() -> None:

command_used = _get_command_from_cli()

available_cmds = [cmd.name for cmd in app.registered_commands]
available_cmds = [
cmd.name for cmd in app.registered_commands if cmd.name is not None
]
suggestion = suggest_command(command_used, available_cmds)

typer.secho(
Expand Down
6 changes: 5 additions & 1 deletion src/makim/console.py
Original file line number Diff line number Diff line change
@@ -1,9 +1,13 @@
"""Functions about console."""

from __future__ import annotations

import os


def get_terminal_size(default_size=(80, 24)):
def get_terminal_size(
default_size: tuple[int, int] = (80, 24),
) -> tuple[int, int]:
"""Return the height (number of lines) of the terminal using os module."""
try:
size = os.get_terminal_size()
Expand Down
70 changes: 39 additions & 31 deletions src/makim/core.py
Original file line number Diff line number Diff line change
Expand Up @@ -95,25 +95,25 @@ class Makim:
file: str = '.makim.yaml'
dry_run: bool = False
verbose: bool = False
global_data: dict = {}
global_data: dict[str, Any] = {}
shell_app: sh.Command = sh.xonsh
shell_args: list[str] = []
tmp_suffix: str = '.makim'

# temporary variables
env: dict = {} # initial env
env_scoped: dict = {} # current env
env: dict[str, Any] = {} # initial env
env_scoped: dict[str, Any] = {} # current env
# initial working directory
working_directory: Optional[Path] = None
# current working directory
working_directory_scoped: Optional[Path] = None
args: Optional[object] = None
args: Optional[dict[str, Any]] = None
group_name: str = 'default'
group_data: dict = {}
group_data: dict[str, Any] = {}
task_name: str = ''
task_data: dict = {}
task_data: dict[str, Any] = {}

def __init__(self):
def __init__(self) -> None:
"""Prepare the Makim class with the default configuration."""
os.environ['RAISE_SUBPROC_ERROR'] = '1'
os.environ['XONSH_SHOW_TRACEBACK'] = '0'
Expand All @@ -126,7 +126,7 @@ def __init__(self):
self.shell_args: list[str] = []
self.tmp_suffix: str = '.makim'

def _call_shell_app(self, cmd):
def _call_shell_app(self, cmd: str) -> None:
self._load_shell_app()

fd, filepath = tempfile.mkstemp(suffix=self.tmp_suffix, text=True)
Expand Down Expand Up @@ -170,7 +170,7 @@ def _call_shell_app(self, cmd):
def _check_makim_file(self, file_path: str = '') -> bool:
return Path(file_path or self.file).exists()

def _verify_task_conditional(self, conditional) -> bool:
def _verify_task_conditional(self, conditional: Any) -> bool:
# todo: implement verification
print(f'condition {conditional} not verified')
return False
Expand Down Expand Up @@ -208,7 +208,7 @@ def _change_task(self, task_name: str) -> None:
MakimError.MAKIM_TARGET_NOT_FOUND,
)

def _change_group_data(self, group_name=None) -> None:
def _change_group_data(self, group_name: Optional[str] = None) -> None:
groups = self.global_data['groups']

if group_name is not None:
Expand Down Expand Up @@ -342,7 +342,7 @@ def _load_shell_app(self) -> None:
self.shell_args = cmd_args
self.tmp_suffix = cmd_tmp_suffix

def _load_dotenv(self, data_scope: dict) -> dict[str, str]:
def _load_dotenv(self, data_scope: dict[str, Any]) -> dict[str, str]:
env_file = data_scope.get('env-file')
if not env_file:
return {}
Expand All @@ -369,8 +369,11 @@ def _load_scoped_data(
raise Exception(f'The given scope `{scope}` is not valid.')

def _render_env_inplace(
env_user: dict, env_file: dict, variables: dict, env: dict
):
env_user: dict[str, Any],
env_file: dict[str, Any],
variables: dict[str, Any],
env: dict[str, Any],
) -> None:
env.update(env_file)
for k, v in env_user.items():
env[k] = TEMPLATE.from_string(str(v)).render(
Expand All @@ -380,29 +383,29 @@ def _render_env_inplace(
scope_id = scope_options.index(scope)

env = deepcopy(dict(os.environ))
variables: dict = {}
variables: dict[str, Any] = {}

if scope_id >= SCOPE_GLOBAL:
env_user = self.global_data.get('env', {})
env_file = self._load_dotenv(self.global_data)
_render_env_inplace(env_user, env_file, variables, env)
variables.update(self._load_scoped_vars('global', env=env))
variables.update(self._load_scoped_vars('global'))

if scope_id >= SCOPE_GROUP:
env_user = self.group_data.get('env', {})
env_file = self._load_dotenv(self.group_data)
_render_env_inplace(env_user, env_file, variables, env)
variables.update(self._load_scoped_vars('group', env=env))
variables.update(self._load_scoped_vars('group'))

if scope_id == SCOPE_TARGET:
env_user = self.task_data.get('env', {})
env_file = self._load_dotenv(self.task_data)
_render_env_inplace(env_user, env_file, variables, env)
variables.update(self._load_scoped_vars('task', env=env))
variables.update(self._load_scoped_vars('task'))

return env, variables

def _load_scoped_vars(self, scope: str, env) -> dict:
def _load_scoped_vars(self, scope: str) -> dict[str, Any]:
scope_options = ('global', 'group', 'task')
if scope not in scope_options:
raise Exception(f'The given scope `{scope}` is not valid.')
Expand Down Expand Up @@ -432,21 +435,24 @@ def _load_scoped_vars(self, scope: str, env) -> dict:
}
)

return fix_dict_keys_recursively(variables)
return cast(Dict[str, Any], fix_dict_keys_recursively(variables))

def _load_task_args(self):
def _load_task_args(self) -> None:
if self.args is None:
self.args = {}
for name, value in self.task_data.get('args', {}).items():
qualified_name = f'--{name}'
if self.args.get(qualified_name):
continue
default = value.get('default')
is_bool = value.get('type', '') == 'bool'
self.args[qualified_name] = (
default if default is not None else False if is_bool else None
)
if qualified_name not in self.args:
default = value.get('default')
is_bool = value.get('type', '') == 'bool'
self.args[qualified_name] = (
default
if default is not None
else (False if is_bool else None)
)

# run commands
def _run_hooks(self, args: dict, hook_type: str):
def _run_hooks(self, args: dict[str, Any], hook_type: str) -> None:
if not self.task_data.get('hooks', {}).get(hook_type):
return
makim_hook = deepcopy(self)
Expand Down Expand Up @@ -510,7 +516,7 @@ def _run_hooks(self, args: dict, hook_type: str):

makim_hook.run(deepcopy(args_hook))

def _run_command(self, args: dict):
def _run_command(self, args: dict[str, Any]) -> None:
cmd = self.task_data.get('run', '').strip()
if 'vars' not in self.group_data:
self.group_data['vars'] = {}
Expand Down Expand Up @@ -589,7 +595,9 @@ def _run_command(self, args: dict):

# public methods

def load(self, file: str, dry_run: bool = False, verbose: bool = False):
def load(
self, file: str, dry_run: bool = False, verbose: bool = False
) -> None:
"""Load makim configuration."""
self.file = file
self.dry_run = dry_run
Expand All @@ -599,7 +607,7 @@ def load(self, file: str, dry_run: bool = False, verbose: bool = False):
self._verify_config()
self.env = self._load_dotenv(self.global_data)

def run(self, args: dict):
def run(self, args: dict[str, Any]) -> None:
"""Run makim task code."""
self.args = args

Expand Down
6 changes: 3 additions & 3 deletions src/makim/logs.py
Original file line number Diff line number Diff line change
Expand Up @@ -28,20 +28,20 @@ class MakimLogs:
@staticmethod
def raise_error(
message: str, message_type: MakimError, command_error: int = 1
):
) -> None:
"""Print error message and exit with given error code."""
console = Console(stderr=True, style='bold red')
console.print(f'Makim Error #{message_type.value}: {message}')
raise os._exit(command_error)

@staticmethod
def print_info(message: str):
def print_info(message: str) -> None:
"""Print info message."""
console = Console(style='blue')
console.print(message)

@staticmethod
def print_warning(message: str):
def print_warning(message: str) -> None:
"""Print warning message."""
console = Console(style='yellow')
console.print(message)
Loading