Skip to content

Commit

Permalink
fix(type-check): update mypy configuration (#117)
Browse files Browse the repository at this point in the history
  • Loading branch information
abhijeetSaroha authored Oct 7, 2024
1 parent be5c25b commit f2bf92e
Show file tree
Hide file tree
Showing 5 changed files with 83 additions and 53 deletions.
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)

0 comments on commit f2bf92e

Please sign in to comment.