diff --git a/code/newsletter/N4/switcher.py b/code/newsletter/N4/switcher.py new file mode 100644 index 0000000..4ceed89 --- /dev/null +++ b/code/newsletter/N4/switcher.py @@ -0,0 +1,38 @@ +class Device: + def __init__(self, name: str): + self.name = name + + def boot(self): + print(f"Device<{self.name}> is booting...") + + +class Switcher: + def __init__(self) -> None: + pass + + def __enter__(self): + self._on() + return self + + def __exit__(self, exc_type, exc_value, traceback): + self._off() + + def _on(self): + print(f"Loading switcher...") + + def _off(self): + print(f"Closing switcher...") + + def plug(self, device: Device): + device.boot() + + +def main(): + + with Switcher() as switcher: + device = Device("PC") + switcher.plug(device) + + +if __name__ == '__main__': + main() diff --git a/code/newsletter/N4/threadpool_simulation.py b/code/newsletter/N4/threadpool_simulation.py new file mode 100644 index 0000000..6cc3a3f --- /dev/null +++ b/code/newsletter/N4/threadpool_simulation.py @@ -0,0 +1,33 @@ +import contextlib + + +class Thread: + def __init__(self, max_worker): + self.max_worker = max_worker + + def execute(self, func, *args, **kwargs): + print(f"Using {self.max_worker} workers to execute {func.__name__} function...") + func(*args, **kwargs) + + +@contextlib.contextmanager +def ThreadPoolExecutor(max_worker=1): + pool = Thread(max_worker) + print("Initial for threads...") + try: + yield pool + finally: + print("Recycling threads and closing pool...") + + +def echo(s): + print(f"\techo: {s}") + + +def main(): + with ThreadPoolExecutor(max_worker=2) as pool: + pool.execute(echo, s="Hello, world!") + + +if __name__ == '__main__': + main()