Overview

Features

Installation

pip install memfs

git clone https://github.com/pyfunc/memfs.gitcd memfs

# Create a virtual environmentpython3 -m venv .venvsource .venv/bin/activate# On Linux/macOS# .venv\Scripts\activate  # On Windows# Install in development modepip install --upgrade pippip install setuptools wheelpip install -e.# Build the packagepython -m build

Basic Usage Examples

Basic File Operations

frommemfsimportcreate_fs# Create a file system instancefs=create_fs()# Write to a filefs.writefile('/hello.txt','Hello, world!')# Read from a filecontent=fs.readfile('/hello.txt')print(content)# Outputs: Hello, world!# Check if a file existsiffs.exists('/hello.txt'):print('File exists!')# Create directoriesfs.makedirs('/path/to/directory')# List directory contentsfiles=fs.listdir('/path/to')

Using File-Like Objects

frommemfsimportcreate_fsfs=create_fs()# Write using a file-like objectwithfs.open('/data.txt','w')asf:f.write('Line 1\n')f.write('Line 2\n')# Read using a file-like objectwithfs.open('/data.txt','r')asf:forlineinf:print(line.strip())

Directory Operations

frommemfsimportcreate_fsfs=create_fs()# Create nested directoriesfs.makedirs('/a/b/c')# Walk the directory treeforroot,dirs,filesinfs.walk('/'):print(f"Directory:{root}")print(f"Subdirectories:{dirs}")print(f"Files:{files}")

Advanced Usage Examples

Data Processing Pipeline

frommemfsimportcreate_fsimportjsonimportcsvfs=create_fs()# Create directoriesfs.makedirs('/data/raw',exist_ok=True)fs.makedirs('/data/processed',exist_ok=True)# Write CSV datawithfs.open('/data/raw/input.csv','w',newline='')asf:writer=csv.writer(f)writer.writerows([        ['id','name','value'],        [1,'Alpha',100],        [2,'Beta',200]    ])# Process CSV to JSONwithfs.open('/data/raw/input.csv','r',newline='')asf:reader=csv.DictReader(f)data= [rowforrowinreader]# Transform and save the dataforitemindata:item['value']=int(item['value'])item['double_value']=item['value']*2withfs.open('/data/processed/output.json','w')asf:json.dump(data,f,indent=2)

Parallel Processing

frommemfsimportcreate_fsimportjsonimportconcurrent.futuresfs=create_fs()fs.makedirs('/parallel/input',exist_ok=True)fs.makedirs('/parallel/output',exist_ok=True)# Create input filesforiinrange(10):fs.writefile(f'/parallel/input/file_{i}.json',json.dumps({'id':i}))defprocess_file(filename):withfs.open(f'/parallel/input/{filename}','r')asf:data=json.loads(f.read())# Process datadata['processed']=Truewithfs.open(f'/parallel/output/processed_{filename}','w')asf:f.write(json.dumps(data,indent=2))returndata['id']# Process files in parallelwithconcurrent.futures.ThreadPoolExecutor(max_workers=4)asexecutor:futures= {executor.submit(process_file,f):fforfinfs.listdir('/parallel/input')}forfutureinconcurrent.futures.as_completed(futures):file_id=future.result()print(f"Processed file ID:{file_id}")

Encrypted and Compressed Files

frommemfs.examples.custom_filesystemimportcreate_extended_fs# Create extended filesystem with encryption and compressionfs=create_extended_fs()# Write to an encrypted filewithfs.open_encrypted('/secret.txt','w',password='mysecret')asf:f.write('This is sensitive information')# Read from an encrypted filewithfs.open_encrypted('/secret.txt','r',password='mysecret')asf:content=f.read()print(content)# Write to a compressed file (good for large text)withfs.open_compressed('/compressed.txt','w',compression_level=9)asf:f.write('This content will be compressed '*1000)# Check the file sizesnormal_size=len(fs.readfile('/secret.txt'))compressed_size=len(fs._FS_DATA['files']['/compressed.txt'])print(f"Compression ratio:{normal_size/compressed_size:.2f}x")

gRPC Service Pipeline

frommemfsimportcreate_fsfrommemfs.apiimportDynamicgRPCComponent,PipelineOrchestrator# Define transformation functionsdeftransform_data(data):ifisinstance(data,dict):data['transformed']=Truereturndatadefformat_data(data):ifisinstance(data,dict):data['formatted']=Truereturndata# Create virtual directoriesfs=create_fs()fs.makedirs('/proto/transform',exist_ok=True)fs.makedirs('/proto/format',exist_ok=True)fs.makedirs('/generated/transform',exist_ok=True)fs.makedirs('/generated/format',exist_ok=True)# Create componentstransform_component=DynamicgRPCComponent(transform_data,proto_dir="/proto/transform",generated_dir="/generated/transform",port=50051)format_component=DynamicgRPCComponent(format_data,proto_dir="/proto/format",generated_dir="/generated/format",port=50052)# Create and execute pipelinepipeline=PipelineOrchestrator()pipeline.add_component(transform_component)pipeline.add_component(format_component)result=pipeline.execute_pipeline({"input":"data"})print(result)# {"input": "data", "transformed": true, "formatted": true}

Command-line Interface

Basic CLI Usage

# Initialize a new filesystem (clears any existing state)memfs init# Display filesystem as a treememfs tree /# Create a directory with parentsmemfs mkdir -p /data/subdir# Create an empty filememfs touch /data/hello.txt# Write content to a filememfs write /data/hello.txt"Hello, virtual world!"# Read file contentmemfsread /data/hello.txt# Dump filesystem content as JSONmemfs dump

Interactive Shell Mode

memfs shell

memfs> mkdir -p /datamemfs> touch /data/hello.txtmemfs> write /data/hello.txt "Hello from shell mode!"memfs> tree /memfs> exit

CLI State Management

mkdir -p /datatouch /data/hello.txtwrite /data/hello.txt "Hello from shell mode!"tree /exit

CLI State Management

# Check if state file existsls -la~/.memfs_state.json# Reset state by initializing a new filesystemmemfs init# Or manually create an empty state fileecho'{"files": {}, "dirs": ["/"]}'>~/.memfs_state.json

Creating a Custom CLI Command

#!/usr/bin/env pythonfrommemfsimportcreate_fsfs=create_fs()fs.makedirs('/data',exist_ok=True)fs.writefile('/data/hello.txt','Hello, world!')print("Filesystem contents:")forroot,dirs,filesinfs.walk('/'):print(f"Directory:{root}")fordindirs:print(f"  Dir:{d}")forfinfiles:print(f"  File:{f}")

Project Structure

memfs/├── setup.py          # Package installation configuration├── setup.cfg         # Setup configuration├── README.md         # Project documentation├── src/              # Source code│   └── memfs/        # Main package│       ├── __init__.py     # Basic component imports│       ├── _version.py     # Version information│       ├── memfs.py        # Virtual filesystem implementation│       ├── api.py          # gRPC service generation module│       └── cli.py          # Command-line interface├── tests/            # Unit tests│   ├── __init__.py│   ├── test_memfs.py       # Tests for memfs module│   └── test_api.py         # Tests for API module└── examples/         # Usage examples    ├── basic_usage.py      # Basic operations    └── advanced_usage.py   # Advanced scenarios

API Reference

MemoryFS Class

Extended MemoryFS Class

API Module

Use Cases

License