Last active
April 17, 2024 03:44
-
-
Save toolittlecakes/d538dfedb2169f9e3ccea6bb83b11f5a to your computer and use it in GitHub Desktop.
Inspired by this article about unobvious usage of polymorphism in order to embrace functional approach to create alternative solution
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| import syslog | |
| from dataclasses import dataclass | |
| from typing import Callable, Generic, Literal, Protocol, Type, TypeVar, TypeVarTuple | |
| # Initial solution with classes | |
| # | |
| # class Filter(Protocol): | |
| # def filter(self, message: str) -> bool: ... | |
| # class TextFilter(Filter): | |
| # def __init__(self, pattern: str): | |
| # self.pattern = pattern | |
| # def filter(self, message: str): | |
| # return self.pattern in message | |
| # class Handler(Protocol): | |
| # def emit(self, message: str): ... | |
| # class FileHandler(Handler): | |
| # def __init__(self, file): | |
| # self.file = file | |
| # def emit(self, message): | |
| # self.file.write(message + "\n") | |
| # self.file.flush() | |
| # class SocketHandler(Handler): | |
| # def __init__(self, sock): | |
| # self.sock = sock | |
| # def emit(self, message): | |
| # self.sock.sendall((message + "\n").encode("ascii")) | |
| # class SyslogHandler(Handler): | |
| # def __init__(self, priority): | |
| # self.priority = priority | |
| # def emit(self, message): | |
| # syslog.syslog(self.priority, message) | |
| # class Logger: | |
| # def __init__(self, filters, handlers): | |
| # self.filters = filters | |
| # self.handlers = handlers | |
| # def log(self, message): | |
| # if not all(filter(message) for filter in self.filters): | |
| # return | |
| # for handler in self.handlers: | |
| # handler(message) | |
| # ---------------------------- | |
| # Simple Version | |
| # Filter = Callable[[str], bool] | |
| # Emitter = Callable[[str], None] | |
| # ---------------------------- | |
| # Custom Version (functions only) | |
| # def function_protocol[F: Callable](_: F) -> type[F]: ... | |
| # def implement_protocol[P](_: Type[P]) -> Callable[[P], P]: | |
| # def decorator[T](func: T) -> T: | |
| # return func | |
| # return decorator | |
| # @function_protocol | |
| # def Filter(message: str) -> bool: ... | |
| # @function_protocol | |
| # def Emitter(message: str) -> None: ... | |
| # ---------------------------- | |
| # More static checks (analog of Protocol for classes (implicit)): | |
| class Filter(Protocol): | |
| def __call__(self, message: str) -> bool: ... | |
| class Emitter(Protocol): | |
| def __call__(self, message: str) -> None: ... | |
| def build_text_filter(pattern: str) -> Filter: | |
| def filter_text(message: str): | |
| return pattern in message | |
| return filter_text | |
| def build_file_emmiter(file) -> Emitter: | |
| def emit(message: str): | |
| file.write(message + "\n") | |
| file.flush() | |
| return emit | |
| def build_socket_emmiter(sock) -> Emitter: | |
| def emit(message: str): | |
| sock.sendall((message + "\n").encode("ascii")) | |
| return emit | |
| def build_syslog_emmiter(priority): | |
| def emit(message: str): | |
| syslog.syslog(priority, message) | |
| return emit | |
| # ---------------------------- | |
| # Even more static checks (with explicit Named Functors) | |
| # Analog of classes Interface (explicit) | |
| # class TypeWrapper[F: Callable]: | |
| # __call__: F | |
| # class CallWrapper[F: Callable]: | |
| # def __init__(self, func: F): | |
| # self.func = func | |
| # def __call__(self, *args, **kwargs): | |
| # return self.func(*args, **kwargs) | |
| # class Functor[F: Callable](TypeWrapper[F], CallWrapper[F]): ... | |
| # def adapter[F: Callable](_: F) -> type[Functor[F]]: | |
| # return Functor[F] | |
| # def _(message: str) -> bool: ... | |
| # class Filter(adapter(_)): ... | |
| # def _(message: str) -> None: ... | |
| # class Emitter(adapter(_)): ... | |
| # def build_text_filter(pattern: str) -> Filter: | |
| # def filter_text(message: str): | |
| # return pattern in message | |
| # return Filter(filter_text) | |
| # def build_file_emmiter(file) -> Emitter: | |
| # def emit(message: str): | |
| # file.write(message + "\n") | |
| # file.flush() | |
| # return Emitter(emit) | |
| # def build_socket_emmiter(sock) -> Emitter: | |
| # def emit(message: str): | |
| # sock.sendall((message + "\n").encode("ascii")) | |
| # return Emitter(emit) | |
| # def build_syslog_emmiter(priority): | |
| # def emit(message: str): | |
| # syslog.syslog(priority, message) | |
| # return Emitter(emit) | |
| # ---------------------------- | |
| def build_logger(filters: list[Filter], emmiters: list[Emitter]): | |
| def logger(message: str): | |
| if not all(filter(message) for filter in filters): | |
| return | |
| for emit in emmiters: | |
| emit(message) | |
| return logger | |
| log = build_logger( | |
| [build_text_filter("error")], | |
| [ | |
| build_file_emmiter(open("error.log", "a")), | |
| build_syslog_emmiter(syslog.LOG_ERR), | |
| ], | |
| ) | |
| log("error: something went wrong") | |
| log("info: everything is fine") |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment