HiveBrain v1.2.0
Get Started
← Back to all entries
patternpythonMinor

Python wrapper for Windows pipes

Submitted by: @import:stackexchange-codereview··
0
Viewed 0 times
pipeswrapperforpythonwindows

Problem

This is done in the style of a single class, that should work like any other io-based object in Python. Help is much appreciated with anything regarding performance, documentation, or anything else.

```
import io,win32file,win32pipe, win32api
import msvcrt as ms # for fd magic
class pipe(io.IOBase):
def __init__(self, name, pipetype = 'server', openmode = win32pipe.PIPE_ACCESS_DUPLEX|win32file.FILE_FLAG_OVERLAPPED,
pipemode = win32pipe.PIPE_TYPE_BYTE|win32pipe.PIPE_NOWAIT,maxinstances=255,outbuffersize=1000000,inbuffersize=1000000,
defaulttimeout=50, securityattrib = None):
""" An implementation of a file-like python object pipe. Documentation can be found at https://msdn.microsoft.com/en-us/library/windows/desktop/aa365150(v=vs.85).aspx"""
self.pipetype = pipetype
self.name = name
self.openmode = openmode
self.pipemode = pipemode
self.__enter__ = self.connect
if pipetype == 'server':
self.handle = win32pipe.CreateNamedPipe(r"\.\pipe\%s" % name,
openmode, # default PIPE_ACCESS_DUPLEX|FILE_FLAG_OVERLAPPED
pipemode, # default PIPE_TYPE_BYTE|PIPE_NOWAIT
maxinstances, # default 255
outbuffersize, # default 1000000
inbuffersize, # default 1000000
defaulttimeout,# default 50
securityattrib)# default None
elif pipetype == 'client':
# it doesn't matter what type of pipe the server is so long as we know the name

Solution

Here I have fixed some of the more egregious violations of the style guide. I think the key point, more important than following any particular style, is consistency, which your code lacks.

I have also included some notes on implementation. Unfortunately I can't test it, as I'm using a Mac.

import io
import msvcrt as ms # for fd magic

import win32api, win32file, win32pipe

class pipe(io.IOBase):

    def __init__(self, name, pipetype='server',
                 openmode = (win32pipe.PIPE_ACCESS_DUPLEX |
                             win32file.FILE_FLAG_OVERLAPPED),
                 pipemode = win32pipe.PIPE_TYPE_BYTE | win32pipe.PIPE_NOWAIT,
                 maxinstances=255,outbuffersize=1000000,inbuffersize=1000000,
                 defaulttimeout=50, securityattrib=None):
        """ An implementation of a file-like Python object pipe.

        Documentation can be found at
        https://msdn.microsoft.com/en-us/library/windows/desktop/aa365150(v=vs.85).aspx

        """
        self.pipetype = pipetype
        self.name = name
        self.openmode = openmode
        self.pipemode = pipemode
        if pipetype == 'server':
            self.handle = win32pipe.CreateNamedPipe(
                r"\.\pipe\%s" % name,
                # the defaults don't need repeating here
                openmode,
                pipemode,
                maxinstances,
                outbuffersize,
                inbuffersize,
                defaulttimeout,
                securityattrib,
            )
        elif pipetype == 'client':
            # it doesn't matter what type of pipe the server is
            # so long as we know the name
            self.handle = win32file.CreateFile(
                r"\.\pipe\%s" % name,
                win32file.GENERIC_READ | win32file.GENERIC_WRITE,
                0,
                None,
                win32file.OPEN_EXISTING,
                0,
                None,
            )
        self.fd = ms.open_osfhandle(self.handle, 0)
        self.is_connected = False
        (self.flags, self.outbuffersize, self.inbuffersize, self.maxinstances =
         win32pipe.GetNamedPipeInfo(self.handle))

    def connect(self): # TODO: WaitNamedPipe ?
        win32pipe.ConnectNamedPipe(self.handle,None)
        self.is_connected = True

    # do this at class level rather than in __init__
    __enter__ = connect  

    def __del__(self):
        print("del initiated")
        try:
            self.write(b'') # try to clear up anyone waiting
        except win32pipe.error: # no one's listening
            pass
        self.close()

    def __exit__(self):
        print("exit started")
        self.__del__()

    # Use docstrings, not comments
    def isatty(self):
        """Is the stream interactive (connected to a terminal/tty)?"""
        return False

    def seekable(self):
        return False

    def fileno(self):
        return self.fd

    def seek(self):
        # I think this is clearer than an IOError
        raise NotImplementedError

    def tell(self):
        # as above
        raise NotImplementedError

    def write(self, data):
        """WriteFileEx impossible due to callback issues."""
        if not self.is_connected and self.pipetype == 'server':
            self.connect()
        # there is no need to compare the __name__ of the type!
        if not isinstance(data, bytes):
            data = bytes(data, 'utf-8')
        win32file.WriteFile(self.handle, data)
        return len(data)

    def close(self):
        print("closure started")
        win32pipe.DisconnectNamedPipe(self.handle)

    def read(self, length=None):
        # Always compare None by identity, not equality
        if length is None:
            length = self.inbuffersize
        resp = win32file.ReadFile(self.handle, length)
        if resp[0] != 0:
            raise __builtins__.BrokenPipeError(win32api.FormatMessage(resp[0]))
        else:
            return resp[1]

Code Snippets

import io
import msvcrt as ms # for fd magic

import win32api, win32file, win32pipe


class pipe(io.IOBase):

    def __init__(self, name, pipetype='server',
                 openmode = (win32pipe.PIPE_ACCESS_DUPLEX |
                             win32file.FILE_FLAG_OVERLAPPED),
                 pipemode = win32pipe.PIPE_TYPE_BYTE | win32pipe.PIPE_NOWAIT,
                 maxinstances=255,outbuffersize=1000000,inbuffersize=1000000,
                 defaulttimeout=50, securityattrib=None):
        """ An implementation of a file-like Python object pipe.

        Documentation can be found at
        https://msdn.microsoft.com/en-us/library/windows/desktop/aa365150(v=vs.85).aspx

        """
        self.pipetype = pipetype
        self.name = name
        self.openmode = openmode
        self.pipemode = pipemode
        if pipetype == 'server':
            self.handle = win32pipe.CreateNamedPipe(
                r"\.\pipe\%s" % name,
                # the defaults don't need repeating here
                openmode,
                pipemode,
                maxinstances,
                outbuffersize,
                inbuffersize,
                defaulttimeout,
                securityattrib,
            )
        elif pipetype == 'client':
            # it doesn't matter what type of pipe the server is
            # so long as we know the name
            self.handle = win32file.CreateFile(
                r"\.\pipe\%s" % name,
                win32file.GENERIC_READ | win32file.GENERIC_WRITE,
                0,
                None,
                win32file.OPEN_EXISTING,
                0,
                None,
            )
        self.fd = ms.open_osfhandle(self.handle, 0)
        self.is_connected = False
        (self.flags, self.outbuffersize, self.inbuffersize, self.maxinstances =
         win32pipe.GetNamedPipeInfo(self.handle))

    def connect(self): # TODO: WaitNamedPipe ?
        win32pipe.ConnectNamedPipe(self.handle,None)
        self.is_connected = True

    # do this at class level rather than in __init__
    __enter__ = connect  

    def __del__(self):
        print("del initiated")
        try:
            self.write(b'') # try to clear up anyone waiting
        except win32pipe.error: # no one's listening
            pass
        self.close()

    def __exit__(self):
        print("exit started")
        self.__del__()

    # Use docstrings, not comments
    def isatty(self):
        """Is the stream interactive (connected to a terminal/tty)?"""
        return False

    def seekable(self):
        return False

    def fileno(self):
        return self.fd

    def seek(self):
        # I think this is clearer than an IOError
        raise NotImplementedError

    def tell(self):
        # as above
        raise NotImplementedError

    def write(self, data):
        """WriteFileEx impossible due to callback issues."""
        if not self.is_connected and self.pipetype == 'server':
            self.connect()
        # there 

Context

StackExchange Code Review Q#88672, answer score: 3

Revisions (0)

No revisions yet.