Python Generators and Friends

22. October 2014

Ole Martin Bjørndalen (ole.martin.bjorndalen@uit.no)

(Press "a" for slide view.)

Generators and Friends

This talk will cover:

Generators

for line in file:

for node in graph:

for message in port:

A Simple Generator

def numbers():
    yield 1
    yield 2
    yield 3

for n in numbers():
    print(n)
1
2
3

A More Typical Generator

def numbers():
    for i in [1, 2, 3]:
        yield i + 1
    yield 5
>>> list(numbers())
[2, 3, 4, 5]
>>> set(numbers())
set([2, 3, 4, 5])

Example: Important Lines

def important_lines(lines):
    for line in lines:
        line = line.split('#')[0].strip()
        if line != '':
            yield line

for line in important_lines(open('/etc/fstab')):
    print(line)
/dev/sdb1  /mnt/data        ext4  auto  0 0
/dev/sdc1  /mnt/virtualbox  ext4  auto  0 0
...

Example: Blocking Receive

while True:
    message = port.receive()
    print(message)

Rewritten: Blocking Receive

class MidiPort(object):
    def __iter__(self):
        while True:
            message = self.receive()
            yield message
for message in port:
    print(message)

Example: Non-Blocking Receive

while True:
    while True:
        message = port.receive(block=False)
        if message:
            print(message)
        else:
            break
    # ... do something else

Rewritten: Non-Blocking Receive

class MidiPort(object):
    def iter_pending(self):
        while True:
            message = self.receive(block=False)
            if message:
                yield message
            else:
                break
while True:
    for message in port.iter_pending():
        print(message)
    # ... do something else

Example: Read FLAC Tags

FLAC metadata (Vorbis comments):

$ metaflac --export-tags-to=- test.flac
artist=Some Artist
album=Some Album
year=2014
from subprocess import Popen, PIPE

def read_flac_tags(filename):
    tags = {}
    args = ['metaflac', '--export-tags-to=-', filename]
    process = Popen(args, stdout=PIPE)
    for line in process.stdout:
        key, value = line.decode('utf-8').strip().split('=', 1)
        tags[key] = value
    return tags

Rewritten: Read FLAC Tags (Part 1)

from subprocess import Popen, PIPE

def inpipe(args, encoding='utf-8'):
    process = Popen(args, stdout=PIPE)
    for line in process.stdout:
        yield line.decode(encoding).strip()
>>> for line in inpipe(['metaflac', '--export-tags-to=-', 'test.flac']):
...     print(line)
artist=Some Artist
album=Some Album
year=2014

Rewritten: Read FLAC Tags (Part 2)

def read_flac_tags(filename):
    args = ['metaflac', '--export-tags-to=-', filename]
    return dict(line.split('=', 1) for line in inpipe(args))

>>> tags = read_flac_tags('test.flac')
>>> print(tags)
{'album': 'Some Album',
 'artist': 'Some Artist',
 'year': '2014'}

Comprehensions

>>> [n for n in range(10) if n % 2 == 0]
[0, 2, 4, 6, 8]

Three Types of Comprehensions

List comprehension:

[math.sqrt(n) for n in numbers()]

Generator comprehension:

(math.sqrt(n) for n in numbers())

Dictionary comprehension:

{n: math.sqrt(n) for n in numbers()}

Example: List Comprehension

Typical tedious code:

messages = []
for message in parser:
    if message.type == 'sysex':
        messages.append(message)
return messages

With list comprehension:

return [message for message in parser if message.type == 'sysex']

A bit SQL-like.

Example: Generator Comprehension

Without generator expression:

seconds = 0
for message in messages:
    seconds += message.time
return seconds

With generator expression:

return sum(message.time for message in messages)

Example: Dictionary Comprehension

>>> import dbfread
>>> table = dbfread.DBF('kabreg.dbf')
>>> cables = {record['CABLE']: record for record in table}
>>> cables.keys()
['AAB-1-019', 'AAB-1-018', 'AAB-1-017', ...]
>>> cables
{'AAB-1-019': OrderedDict([('CABLE', 'AAB-1-019'),
                           ('OWNER', 'UIT'),
                           ...]),
{'AAB-1-022': OrderedDict([('CABLE', 'AAB-1-022'),
                           ('OWNER', 'UIT'),
                           ...]),
 ...}

Context Managers

with lock:
    do_dangerous_stuff()

Example: chdir()

Change directory temporarily:

import os

olddir = os.getcwd()
os.chdir('/tmp/')
try:
    # ... do stuff
finally:
    os.chdir(olddir)

Rewritten: chdir()

from contextlib import contextmanager

@contextmanager
def chdir(dirname):
    olddir = os.getcwd()
    os.chdir(dirname)
    yield
    os.chdir(olddir)
with chdir('/tmp/'):
    # ... do stuff

Example: Database Transaction

Automatically roll back transaction on exception:

try:
    cursor.execute('INSERT something')
    cursor.execute('this will fail')
except:
    conn.rollback()
    raise
finally:
    conn.commit()

Rewritten: Database Transaction

from contextlib import contextmanager

@contextmanager
def transaction(conn):
    try:
        yield
    except:
        conn.rollback()
        raise
    finally:
        conn.commit()
with transaction(conn):
    cursor.execute('INSERT something')
    cursor.execute('this will fail')

Example: Closing a Port

port = MidiPort('MIDI-1')
try:
    port.send(message)
finally:
    port.close()

Rewritten: Closing a Port

class MidiPort(object):
    # ...
    def __enter__(self):
        return self

    def __exit__(self, type, value, traceback):
        self.close()
        return False
with MidiPort('MIDI-1') as port:
    port.send(message)

Properties

class User(object):
    @property
    def display_name(self):
        return self.name or 'Anonymous'
>>> print(user.display_name)
Anonymous

Getters

class NmapScanner(object):
    def __init__(self, ip):
        self.ip = ip
        self.process = Popen(['nmap', 'ip'], ...)

    @property
    def done(self):
        return self.process.returnvalue is not None
if scanner.done:
    insert_data(scanner.data)

Setters

Not used a lot, but sometimes very useful.

class SynthPatch(object):
    @property
    def name(self):
        return decode_name(self['common'][:12])

    @name.setter
    def name(self, name):
        self['common'][:12] = encode_name(name)
>>> patch.name = 'Dark Strings'
>>> print(track.name)
Dark Strings

Decorators

A decorator:

But using them is easy:

@cache
def is_prime():
    ...

@property
def done(self):
    return self.process.returnvalue is not None

How Decorators Work

@property
def done(self):
    return self.process.returnvalue is not None

is the same as:

def done(self):
    return self.process.returnvalue is not None

done = property(done)

Example: Return JSON Data

Return JSON data from a Django view:

def get_some_data(request):
    data = {'Some': 'data'}
    return HttpResponse(json.dumps(data, indent=2),
                        mimetype="application/json")

Rewritten: Return JSON Data

import json

def as_json(func):
    def wrapper(request, *args, **kwargs):
        data = func(request, *args, **kwargs)
        return HttpResponse(json.dumps(data, indent=2),
                            mimetype="application/json")
    return wrapper
@as_json
def get_some_data(request):
    return {'Some': 'data'}

Example: Template

TEMPLATES = {'page': '<h1>{title}</h1> <p>{body}</p>'}

def test_page(request):
    variables = {'title': 'Hello', 'body': 'This is a test!'}
    return HttpResponse(TEMPLATES['page'].format(**variables))

Rewritten: Template

TEMPLATES = {'page': '<h1>{title}</h1> <p>{body}</p>'}

def with_template(template_name):
    def decorator(func):
        def wrapper(*args, **kwargs):
            template = TEMPLATES[template_name]
            variables = func(*args, **kwargs)
            return HttpResponse(template.format(**variables))
        return wrapper
    return decorator
@with_template('page')
def test_page(request):
    return {'title': 'Hello', 'body': 'This is a test!'}

Python Generators and Friends

See also:

Examples from: