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
5 changes: 3 additions & 2 deletions simple_typing_application/key_monitor/factory.py
Original file line number Diff line number Diff line change
Expand Up @@ -37,10 +37,11 @@ def create_key_monitor(

# create key monitor
logger.debug(f"create {key_monitor_cls.__name__}")
key_monitor_config: BaseKeyMonitorConfigModel
if isinstance(config, key_monitor_config_model):
key_monitor_config: BaseKeyMonitorConfigModel = config
key_monitor_config = config
else:
key_monitor_config: BaseKeyMonitorConfigModel = key_monitor_config_model(**config.model_dump()) # noqa
key_monitor_config = key_monitor_config_model(**config.model_dump()) # noqa
key_monitor: BaseKeyMonitor = key_monitor_cls(**key_monitor_config.model_dump()) # type: ignore # noqa

return key_monitor
5 changes: 3 additions & 2 deletions simple_typing_application/sentence_generator/factory.py
Original file line number Diff line number Diff line change
Expand Up @@ -50,10 +50,11 @@ def create_sentence_generator(

# create sentence generator
logger.debug(f"create {sentence_generator_cls.__name__}")
sentence_generator_config: BaseSentenceGeneratorConfigModel
if isinstance(config, sentence_generator_config_model):
sentence_generator_config: BaseSentenceGeneratorConfigModel = config
sentence_generator_config = config
else:
sentence_generator_config: BaseSentenceGeneratorConfigModel = sentence_generator_config_model(**config.model_dump()) # noqa
sentence_generator_config = sentence_generator_config_model(**config.model_dump()) # noqa
sentence_generator: BaseSentenceGenerator = sentence_generator_cls(**sentence_generator_config.model_dump()) # type: ignore # noqa

return sentence_generator
Original file line number Diff line number Diff line change
Expand Up @@ -90,8 +90,8 @@ async def generate(
self._logger.debug(f"agent input messages: {messages}")
with stopwatch(level=DEBUG, logger=self._logger, prefix="OpenAI agent invocation"):
ret: dict[str, Any] = await self._agent.ainvoke(
{"messages": messages},
) # type: ignore
{"messages": messages}, # type: ignore
)
self._logger.debug(f"agent response: {ret}")

# store to memory
Expand Down
5 changes: 3 additions & 2 deletions simple_typing_application/ui/factory.py
Original file line number Diff line number Diff line change
Expand Up @@ -32,10 +32,11 @@ def create_user_interface(

# create user interface
logger.debug(f"create {user_interface_cls.__name__}")
user_interface_config: BaseUserInterfaceConfigModel
if isinstance(config, user_interface_config_model):
user_interface_config: BaseUserInterfaceConfigModel = config
user_interface_config = config
else:
user_interface_config: BaseUserInterfaceConfigModel = user_interface_config_model(**config.model_dump()) # noqa
user_interface_config = user_interface_config_model(**config.model_dump()) # noqa
user_interface: BaseUserInterface = user_interface_cls(**user_interface_config.model_dump()) # type: ignore # noqa

return user_interface
55 changes: 52 additions & 3 deletions simple_typing_application/utils/stopwatch.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,7 +3,7 @@
from contextlib import contextmanager
from functools import wraps, partial
from logging import Logger, getLogger
from typing import Callable, Generator, TypeVar, ParamSpec
from typing import Callable, Generator, TypeVar, ParamSpec, overload

T = TypeVar("T")
P = ParamSpec("P")
Expand Down Expand Up @@ -67,8 +67,9 @@ def stopwatch(
log_func(log_msg_fmt.format(elapsed_time=end_time - start_time))


@overload
def stopwatch_deco(
func: Callable[P, T] | None = None,
func: Callable[P, T],
*,
level: int = logging.INFO,
prefix: str | None = None,
Expand All @@ -77,9 +78,57 @@ def stopwatch_deco(
) -> Callable[P, T]:
"""Decorator to measure the execution time of a function.

Args:
func (Callable[P, T]): function to be decorated.
level (int, optional): log level. Defaults to logging.INFO.
Must be one of logging.DEBUG, logging.INFO, logging.WARNING, logging.ERROR, logging.CRITICAL.
prefix (str | None, optional): prefix of the log message. Defaults to None.
postfix (str, optional): postfix of the log message. Defaults to "".
logger (Logger, optional): logger. Defaults to logger.

Returns:
Callable[P, T]: decorated function.
""" # noqa
...


@overload
def stopwatch_deco(
func: None = None,
*,
level: int = logging.INFO,
prefix: str | None = None,
postfix: str = "",
logger: Logger = logger,
) -> Callable[[Callable[P, T]], Callable[P, T]]:
"""Decorator to measure the execution time of a function.

Args:
func (None): Not used in this overload. This signature is for parameterized decoration (e.g., @stopwatch_deco(level=DEBUG)).
level (int, optional): log level. Defaults to logging.INFO.
Must be one of logging.DEBUG, logging.INFO, logging.WARNING, logging.ERROR, logging.CRITICAL.
prefix (str | None, optional): prefix of the log message. Defaults to None.
postfix (str, optional): postfix of the log message. Defaults to "".
logger (Logger, optional): logger. Defaults to logger.

Returns:
Callable[[Callable[P, T]], Callable[P, T]]: decorator.
""" # noqa
...


def stopwatch_deco(
func: Callable[P, T] | None = None,
*,
level: int = logging.INFO,
prefix: str | None = None,
postfix: str = "",
logger: Logger = logger,
) -> Callable[P, T] | Callable[[Callable[P, T]], Callable[P, T]]:
"""Decorator to measure the execution time of a function.

Args:
func (Callable[P, T], optional): function to be decorated. Defaults to None.
*,
level (int, optional): log level. Defaults to logging.INFO.
Must be one of logging.DEBUG, logging.INFO, logging.WARNING, logging.ERROR, logging.CRITICAL.
prefix (str | None, optional): prefix of the log message. Defaults to None.
Expand Down
Loading