This tutorial is from open-source community. Access the source code
In this lab, you will learn how to manage what happens at the yield statements in Python. You'll gain an understanding of how to handle the operations and behaviors associated with these statements effectively.
Additionally, you will learn about generator lifetime and exception handling in generators. The files follow.py and cofollow.py will be modified during this learning process.
In this step, we're going to explore the lifetime of Python generators and learn how to close them properly. Generators in Python are a special type of iterator that allow you to generate a sequence of values on-the-fly, rather than computing them all at once and storing them in memory. This can be very useful when dealing with large datasets or infinite sequences.
follow() Generator?Let's start by looking at the follow.py file in the project directory. This file contains a generator function named follow(). A generator function is defined like a normal function, but instead of using the return keyword, it uses yield. When a generator function is called, it returns a generator object, which you can iterate over to get the values it yields.
The follow() generator function continuously reads lines from a file and yields each line as it is read. This is similar to the Unix tail -f command, which continuously monitors a file for new lines.
Open the follow.py file in the WebIDE editor:
import os
import time
def follow(filename):
with open(filename,'r') as f:
f.seek(0,os.SEEK_END)
while True:
line = f.readline()
if line == '':
time.sleep(0.1) ## Sleep briefly to avoid busy wait
continue
yield lineIn this code, the with open(filename, 'r') as f statement opens the file in read mode and ensures that it is properly closed when the block is exited. The f.seek(0, os.SEEK_END) line moves the file pointer to the end of the file, so that the generator starts reading from the end. The while True loop continuously reads lines from the file. If the line is empty, it means there are no new lines yet, so the program sleeps for 0.1 seconds to avoid a busy wait and then continues to the next iteration. If the line is not empty, it is yielded.
This generator runs in an infinite loop, which raises an important question: what happens when we stop using the generator or want to terminate it early?
We need to modify the follow() function in follow.py to handle the case when the generator is closed properly. To do this, we'll add a try-except block that catches the GeneratorExit exception. The GeneratorExit exception is raised when a generator is closed, either by garbage collection or by calling the close() method.
import os
import time
def follow(filename):
try:
with open(filename,'r') as f:
f.seek(0,os.SEEK_END)
while True:
line = f.readline()
if line == '':
time.sleep(0.1) ## Sleep briefly to avoid busy wait
continue
yield line
except GeneratorExit:
print('Following Done')In this modified code, the try block contains the main logic of the generator. If a GeneratorExit exception is raised, the except block catches it and prints the message 'Following Done'. This is a simple way to perform cleanup actions when the generator is closed.
Save the file after making these changes.
Now, let's conduct some experiments to see how generators behave when they are garbage collected or explicitly closed.
Open a terminal and run the Python interpreter:
cd ~/project
python3>>> from follow import follow
>>> ## Experiment: Garbage collection of a running generator
>>> f = follow('stocklog.csv')
>>> next(f)
'"MO",70.29,"6/11/2007","09:30.09",-0.01,70.25,70.30,70.29,365314\n'
>>> del f ## Delete the generator object
Following Done ## This message appears because of our GeneratorExit handlerIn this experiment, we first import the follow function from the follow.py file. Then we create a generator object f by calling follow('stocklog.csv'). We use the next() function to get the next line from the generator. Finally, we delete the generator object using the del statement. When the generator object is deleted, it is automatically closed, which triggers our GeneratorExit exception handler, and the message 'Following Done' is printed.
>>> f = follow('stocklog.csv')
>>> for line in f:
... print(line, end='')
... if 'IBM' in line:
... f.close() ## Explicitly close the generator
...
"MO",70.29,"6/11/2007","09:30.09",-0.01,70.25,70.30,70.29,365314
"VZ",42.91,"6/11/2007","09:34.28",-0.16,42.95,42.91,42.78,210151
"HPQ",45.76,"6/11/2007","09:34.29",0.06,45.80,45.76,45.59,257169
"GM",31.45,"6/11/2007","09:34.31",0.45,31.00,31.50,31.45,582429
"IBM",102.86,"6/11/2007","09:34.44",-0.21,102.87,102.86,102.77,147550
Following Done
>>> for line in f:
... print(line, end='') ## No output: generator is closed
...In this experiment, we create a new generator object f and iterate over it using a for loop. Inside the loop, we print each line and check if the line contains the string 'IBM'. If it does, we call the close() method on the generator to explicitly close it. When the generator is closed, the GeneratorExit exception is raised, and our exception handler prints the message 'Following Done'. After the generator is closed, if we try to iterate over it again, there will be no output because the generator is no longer active.
>>> f = follow('stocklog.csv')
>>> for line in f:
... print(line, end='')
... if 'IBM' in line:
... break ## Break out of the loop, but don't close the generator
...
"MO",70.29,"6/11/2007","09:30.09",-0.01,70.25,70.30,70.29,365314
"VZ",42.91,"6/11/2007","09:34.28",-0.16,42.95,42.91,42.78,210151
"HPQ",45.76,"6/11/2007","09:34.29",0.06,45.80,45.76,45.59,257169
"GM",31.45,"6/11/2007","09:34.31",0.45,31.00,31.50,31.45,582429
"IBM",102.86,"6/11/2007","09:34.44",-0.21,102.87,102.86,102.77,147550
>>> ## Resume iteration - the generator is still active
>>> for line in f:
... print(line, end='')
... if 'IBM' in line:
... break
...
"CAT",78.36,"6/11/2007","09:37.19",-0.16,78.32,78.36,77.99,237714
"VZ",42.99,"6/11/2007","09:37.20",-0.08,42.95,42.99,42.78,268459
"IBM",102.91,"6/11/2007","09:37.31",-0.16,102.87,102.91,102.77,190859
>>> del f ## Clean up
Following DoneIn this experiment, we create a generator object f and iterate over it using a for loop. Inside the loop, we print each line and check if the line contains the string 'IBM'. If it does, we use the break statement to break out of the loop. Breaking out of the loop does not close the generator, so the generator is still active. We can then resume the iteration by starting a new for loop over the same generator object. Finally, we delete the generator object to clean up, which triggers the GeneratorExit exception handler.
close()), a GeneratorExit exception is raised inside the generator.break) does not close the generator, allowing it to be resumed later.Exit the Python interpreter by typing exit() or pressing Ctrl+D.
In this step, we're going to learn how to handle exceptions in generators and coroutines. But first, let's understand what exceptions are. An exception is an event that occurs during the execution of a program and disrupts the normal flow of the program's instructions. In Python, we can use the throw() method to handle exceptions in generators and coroutines.
A coroutine is a special type of generator. Unlike regular generators that mainly yield values, coroutines can both consume values (using the send() method) and yield values. The cofollow.py file has a simple implementation of a coroutine.
Let's open the cofollow.py file in the WebIDE editor. Here's the code inside:
def consumer(func):
def start(*args,**kwargs):
c = func(*args,**kwargs)
next(c)
return c
return start
@consumer
def printer():
while True:
item = yield
print(item)Now, let's break down this code. The consumer is a decorator. A decorator is a function that takes another function as an argument, adds some functionality to it, and then returns the modified function. In this case, the consumer decorator automatically moves the generator to its first yield statement. This is important because it makes the generator ready to receive values.
The printer() coroutine is defined with the @consumer decorator. Inside the printer() function, we have an infinite while loop. The item = yield statement is where the magic happens. It pauses the execution of the coroutine and waits to receive a value. When a value is sent to the coroutine, it resumes execution and prints the received value.
Now, we're going to modify the printer() coroutine to handle exceptions. We'll update the printer() function in cofollow.py like this:
@consumer
def printer():
while True:
try:
item = yield
print(item)
except Exception as e:
print('ERROR: %r' % e)The try block contains the code that might raise an exception. In our case, it's the code that receives and prints the value. If an exception occurs in the try block, the execution jumps to the except block. The except block catches the exception and prints an error message. After making these changes, save the file.
Let's start experimenting with throwing exceptions into the coroutine. Open a terminal and run the Python interpreter using the following commands:
cd ~/project
python3>>> from cofollow import printer
>>> p = printer()
>>> p.send('hello') ## Send a value to the coroutine
hello
>>> p.send(42) ## Send another value
42Here, we first import the printer coroutine from the cofollow module. Then we create an instance of the printer coroutine named p. We use the send() method to send values to the coroutine. As you can see, the coroutine processes the values we send to it without any problems.
>>> p.throw(ValueError('It failed')) ## Throw an exception into the coroutine
ERROR: ValueError('It failed')In this experiment, we use the throw() method to inject a ValueError exception into the coroutine. The try-except block in the printer() coroutine catches the exception and prints an error message. This shows that our exception handling is working as expected.
>>> try:
... int('n/a') ## This will raise a ValueError
... except ValueError as e:
... p.throw(e) ## Throw the caught exception into the coroutine
...
ERROR: ValueError("invalid literal for int() with base 10: 'n/a'")Here, we first try to convert the string 'n/a' to an integer, which raises a ValueError. We catch this exception and then use the throw() method to pass it to the coroutine. The coroutine catches the exception and prints the error message.
>>> p.send('still working') ## The coroutine continues to run after handling exceptions
still workingAfter handling the exceptions, we send another value to the coroutine using the send() method. The coroutine is still active and can process the new value. This shows that our coroutine can continue running even after encountering errors.
yield statement. This means that we can catch and handle errors that occur when the coroutine is waiting for or processing a value.throw() method allows you to inject exceptions into a generator or coroutine. This is useful for testing and for handling errors that occur outside the coroutine.To exit the Python interpreter, you can type exit() or press Ctrl+D.
In this step, we're going to explore how to apply the concepts we've learned about managing generators and handling exceptions in generators to real - world scenarios. Understanding these practical applications will help you write more robust and efficient Python code.
Let's build a more reliable version of our file monitoring system. This system will be able to handle different situations, such as timeouts and user requests to stop.
First, open the WebIDE editor and create a new file named robust_follow.py. Here's the code you need to write in this file:
import os
import time
import signal
class TimeoutError(Exception):
pass
def timeout_handler(signum, frame):
raise TimeoutError("Operation timed out")
def follow(filename, timeout=None):
"""
A generator that yields new lines in a file.
With timeout handling and proper cleanup.
"""
try:
## Set up timeout if specified
if timeout:
signal.signal(signal.SIGALRM, timeout_handler)
signal.alarm(timeout)
with open(filename, 'r') as f:
f.seek(0, os.SEEK_END)
while True:
line = f.readline()
if line == '':
## No new data, wait briefly
time.sleep(0.1)
continue
yield line
except TimeoutError:
print(f"Following timed out after {timeout} seconds")
except GeneratorExit:
print("Following stopped by request")
finally:
## Clean up timeout alarm if it was set
if timeout:
signal.alarm(0)
print("Follow generator cleanup complete")In this code, we first define a custom TimeoutError class. The timeout_handler function is used to raise this error when a timeout occurs. The follow function is a generator that reads a file and yields new lines. If a timeout is specified, it sets up an alarm using the signal module. If there's no new data in the file, it waits for a short time before trying again. The try - except - finally block is used to handle different exceptions and ensure proper cleanup.
After writing the code, save the file.
Now, let's test our improved file monitoring system. Open a terminal and run the Python interpreter with the following commands:
cd ~/project
python3In the Python interpreter, we'll test the basic functionality of our follow generator. Here's the code to run:
>>> from robust_follow import follow
>>> f = follow('stocklog.csv')
>>> for i, line in enumerate(f):
... print(f"Line {i+1}: {line.strip()}")
... if i >= 2: ## Just read a few lines for the example
... break
...
Line 1: "MO",70.29,"6/11/2007","09:30.09",-0.01,70.25,70.30,70.29,365314
Line 2: "VZ",42.91,"6/11/2007","09:34.28",-0.16,42.95,42.91,42.78,210151
Line 3: "HPQ",45.76,"6/11/2007","09:34.29",0.06,45.80,45.76,45.59,257169Here, we import the follow function from our robust_follow.py file. Then we create a generator object f that follows the stocklog.csv file. We use a for loop to iterate over the lines yielded by the generator and print the first three lines.
Let's see how the timeout feature works. Run the following code in the Python interpreter:
>>> ## Create a generator that will time out after 3 seconds
>>> f = follow('stocklog.csv', timeout=3)
>>> for line in f:
... print(line.strip())
... time.sleep(1) ## Process each line slowly
...
"MO",70.29,"6/11/2007","09:30.09",-0.01,70.25,70.30,70.29,365314
"VZ",42.91,"6/11/2007","09:34.28",-0.16,42.95,42.91,42.78,210151
"HPQ",45.76,"6/11/2007","09:34.29",0.06,45.80,45.76,45.59,257169
Following timed out after 3 seconds
Follow generator cleanup completeIn this experiment, we create a generator with a 3 - second timeout. We process each line slowly by sleeping for 1 second between each line. After about 3 seconds, the generator raises a timeout exception, and the cleanup code in the finally block is executed.
Let's test how the generator handles an explicit closure. Run the following code:
>>> f = follow('stocklog.csv')
>>> for i, line in enumerate(f):
... print(f"Line {i+1}: {line.strip()}")
... if i >= 1:
... print("Explicitly closing the generator...")
... f.close()
...
Line 1: "MO",70.29,"6/11/2007","09:30.09",-0.01,70.25,70.30,70.29,365314
Line 2: "VZ",42.91,"6/11/2007","09:34.28",-0.16,42.95,42.91,42.78,210151
Explicitly closing the generator...
Following stopped by request
Follow generator cleanup completeHere, we create a generator and start iterating over its lines. After processing two lines, we explicitly close the generator using the close method. The generator then handles the GeneratorExit exception and performs the necessary cleanup.
Next, we'll create a simple data processing pipeline using coroutines. This pipeline will be able to handle errors at different stages.
Open the WebIDE editor and create a new file named pipeline.py. Here's the code to write in this file:
def consumer(func):
def start(*args,**kwargs):
c = func(*args,**kwargs)
next(c)
return c
return start
@consumer
def grep(pattern, target):
"""Filter lines containing pattern and send to target"""
try:
while True:
line = yield
if pattern in line:
target.send(line)
except Exception as e:
target.throw(e)
@consumer
def printer():
"""Print received items"""
try:
while True:
item = yield
print(f"PRINTER: {item}")
except Exception as e:
print(f"PRINTER ERROR: {repr(e)}")
def follow_and_process(filename, pattern):
"""Follow a file and process its contents"""
import time
import os
output = printer()
filter_pipe = grep(pattern, output)
try:
with open(filename, 'r') as f:
f.seek(0, os.SEEK_END)
while True:
line = f.readline()
if not line:
time.sleep(0.1)
continue
filter_pipe.send(line)
except KeyboardInterrupt:
print("Processing stopped by user")
finally:
filter_pipe.close()
output.close()In this code, the consumer decorator is used to initialize coroutines. The grep coroutine filters lines that contain a specific pattern and sends them to another coroutine. The printer coroutine prints the received items. The follow_and_process function reads a file, filters its lines using the grep coroutine, and prints the matching lines using the printer coroutine. It also handles the KeyboardInterrupt exception and ensures proper cleanup.
After writing the code, save the file.
Let's test our data processing pipeline. In a terminal, run the following command:
cd ~/project
python3 -c "from pipeline import follow_and_process; follow_and_process('stocklog.csv', 'IBM')"You should see output similar to this:
PRINTER: "IBM",102.86,"6/11/2007","09:34.44",-0.21,102.87,102.86,102.77,147550
PRINTER: "IBM",102.91,"6/11/2007","09:37.31",-0.16,102.87,102.91,102.77,190859
PRINTER: "IBM",102.95,"6/11/2007","09:39.44",-0.12,102.87,102.95,102.77,225350This output shows that the pipeline is working correctly, filtering and printing lines that contain the "IBM" pattern.
To stop the process, press Ctrl+C. You should see the following message:
Processing stopped by userfinally block in generators ensures cleanup operations are performed, regardless of how the generator terminates. This helps maintain the integrity of your program and prevents resource leaks.In this lab, you have learned essential techniques for managing yield statements in Python generators and coroutines. You've explored generator lifetime management, including handling the GeneratorExit exception during closure or garbage collection and controlling iteration break and resume. Additionally, you've learned about exception handling in generators, such as using the throw() method and writing robust generators to handle exceptions gracefully.
These techniques are fundamental for building robust, maintainable Python applications. They are useful for data processing, asynchronous operations, and resource management. By properly managing generator lifetime and handling exceptions, you can create resilient systems that gracefully handle errors and clean up resources when they are no longer needed.
