Skip to content

Commit

Permalink
More typing fixes
Browse files Browse the repository at this point in the history
  • Loading branch information
venekamp committed Jul 19, 2024
1 parent 0d3f044 commit 092e7d8
Show file tree
Hide file tree
Showing 6 changed files with 196 additions and 99 deletions.
14 changes: 9 additions & 5 deletions SRAMsync/cba_script_generator.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,7 +6,7 @@ class is derived from CuaScriptGenerator, because of its strong relation. By

import json
import sys
from typing import Any, Callable, Union
from typing import Any, Callable, Union, cast

import click
from pathlib import Path
Expand All @@ -17,7 +17,7 @@ class is derived from CuaScriptGenerator, because of its strong relation. By
from SRAMsync.sramlogger import logger
from SRAMsync.state import State
from SRAMsync.sync_with_sram import ConfigValidationError
from SRAMsync.typing import EventHandlerConfig
from SRAMsync.typing import CbaNotificationConfig, EventHandlerConfig


class CbaScriptGenerator(CuaScriptGenerator):
Expand All @@ -40,19 +40,23 @@ class CbaScriptGenerator(CuaScriptGenerator):
"required": ["cba_add_cmd", "cba_del_cmd", "cba_machine", "cba_budget_account", "cua_config"],
}

def __init__(self, service: str, cfg: dict[str, Any], state: JsonFile, path: Path) -> None:
def __init__(self, service: str, cfg: EventHandlerConfig, state: JsonFile, path: Path) -> None:
try:
validate(
schema=CbaScriptGenerator._schema,
instance=cfg["event_handler_config"],
format_checker=Draft202012Validator.FORMAT_CHECKER,
)

cua_config: EventHandlerConfig = {
"event_handler_config": cfg["event_handler_config"]["cua_config"],
"event_handler_config": cast(CbaNotificationConfig, cfg["event_handler_config"])[
"cua_config"
],
"secrets": cfg["secrets"],
}
super().__init__(service, cfg=cua_config, state=state, cfg_path=path)
self.cfg = cfg["event_handler_config"]

self.cfg = cast(CbaNotificationConfig, cfg["event_handler_config"])
self._cba_co_budget_mapping_filename = ""
except ConfigValidationError as e:
raise e
Expand Down
6 changes: 3 additions & 3 deletions SRAMsync/config.py
Original file line number Diff line number Diff line change
Expand Up @@ -18,7 +18,7 @@
from SRAMsync.event_handler_proxy import EventHandlerProxy
from SRAMsync.sramlogger import logger
from SRAMsync.state import NoGracePeriodForGroupError, State
from SRAMsync.typing import StateFile, StatusConfig, EventHandlerConfig, ConfigGroup
from SRAMsync.typing import Secrets, StateFile, StatusConfig, EventHandlerConfig, ConfigGroup

import SRAMsync.typing

Expand Down Expand Up @@ -185,7 +185,7 @@ def __init__(self, config_file: str, args: Union[dict[str, str], None] = None) -

self.config = config

self.secrets: dict[str, dict[str, str]] = {}
self.secrets: Secrets = cast(Secrets, {})
if "secrets" in config:
with open(file=config["secrets"]["file"], encoding="utf8") as fd:
self.secrets = yaml.safe_load(stream=fd)
Expand Down Expand Up @@ -231,7 +231,7 @@ def get_event_handlers(self, state: State, args: dict[str, str]) -> list[EventHa
EventHandler, deduct_event_handler_class(event_handler_full_name=event_handler["name"])
)

event_handler_cfg = cast(EventHandlerConfig, {})
event_handler_cfg: EventHandlerConfig = cast(EventHandlerConfig, {})
if "config" in event_handler:
event_handler_cfg["event_handler_config"] = event_handler["config"]

Expand Down
4 changes: 2 additions & 2 deletions SRAMsync/cua_script_generator.py
Original file line number Diff line number Diff line change
Expand Up @@ -24,7 +24,7 @@
from SRAMsync.json_file import JsonFile
from SRAMsync.sramlogger import logger
from SRAMsync.sync_with_sram import ConfigValidationError
from SRAMsync.typing import EventHandlerConfig
from SRAMsync.typing import CuaNotificationsConfig, EventHandlerConfig


class CuaScriptGenerator(EventHandler):
Expand Down Expand Up @@ -62,7 +62,7 @@ def __init__(self, service: str, cfg: EventHandlerConfig, state: JsonFile, cfg_p
self.run = False
self.org_co_uuids = cast(dict[str, str], {})

self.cfg = cfg["event_handler_config"]
self.cfg = cast(CuaNotificationsConfig, cfg["event_handler_config"])
self.state = state
self.script_name = render_templated_string(template_string=self.cfg["filename"], service=service)
self.script_file_descriptor = open(file=self.script_name, mode="w+", encoding="utf8")
Expand Down
18 changes: 10 additions & 8 deletions SRAMsync/dummy_event_handler.py
Original file line number Diff line number Diff line change
Expand Up @@ -43,7 +43,7 @@ def add_new_user(
self,
entry: dict[str, list[bytes]],
**kwargs: str,
):
) -> None:
"""Log the add_new_user event."""
org: str = kwargs["org"]
co: str = kwargs["co"]
Expand All @@ -68,7 +68,7 @@ def get_supported_arguments(
) -> dict[str, dict[str, Union[Union[Callable[[str], None], Callable[[], None]], str]]]:
return {}

def add_public_ssh_key(self, co: str, user: str, key: str):
def add_public_ssh_key(self, co: str, user: str, key: str) -> None:
"""Log the add_public_ssh_key event."""
logger.info(
" add_public_ssh_key(%s, %s, %s)",
Expand All @@ -77,11 +77,11 @@ def add_public_ssh_key(self, co: str, user: str, key: str):
click.style(text=key, fg="white", dim=True),
)

def delete_public_ssh_key(self, co: str, user: str, key: str):
def delete_public_ssh_key(self, co: str, user: str, key: str) -> None:
"""Log the delete_public_ssh_key event."""
logger.info(" delete_public_ssh_key(%s, %s, %s)", co, user, key)

def add_new_groups(self, co: str, groups: list[str], group_attributes: list[str]):
def add_new_groups(self, co: str, groups: list[str], group_attributes: list[str]) -> None:
"""Log the add_new_group event."""
logger.info(
" add_new_group(%s, %s, %s)",
Expand All @@ -90,7 +90,7 @@ def add_new_groups(self, co: str, groups: list[str], group_attributes: list[str]
group_attributes,
)

def remove_group(self, co: str, group: str, group_attributes: list[str]):
def remove_group(self, co: str, group: str, group_attributes: list[str]) -> None:
"""Log the remove_group event."""
logger.info(
" remove_group(%s, %s, %s)",
Expand All @@ -99,7 +99,7 @@ def remove_group(self, co: str, group: str, group_attributes: list[str]):
group_attributes,
)

def add_user_to_group(self, **kwargs: str):
def add_user_to_group(self, **kwargs: str) -> None:
"""Log the add_user_to_group event."""
try:
org: str = kwargs["org"]
Expand Down Expand Up @@ -133,7 +133,7 @@ def start_grace_period_for_user(
duration,
)

def remove_user_from_group(self, co: str, group: str, group_attributes: list[str], user: str):
def remove_user_from_group(self, co: str, group: str, group_attributes: list[str], user: str) -> None:
"""Log the remove_user_from_group event."""
logger.info(
" remove_user_from_group(%s, %s, %s, %s)",
Expand All @@ -143,7 +143,9 @@ def remove_user_from_group(self, co: str, group: str, group_attributes: list[str
click.style(text=user, fg="cyan"),
)

def remove_graced_user_from_group(self, co: str, group: str, group_attributes: list[str], user: str):
def remove_graced_user_from_group(
self, co: str, group: str, group_attributes: list[str], user: str
) -> None:
"""Log the remove_graced_user_from_group event."""
logger.info(
" remove_graced_user_from_group(%s, %s, %s, %s)",
Expand Down
Loading

0 comments on commit 092e7d8

Please sign in to comment.