Python coroutines to write files

Question 1

I'm refactoring a monster 176-line function into something more sensible and more testable. The function as it stands fails the 'one thing and thing well' test by doing many things:

Indexing through a data structure
Compiling astr that represents state
Defining a file path to write that string into
Checking if the file path already exists and backing it up if so
Writing the string to filepath

I'm influenced by Gary Bernhardt's talk about FunctionalCore, ImperativeShell and recognise that the first 3 activities above could be functionalCore, and the last two having I/O have to be out of that core. I'm also exploring Python coroutines so of course I've digested David Beasley's talks.

I have mocked up activity_1, 2, 3 with the function below. Theroot_fpath is set to my home path for testing.

import randomdef write_files(writer=write_w()):'''mock source generatorfiles `file_{10..19}` have contents `content_{10..19}`; At random thatcontent is substituted by content at random selected from {10..19}; thissporadic random substitution mocks existing file withdifferent content (i.e. that the content of file_x has changed)'''    for i in xrange(10,20):        fpath = '{}/file_0{}'.format(root_fpath, i)        content = 'content_0{}'.format(i)        print (fpath, content)        if random.randint(0,1):            content = 'content_0{}'.format(random.randint(10,19))            print('random content:',content)            writer.send((fpath, content))                                 else:                                              writer.send((fpath, content))    writer.close()

This function uses thewriter.send() method to send filepath and content to the co-routine writer,write_w, defined and initialised by the functions below:

def coroutine(f):'''initializes f as coroutine by calling f.send(None)'''    def init(*args, **kwargs):        cor = f(*args, **kwargs)        cor.send(None)        return cor    return init@coroutinedef write_w():'''fpath content writer'''    while True:        fpath, content = (yield)        with open(fpath, 'w') as wFH:            wFH.write(content)

So far these do activities 1,2,3,5; the backup of any existing file is done using the functions below:

def isnew_fpathcontent(fpath,content):'''determines if fpath:content combination is new'''    if not os.path.isfile(fpath):        return True    with open(fpath) as rFH:        redlines = rFH.readlines()        if not redlines == [content]:                                                            print('existing content:', redlines)           return True        else:                                                                                            print('existing content:', redlines)            print('not new fpathcontent') #tmp#            return False@coroutinedef update(writer=None):'''writer wrapper forwarding new fpath:content combinations'''    while True:        fpath, content = (yield)        if isnew_fpathcontent(fpath, content):            print('new fpathcontent')   #tmp#            # need a BU call here            writer.send((fpath, content))

These functions can be called to do all activities 1..5 usingwrite_files(writer=update(writer=write_w())).

The outcome is -- despite the developmental bloat from a lot of print stmts -- way shorter than the original, has nicely separated the original concerns into individual functions, which will be (I haven't written tests yet; too early in this learning process for TDD) way more testable than the original.

And I have some questions:

Does this look sensible?
Is mywriter.close() call at the and of thewrite_files() mock source adequate? Have I missed something co-routine~y which will prove a liability? I'm deep in the learning process here.
Is there a recommended docstring practice for co-routine functions? Google has not yielded anything very useful. How to document, e.g., co-routinewrite_w() orupdate()? Their interfaces suggest no arguments; it gets these by theyield expression; is there an accepted docstring terminology to use? (params: None at call-time; receives (fpath,content) via yield?) And for thereturn from update()?`returns: (fpath,content); tuple, sent to writer co-routine?

Question 2

Please ensure that the code you posted is indented as intended — particularly the docstrings.

Question 3

Portability

I get syntax errors such as this:

    '''initializes f as coroutine by calling f.send(None)'''                                                           ^IndentationError: expected an indented block

Perhaps the version of Python you used is more forgiving. Regardless,the docstrings should be indented inside the functions. For example:

def coroutine(f):    '''initializes f as coroutine by calling f.send(None)'''

DRY

In thewrite_files function, this line is repeated twice:

writer.send((fpath, content))

It is the last statement in both branches of theif/else statement.In this case, there is no need for theelse:

if random.randint(0,1):    content = 'content_0{}'.format(random.randint(10,19))    print('random content:',content)writer.send((fpath, content))

There is a similar situation in theisnew_fpathcontent functionwhere this line is repeated as the first line in anif/else:

print('existing content:', redlines)

It can be moved before theif/else.

You can also simplify the logic by removing thenot andswapping the branches of theif/else:

print('existing content:', redlines)if redlines == [content]:                                                     return Falseelse:                                                                                    print('not new fpathcontent')    return True

Other improvements to these lines of code:

Used consistent indentation in both branches
Removed the vague#tmp# comment. Comments should be more descriptive.

toolic 16.4k6 gold badges29 silver badges220 bronze badges · Answer 1 · 2025-03-31 12:12:43Z

Portability

I get syntax errors such as this:

    '''initializes f as coroutine by calling f.send(None)'''                                                           ^IndentationError: expected an indented block

Perhaps the version of Python you used is more forgiving. Regardless,the docstrings should be indented inside the functions. For example:

def coroutine(f):    '''initializes f as coroutine by calling f.send(None)'''

DRY

In thewrite_files function, this line is repeated twice:

writer.send((fpath, content))

It is the last statement in both branches of theif/else statement.In this case, there is no need for theelse:

if random.randint(0,1):    content = 'content_0{}'.format(random.randint(10,19))    print('random content:',content)writer.send((fpath, content))

There is a similar situation in theisnew_fpathcontent functionwhere this line is repeated as the first line in anif/else:

print('existing content:', redlines)

It can be moved before theif/else.

You can also simplify the logic by removing thenot andswapping the branches of theif/else:

print('existing content:', redlines)if redlines == [content]:                                                     return Falseelse:                                                                                    print('not new fpathcontent')    return True

Other improvements to these lines of code:

Used consistent indentation in both branches
Removed the vague#tmp# comment. Comments should be more descriptive.

Movatterモバイル変換

Stack Exchange Network

Python coroutines to write files

1 Answer1

Portability

DRY

You mustlog in to answer this question.

Hot Network Questions

Subscribe to RSS