A Python library that provides a pipe-based I/O system for efficient data transfer between threads, with support for data transformation filters and memory management.
- Unidirectional Pipes: Separate
PipeWriterandPipeReaderclasses for clear data flow - Pipe Operator: Chain components using the
|operator (like Unix pipes) - Memory Management: Built-in backpressure with configurable memory limits
- Data Filters: Transform data as it flows through the pipeline (e.g., hash computation)
- Thread-Safe: Designed for multi-threaded producer/consumer patterns
- S3 Integration: Optimized for streaming large files to/from S3 with integrity verification
pip install queuepipeiofromqueuepipeioimportPipeWriter, PipeReader# Create a pipewriter=PipeWriter()
reader=PipeReader()
writer|reader# Connect with pipe operator# In producer threadwriter.write(b"Hello, World!")
writer.close()
# In consumer threaddata=reader.read()
print(data) # b"Hello, World!"fromqueuepipeioimportPipeWriter, PipeReaderMB=1024*1024# Limit memory usage to 10MBwriter=PipeWriter(memory_limit=10*MB, chunk_size=1*MB)
reader=PipeReader()
writer|reader# Memory limit provides backpressure if consumer is slower# Writer will block when queue is fullfromqueuepipeioimportPipeWriter, PipeReader, HashingFilter# Create a pipeline: writer -> hasher -> readerwriter=PipeWriter()
hasher=HashingFilter(algorithm='sha256')
reader=PipeReader()
# Chain componentswriter|hasher|reader# Write datawriter.write(b"Important data")
writer.close()
# Read data (unchanged)data=reader.read()
# Get computed hashfile_hash=hasher.get_hash()
print(f"SHA256: {file_hash}")importboto3importthreadingfromqueuepipeioimportPipeWriter, PipeReader, HashingFilterdefstream_s3_download(s3_client, bucket, key, local_file):
"""Download from S3 with hash verification"""# Create pipelinewriter=PipeWriter(memory_limit=50*1024*1024) # 50MB limithasher=HashingFilter('sha256')
reader=PipeReader()
writer|hasher|readerdefdownload_thread():
"""Download from S3 to pipe"""response=s3_client.get_object(Bucket=bucket, Key=key)
forchunkinresponse['Body'].iter_chunks(chunk_size=1024*1024):
writer.write(chunk)
writer.close()
defsave_thread():
"""Read from pipe and save to file"""withopen(local_file, 'wb') asf:
whileTrue:
chunk=reader.read(1024*1024)
ifnotchunk:
breakf.write(chunk)
# Start threadsdl_thread=threading.Thread(target=download_thread)
save_thread=threading.Thread(target=save_thread)
dl_thread.start()
save_thread.start()
dl_thread.join()
save_thread.join()
# Return computed hash for verificationreturnhasher.get_hash()- PipeWriter: Write-only endpoint that puts data into a queue
- PipeReader: Read-only endpoint that reads data from a queue
- PipeFilter: Base class for data transformation filters
- HashingFilter: Computes hash of data passing through
- No Deadlocks: Unidirectional flow eliminates race conditions
- Composable: Easy to chain components and add filters
- Memory Efficient: Automatic backpressure prevents memory overflow
- High Performance: Minimal overhead, suitable for large file transfers
# Run all tests
python -m unittest discover -s tests
# Run with S3 integration tests (requires Docker)# S3 tests automatically start LocalStack if needed
python -m unittest tests.test_s3_integration -v
# Keep LocalStack running between test runs
KEEP_LOCALSTACK=true python -m unittest discover -s testsIf you're using the old QueueIO class:
# Old wayfromqueuepipeioimportQueueIOqio=QueueIO()
qio.write(data)
result=qio.read()
# New way fromqueuepipeioimportPipeWriter, PipeReaderwriter=PipeWriter()
reader=PipeReader()
writer|readerwriter.write(data)
writer.close()
result=reader.read()Contributions are welcome! Please feel free to submit a Pull Request.
This project is licensed under the MIT License - see the LICENSE file for details.