# Python wrapper for Windows pipes

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
self.handle = win32file.CreateFile(r"\.\pipe\%s" % name,
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
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__()
def isatty(self): #Return True if the stream is interactive (i.e., connected to a terminal/tty device).
return False
def seekable(self):
return False
def fileno(self):
return self.fd
def seek(self): # seek family
raise IOError("Not supported")
def tell(self): # Part of the seek family. Not supported
raise IOError("Not supported")
def write(self,data): # WriteFileEx impossible due to callback issues.
if not self.is_connected and self.pipetype == 'server':
self.connect()
if type(data).__name__ != 'bytes': # if we don't get bytes, make it bytes
data = bytes(data,'utf-8')
win32file.WriteFile(self.handle,data)
return len(data)
def close(self):
print("closure started")
win32pipe.DisconnectNamedPipe(self.handle)
if length == None:
length=self.inbuffersize
if resp[0] != 0:
raise __builtins__.BrokenPipeError(win32api.FormatMessage(resp[0]))
else:
return resp[1]


Usage looks like this:

>>> server = pipe("Pipename")
>>> client = pipe("Pipename","client")
>>> client.write("hello")
b'hello'
>>> server.write("words")
b'words'

• Your code is close to unreadable, start by making it PEP8 compliant. – Jaime May 3 '15 at 5:16
• @JaDogg The problem with that is that I am trying to extent the io.IOBase class, and it treats these as functions and not properties. Source: docs.python.org/3/library/io.html – computergorl May 3 '15 at 5:16
• I wish I could upvote @Jaime's comment more than once - you seem to have different indentation amounts at different levels and in different methods and the lines are way too long. – jonrsharpe May 3 '15 at 8:27
• @jonrsharpe The reason why there are different indentation amounts and such is because for some reason it gave me the indentation error so I said screw this after a while and just untabified -> tabified. It looks terrible but I'd rather that it works. – computergorl May 3 '15 at 22:41

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,
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__()

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)

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

• Why did you change some of the arguments for the functions? It doesn't work without those arguments being the way that they currently are. – computergorl May 3 '15 at 22:06
• @williambellwisdo I don't think I did, most of the changes were to do with layout rather than functionality. – jonrsharpe May 3 '15 at 22:07
• you changed the name of the pipe to r"\.pipe\%s" when it should have been r"\\.pipe\%s", as I had it. You also removed the openmode argument from the server CreateNamedPipe call. – computergorl May 4 '15 at 0:00
• @williambellwisdo 1. Please recheck your posted code... 2. You're right, that's a typo; as I stated, I couldn't test it. – jonrsharpe May 4 '15 at 7:04

## AutoPEP8

First and foremost run

autopep8 pipe_wrapper.py --aggressive --in-place


It will make the code extremely more readable adding spaces and blank lines.

## Remove debugging statements

Such as

print("exit started")


the user does not want to see the internals of your class printed.