#!/usr/bin/env PYTHONHASHSEED=1234 python3 # Copyright 2014-2019 Brett Slatkin, Pearson Education Inc. # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. # Reproduce book environment import random random.seed(1234) import logging from pprint import pprint from sys import stdout as STDOUT # Write all output to a temporary directory import atexit import gc import io import os import tempfile TEST_DIR = tempfile.TemporaryDirectory() atexit.register(TEST_DIR.cleanup) # Make sure Windows processes exit cleanly OLD_CWD = os.getcwd() atexit.register(lambda: os.chdir(OLD_CWD)) os.chdir(TEST_DIR.name) def close_open_files(): everything = gc.get_objects() for obj in everything: if isinstance(obj, io.IOBase): obj.close() atexit.register(close_open_files) # Example 1 try: class MyError(Exception): pass def my_generator(): yield 1 yield 2 yield 3 it = my_generator() print(next(it)) # Yield 1 print(next(it)) # Yield 2 print(it.throw(MyError('test error'))) except: logging.exception('Expected') else: assert False # Example 2 def my_generator(): yield 1 try: yield 2 except MyError: print('Got MyError!') else: yield 3 yield 4 it = my_generator() print(next(it)) # Yield 1 print(next(it)) # Yield 2 print(it.throw(MyError('test error'))) # Example 3 class Reset(Exception): pass def timer(period): current = period while current: current -= 1 try: yield current except Reset: current = period # Example 4 RESETS = [ False, False, False, True, False, True, False, False, False, False, False, False, False, False] def check_for_reset(): # Poll for external event return RESETS.pop(0) def announce(remaining): print(f'{remaining} ticks remaining') def run(): it = timer(4) while True: try: if check_for_reset(): current = it.throw(Reset()) else: current = next(it) except StopIteration: break else: announce(current) run() # Example 5 class Timer: def __init__(self, period): self.current = period self.period = period def reset(self): self.current = self.period def __iter__(self): while self.current: self.current -= 1 yield self.current # Example 6 RESETS = [ False, False, True, False, True, False, False, False, False, False, False, False, False] def run(): timer = Timer(4) for current in timer: if check_for_reset(): timer.reset() announce(current) run()