from memfs import create_fs # Create a file system instance fs = create_fs() # Write to a file fs.writefile('/hello.txt', 'Hello, world!') # Read from a file content = fs.readfile('/hello.txt') print(content) # Outputs: Hello, world! # Check if a file exists if fs.exists('/hello.txt'): print('File exists!') # Create directories fs.makedirs('/path/to/directory') # List directory contents files = fs.listdir('/path/to')

from memfs import create_fs fs = create_fs() # Write using a file-like object with fs.open('/data.txt', 'w') as f: f.write('Line 1\n') f.write('Line 2\n') # Read using a file-like object with fs.open('/data.txt', 'r') as f: for line in f: print(line.strip())

from memfs import create_fs fs = create_fs() # Create nested directories fs.makedirs('/a/b/c') # Walk the directory tree for root, dirs, files in fs.walk('/'): print(f"Directory: {root}") print(f"Subdirectories: {dirs}") print(f"Files: {files}")

from memfs import create_fs import json import csv fs = create_fs() # Create directories fs.makedirs('/data/raw', exist_ok=True) fs.makedirs('/data/processed', exist_ok=True) # Write CSV data with fs.open('/data/raw/input.csv', 'w', newline='') as f: writer = csv.writer(f) writer.writerows([ ['id', 'name', 'value'], [1, 'Alpha', 100], [2, 'Beta', 200] ]) # Process CSV to JSON with fs.open('/data/raw/input.csv', 'r', newline='') as f: reader = csv.DictReader(f) data = [row for row in reader] # Transform and save the data for item in data: item['value'] = int(item['value']) item['double_value'] = item['value'] * 2 with fs.open('/data/processed/output.json', 'w') as f: json.dump(data, f, indent=2)

from memfs import create_fs import json import concurrent.futures fs = create_fs() fs.makedirs('/parallel/input', exist_ok=True) fs.makedirs('/parallel/output', exist_ok=True) # Create input files for i in range(10): fs.writefile(f'/parallel/input/file_{i}.json', json.dumps({'id': i})) def process_file(filename): with fs.open(f'/parallel/input/{filename}', 'r') as f: data = json.loads(f.read()) # Process data data['processed'] = True with fs.open(f'/parallel/output/processed_{filename}', 'w') as f: f.write(json.dumps(data, indent=2)) return data['id'] # Process files in parallel with concurrent.futures.ThreadPoolExecutor(max_workers=4) as executor: futures = {executor.submit(process_file, f): f for f in fs.listdir('/parallel/input')} for future in concurrent.futures.as_completed(futures): file_id = future.result() print(f"Processed file ID: {file_id}")

from memfs.examples.custom_filesystem import create_extended_fs # Create extended filesystem with encryption and compression fs = create_extended_fs() # Write to an encrypted file with fs.open_encrypted('/secret.txt', 'w', password='mysecret') as f: f.write('This is sensitive information') # Read from an encrypted file with fs.open_encrypted('/secret.txt', 'r', password='mysecret') as f: content = f.read() print(content) # Write to a compressed file (good for large text) with fs.open_compressed('/compressed.txt', 'w', compression_level=9) as f: f.write('This content will be compressed ' * 1000) # Check the file sizes normal_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")

from memfs import create_fs from memfs.api import DynamicgRPCComponent, PipelineOrchestrator # Define transformation functions def transform_data(data): if isinstance(data, dict): data['transformed'] = True return data def format_data(data): if isinstance(data, dict): data['formatted'] = True return data # Create virtual directories fs = 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 components transform_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 pipeline pipeline = 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}

#!/usr/bin/env python from memfs import create_fs fs = create_fs() fs.makedirs('/data', exist_ok=True) fs.writefile('/data/hello.txt', 'Hello, world!') print("Filesystem contents:") for root, dirs, files in fs.walk('/'): print(f"Directory: {root}") for d in dirs: print(f" Dir: {d}") for f in files: print(f" File: {f}")