📘 Functional Data Structures: Persistent Types

🎯 Introduction

Welcome to this exciting tutorial on functional data structures and persistent types! 🎉 Have you ever wondered how to work with data that never changes, yet still allows you to create new versions efficiently? That’s the magic of persistent data structures!

In this guide, we’ll explore how persistent types can transform your Python development experience. Whether you’re building concurrent applications 🌐, implementing undo/redo functionality 🔄, or creating immutable data pipelines 📊, understanding persistent data structures is essential for writing robust, functional code.

By the end of this tutorial, you’ll feel confident using persistent types in your own projects! Let’s dive into this fascinating world! 🏊‍♂️

📚 Understanding Persistent Data Structures

🤔 What are Persistent Data Structures?

Persistent data structures are like a time machine for your data 🕰️. Think of them as creating a new photo 📸 every time you make a change, rather than erasing and redrawing on the same canvas. The original never changes, but you get a new version with your modifications!

In Python terms, persistent data structures maintain their previous versions when modified. This means you can:

• ✨ Keep all historical versions of your data
• 🚀 Share data safely between threads without locks
• 🛡️ Implement undo/redo with ease
• 🎯 Debug by examining past states

💡 Why Use Persistent Types?

Here’s why developers love persistent data structures:

1. Immutability by Design 🔒: Data never changes unexpectedly
2. Thread Safety 🌟: No race conditions or locks needed
3. Time Travel ⏰: Access any previous version instantly
4. Memory Efficiency 💾: Structural sharing saves space
5. Functional Programming 🎨: Perfect for pure functions

Real-world example: Imagine building a collaborative text editor 📝. With persistent data structures, multiple users can edit simultaneously without conflicts, and you get version history for free!

🔧 Basic Syntax and Usage

📝 Simple Example with pyrsistent

Let’s start with a friendly example using the pyrsistent library:

# 👋 Hello, Persistent Types!
from pyrsistent import pvector, pmap, pset

# 🎨 Creating a persistent vector
numbers = pvector([1, 2, 3, 4, 5])
print(f"Original: {numbers}")  # 📊 Original: pvector([1, 2, 3, 4, 5])

# ✨ Adding an element creates a new version
new_numbers = numbers.append(6)
print(f"New version: {new_numbers}")  # 🆕 pvector([1, 2, 3, 4, 5, 6])
print(f"Original unchanged: {numbers}")  # 🛡️ pvector([1, 2, 3, 4, 5])

# 🎯 Creating a persistent map (dictionary)
user = pmap({
    'name': 'Alice',
    'score': 100,
    'level': 1
})

# 🚀 Updating creates a new version
leveled_up_user = user.set('level', 2).set('score', 150)
print(f"Original user: {user}")  # 👤 Original state preserved
print(f"Leveled up: {leveled_up_user}")  # 🎊 New state created

💡 Explanation: Notice how we use emojis in comments to make code more readable! Each modification returns a new version while preserving the original.

🎯 Common Patterns

Here are patterns you’ll use daily:

from pyrsistent import pvector, pmap, freeze, thaw

# 🏗️ Pattern 1: Converting from regular Python types
regular_list = [1, 2, 3, 4, 5]
persistent_list = freeze(regular_list)  # 🧊 Freeze it!

# 🎨 Pattern 2: Building complex structures
game_state = pmap({
    'player': pmap({
        'name': 'Hero',
        'hp': 100,
        'inventory': pvector(['sword', 'shield'])
    }),
    'enemies': pvector([
        pmap({'type': 'goblin', 'hp': 20}),
        pmap({'type': 'dragon', 'hp': 200})
    ])
})

# 🔄 Pattern 3: Transforming data
def take_damage(state, damage):
    current_hp = state['player']['hp']
    return state.set_in(['player', 'hp'], max(0, current_hp - damage))

# 💫 Each transformation creates a new state
damaged_state = take_damage(game_state, 30)
print(f"Original HP: {game_state['player']['hp']}")  # 💚 100
print(f"After damage: {damaged_state['player']['hp']}")  # 💛 70

💡 Practical Examples

🛒 Example 1: Shopping Cart with History

Let’s build a shopping cart that tracks all changes:

from pyrsistent import pmap, pvector
from datetime import datetime

# 🛍️ Define our shopping cart system
class PersistentCart:
    def __init__(self):
        self.history = pvector([pmap({
            'items': pvector([]),
            'timestamp': datetime.now(),
            'action': 'cart_created'
        })])
    
    @property
    def current(self):
        # 🎯 Always get the latest state
        return self.history[-1]
    
    def add_item(self, product, quantity=1):
        # ➕ Add item to cart
        current_items = self.current['items']
        
        # 🔍 Check if item already exists
        for i, item in enumerate(current_items):
            if item['product'] == product:
                # 📈 Update quantity
                updated_item = item.set('quantity', item['quantity'] + quantity)
                new_items = current_items.set(i, updated_item)
                break
        else:
            # 🆕 Add new item
            new_items = current_items.append(pmap({
                'product': product,
                'quantity': quantity
            }))
        
        # 📸 Create new snapshot
        new_state = pmap({
            'items': new_items,
            'timestamp': datetime.now(),
            'action': f'added_{product}'
        })
        self.history = self.history.append(new_state)
        print(f"✅ Added {product} to cart!")
    
    def remove_item(self, product):
        # ➖ Remove item from cart
        current_items = self.current['items']
        new_items = pvector([item for item in current_items 
                           if item['product'] != product])
        
        new_state = pmap({
            'items': new_items,
            'timestamp': datetime.now(),
            'action': f'removed_{product}'
        })
        self.history = self.history.append(new_state)
        print(f"🗑️ Removed {product} from cart!")
    
    def undo(self):
        # ⏪ Go back to previous state
        if len(self.history) > 1:
            self.history = self.history[:-1]
            print("↩️ Undid last action!")
        else:
            print("⚠️ Nothing to undo!")
    
    def show_cart(self):
        # 📋 Display current cart
        print("\n🛒 Current Cart:")
        total = 0
        for item in self.current['items']:
            price = {'apple': 0.50, 'banana': 0.30, 'coffee': 4.99}.get(
                item['product'], 1.00
            )
            subtotal = price * item['quantity']
            total += subtotal
            print(f"  {item['product']} x{item['quantity']} = ${subtotal:.2f}")
        print(f"💰 Total: ${total:.2f}")
    
    def show_history(self):
        # 📜 Show action history
        print("\n📜 Cart History:")
        for i, state in enumerate(self.history):
            print(f"  {i}: {state['action']} at {state['timestamp'].strftime('%H:%M:%S')}")

# 🎮 Let's use it!
cart = PersistentCart()
cart.add_item('apple', 3)
cart.add_item('banana', 2)
cart.add_item('coffee', 1)
cart.show_cart()

cart.remove_item('banana')
cart.show_cart()

# 😱 Oops! Let's undo that
cart.undo()
cart.show_cart()

cart.show_history()

🎯 Try it yourself: Add a redo method and implement cart merging for multiple users!

🎮 Example 2: Game State Manager

Let’s make a game with save states:

from pyrsistent import pmap, pvector, pset
import copy

# 🏆 Game state manager with persistent data
class GameStateManager:
    def __init__(self):
        self.states = {}  # 💾 Named save states
        self.current_state = self._create_initial_state()
        self.checkpoint_stack = pvector([self.current_state])
    
    def _create_initial_state(self):
        # 🎮 Initial game state
        return pmap({
            'player': pmap({
                'name': 'Hero',
                'hp': 100,
                'mp': 50,
                'level': 1,
                'exp': 0,
                'position': pmap({'x': 0, 'y': 0}),
                'inventory': pvector(['potion', 'sword']),
                'skills': pset(['slash', 'heal'])
            }),
            'world': pmap({
                'current_area': 'town',
                'unlocked_areas': pset(['town']),
                'npcs_met': pset([]),
                'quests_completed': pset([])
            }),
            'stats': pmap({
                'play_time': 0,
                'enemies_defeated': 0,
                'items_collected': 0
            })
        })
    
    def move_player(self, dx, dy):
        # 🚶 Move player and create checkpoint
        old_pos = self.current_state['player']['position']
        new_pos = pmap({'x': old_pos['x'] + dx, 'y': old_pos['y'] + dy})
        
        self.current_state = self.current_state.set_in(
            ['player', 'position'], new_pos
        )
        self._add_checkpoint()
        print(f"🚶 Moved to ({new_pos['x']}, {new_pos['y']})")
    
    def gain_exp(self, amount):
        # ⭐ Gain experience and maybe level up
        player = self.current_state['player']
        new_exp = player['exp'] + amount
        new_level = player['level']
        
        # 🎊 Check for level up
        if new_exp >= new_level * 100:
            new_level += 1
            new_exp = new_exp - (player['level'] * 100)
            print(f"🎉 LEVEL UP! You're now level {new_level}!")
            
            # 💪 Increase stats
            player = player.set('hp', player['hp'] + 10)
            player = player.set('mp', player['mp'] + 5)
        
        player = player.set('exp', new_exp).set('level', new_level)
        self.current_state = self.current_state.set('player', player)
        self._add_checkpoint()
        print(f"⭐ Gained {amount} exp! (Total: {new_exp})")
    
    def learn_skill(self, skill_name):
        # 📚 Learn a new skill
        current_skills = self.current_state['player']['skills']
        if skill_name not in current_skills:
            new_skills = current_skills.add(skill_name)
            self.current_state = self.current_state.set_in(
                ['player', 'skills'], new_skills
            )
            self._add_checkpoint()
            print(f"✨ Learned new skill: {skill_name}!")
        else:
            print(f"💭 You already know {skill_name}")
    
    def save_game(self, slot_name):
        # 💾 Save current state
        self.states[slot_name] = self.current_state
        print(f"💾 Game saved to slot: {slot_name}")
    
    def load_game(self, slot_name):
        # 📂 Load saved state
        if slot_name in self.states:
            self.current_state = self.states[slot_name]
            self.checkpoint_stack = pvector([self.current_state])
            print(f"📂 Game loaded from slot: {slot_name}")
        else:
            print(f"❌ No save found in slot: {slot_name}")
    
    def _add_checkpoint(self):
        # 📸 Add automatic checkpoint
        self.checkpoint_stack = self.checkpoint_stack.append(self.current_state)
        # Keep only last 10 checkpoints
        if len(self.checkpoint_stack) > 10:
            self.checkpoint_stack = self.checkpoint_stack[-10:]
    
    def rewind(self, steps=1):
        # ⏪ Rewind to previous checkpoint
        if len(self.checkpoint_stack) > steps:
            self.checkpoint_stack = self.checkpoint_stack[:-steps]
            self.current_state = self.checkpoint_stack[-1]
            print(f"⏪ Rewound {steps} step(s)")
        else:
            print("⚠️ Can't rewind that far!")
    
    def show_status(self):
        # 📊 Display current status
        player = self.current_state['player']
        print(f"\n🎮 {player['name']} - Level {player['level']}")
        print(f"❤️ HP: {player['hp']} | 💙 MP: {player['mp']}")
        print(f"⭐ EXP: {player['exp']}/{player['level'] * 100}")
        print(f"📍 Position: ({player['position']['x']}, {player['position']['y']})")
        print(f"🎯 Skills: {', '.join(player['skills'])}")
        print(f"🎒 Inventory: {', '.join(player['inventory'])}")

# 🎮 Let's play!
game = GameStateManager()
game.show_status()

# 🎯 Play the game
game.move_player(5, 3)
game.gain_exp(50)
game.learn_skill('fireball')
game.save_game('checkpoint1')

game.move_player(2, -1)
game.gain_exp(100)  # Level up!
game.learn_skill('ice_shield')

# 😱 Oh no! Let's rewind
game.rewind(2)
game.show_status()

# 💾 Or load our save
game.load_game('checkpoint1')
game.show_status()

🚀 Advanced Concepts

🧙‍♂️ Advanced Topic 1: Custom Persistent Types

When you’re ready to level up, create your own persistent types:

from pyrsistent import PClass, field, pvector_field, pmap_field

# 🎯 Define a persistent class
class PersistentUser(PClass):
    name = field(type=str)
    email = field(type=str)
    scores = pvector_field(int)
    preferences = pmap_field(str, str)
    
    def add_score(self, score):
        # ✨ Returns new instance with added score
        return self.set('scores', self.scores.append(score))
    
    def update_preference(self, key, value):
        # 🔧 Returns new instance with updated preference
        return self.set('preferences', self.preferences.set(key, value))
    
    def get_average_score(self):
        # 📊 Calculate average without modifying state
        if not self.scores:
            return 0
        return sum(self.scores) / len(self.scores)

# 🪄 Using our custom type
user = PersistentUser(
    name="Alice",
    email="[email protected]",
    scores=pvector([85, 92, 78]),
    preferences=pmap({'theme': 'dark', 'language': 'en'})
)

# 🚀 All operations return new instances
user2 = user.add_score(95)
user3 = user2.update_preference('theme', 'light')

print(f"Original avg: {user.get_average_score():.1f}")  # 85.0
print(f"After new score: {user2.get_average_score():.1f}")  # 87.5
print(f"Theme unchanged: {user.preferences['theme']}")  # dark
print(f"New theme: {user3.preferences['theme']}")  # light

🏗️ Advanced Topic 2: Structural Sharing

For the brave developers, let’s explore how persistent types save memory:

from pyrsistent import pvector
import sys

# 🚀 Demonstrating structural sharing
def show_sharing():
    # 📊 Create a large vector
    original = pvector(range(10000))
    print(f"Original size: {sys.getsizeof(original)} bytes")
    
    # ✨ Append one element
    modified = original.append(10000)
    print(f"Modified size: {sys.getsizeof(modified)} bytes")
    
    # 🎯 Most of the structure is shared!
    # Only the changed parts are new
    
    # 🔍 Let's visualize with smaller example
    v1 = pvector([1, 2, 3, 4])
    v2 = v1.set(2, 'changed')
    v3 = v1.append(5)
    
    print("\n🌳 Structural sharing:")
    print(f"v1: {v1}")
    print(f"v2: {v2} (shares elements 0,1,3 with v1)")
    print(f"v3: {v3} (shares all elements with v1)")

# 💫 Advanced: Creating a persistent trie
class TrieNode:
    def __init__(self, value=None):
        self.value = value
        self.children = pmap({})
    
    def insert(self, key, value):
        # 🌟 Persistent insertion
        if not key:
            return TrieNode(value).set_children(self.children)
        
        first, *rest = key
        child = self.children.get(first, TrieNode())
        new_child = child.insert(rest, value)
        new_children = self.children.set(first, new_child)
        
        new_node = TrieNode(self.value)
        new_node.children = new_children
        return new_node
    
    def set_children(self, children):
        self.children = children
        return self

show_sharing()

⚠️ Common Pitfalls and Solutions

😱 Pitfall 1: Forgetting Immutability

# ❌ Wrong way - trying to modify in place
from pyrsistent import pvector

numbers = pvector([1, 2, 3])
numbers.append(4)  # This doesn't modify numbers!
print(numbers)  # Still [1, 2, 3] 😰

# ✅ Correct way - capture the new version
numbers = pvector([1, 2, 3])
numbers = numbers.append(4)  # Reassign to new version
print(numbers)  # Now [1, 2, 3, 4] 🎉

🤯 Pitfall 2: Performance Concerns

# ❌ Inefficient - creating many intermediate versions
from pyrsistent import pvector

result = pvector([])
for i in range(1000):
    result = result.append(i)  # 💥 Creates 1000 versions!

# ✅ Efficient - use evolver for bulk updates
from pyrsistent import pvector

v = pvector([])
e = v.evolver()  # 🔧 Create an evolver
for i in range(1000):
    e.append(i)  # 🚀 Efficient mutations
result = e.persistent()  # 📸 Get final version

# ✅ Even better - convert at the end
regular_list = []
for i in range(1000):
    regular_list.append(i)
result = pvector(regular_list)  # 🎯 Convert once

🛠️ Best Practices

1. 🎯 Use Evolvers for Bulk Updates: Don’t create many intermediate versions
2. 📝 Freeze at Boundaries: Convert to persistent types at API boundaries
3. 🛡️ Type Hints: Use type hints with persistent types
4. 🎨 Choose Right Structure: PVector for lists, PMap for dicts, PSet for sets
5. ✨ Leverage Immutability: Design functions that return new states

🧪 Hands-On Exercise

🎯 Challenge: Build a Persistent Undo/Redo System

Create a text editor with full undo/redo functionality:

📋 Requirements:

• ✅ Store document as persistent data structure
• 🏷️ Track cursor position with each change
• 👤 Support multiple operations (insert, delete, replace)
• 📅 Add timestamps to each change
• 🎨 Implement branching history (like Git)

🚀 Bonus Points:

• Add text highlighting
• Implement collaborative editing
• Create operation merging

💡 Solution

from pyrsistent import pmap, pvector
from datetime import datetime
import uuid

# 🎯 Our persistent text editor!
class PersistentTextEditor:
    def __init__(self):
        self.history = pvector([self._create_state("", 0)])
        self.current_index = 0
        self.branches = pmap({})  # 🌳 For branching history
    
    def _create_state(self, text, cursor, operation="init"):
        # 📸 Create a state snapshot
        return pmap({
            'id': str(uuid.uuid4()),
            'text': text,
            'cursor': cursor,
            'timestamp': datetime.now(),
            'operation': operation
        })
    
    @property
    def current(self):
        return self.history[self.current_index]
    
    def insert(self, text_to_insert):
        # ➕ Insert text at cursor
        current_text = self.current['text']
        cursor = self.current['cursor']
        
        new_text = (current_text[:cursor] + 
                   text_to_insert + 
                   current_text[cursor:])
        new_cursor = cursor + len(text_to_insert)
        
        self._add_state(new_text, new_cursor, f"insert '{text_to_insert}'")
        print(f"✏️ Inserted: '{text_to_insert}'")
    
    def delete(self, count):
        # 🗑️ Delete characters before cursor
        current_text = self.current['text']
        cursor = self.current['cursor']
        
        if cursor >= count:
            new_text = current_text[:cursor-count] + current_text[cursor:]
            new_cursor = cursor - count
            self._add_state(new_text, new_cursor, f"delete {count} chars")
            print(f"🗑️ Deleted {count} character(s)")
        else:
            print("⚠️ Not enough characters to delete")
    
    def move_cursor(self, position):
        # 📍 Move cursor
        current_text = self.current['text']
        if 0 <= position <= len(current_text):
            self._add_state(current_text, position, f"move to {position}")
            print(f"📍 Cursor moved to position {position}")
        else:
            print("⚠️ Invalid cursor position")
    
    def _add_state(self, text, cursor, operation):
        # 📝 Add new state to history
        new_state = self._create_state(text, cursor, operation)
        
        # If we're not at the end, we're creating a branch
        if self.current_index < len(self.history) - 1:
            # 🌿 Create a branch
            branch_name = f"branch_{len(self.branches)}"
            self.branches = self.branches.set(
                branch_name,
                self.history[self.current_index + 1:]
            )
            print(f"🌿 Created branch: {branch_name}")
            
            # Truncate main history
            self.history = self.history[:self.current_index + 1]
        
        self.history = self.history.append(new_state)
        self.current_index += 1
    
    def undo(self):
        # ⏪ Undo last operation
        if self.current_index > 0:
            self.current_index -= 1
            op = self.history[self.current_index + 1]['operation']
            print(f"↩️ Undid: {op}")
        else:
            print("⚠️ Nothing to undo")
    
    def redo(self):
        # ⏩ Redo operation
        if self.current_index < len(self.history) - 1:
            self.current_index += 1
            op = self.current['operation']
            print(f"↪️ Redid: {op}")
        else:
            print("⚠️ Nothing to redo")
    
    def show_document(self):
        # 📄 Display current document state
        state = self.current
        text = state['text']
        cursor = state['cursor']
        
        # Insert cursor marker
        display_text = text[:cursor] + '|' + text[cursor:]
        
        print(f"\n📄 Document (cursor at {cursor}):")
        print(f"'{display_text}'")
        print(f"🕒 Last operation: {state['operation']}")
    
    def show_history(self):
        # 📜 Show operation history
        print("\n📜 History:")
        for i, state in enumerate(self.history):
            marker = "→" if i == self.current_index else " "
            print(f"{marker} {i}: {state['operation']} at "
                  f"{state['timestamp'].strftime('%H:%M:%S')}")
        
        if self.branches:
            print("\n🌳 Branches:")
            for name, states in self.branches.items():
                print(f"  {name}: {len(states)} state(s)")

# 🎮 Test our editor!
editor = PersistentTextEditor()

# Type some text
editor.insert("Hello ")
editor.insert("World")
editor.insert("!")
editor.show_document()

# Make some edits
editor.move_cursor(6)
editor.insert("Python ")
editor.show_document()

# Try undo/redo
editor.undo()
editor.undo()
editor.show_document()

# Create a branch by making different edits
editor.insert("Beautiful ")
editor.show_document()

editor.show_history()

🎓 Key Takeaways

You’ve learned so much! Here’s what you can now do:

• ✅ Create persistent data structures with confidence 💪
• ✅ Avoid mutation bugs that trip up many developers 🛡️
• ✅ Apply functional programming patterns in real projects 🎯
• ✅ Debug with time travel like a pro 🐛
• ✅ Build thread-safe applications with persistent types! 🚀

Remember: Persistent data structures are your friend for safe, functional programming! They help you write code that’s easier to reason about and debug. 🤝

🤝 Next Steps

Congratulations! 🎉 You’ve mastered persistent data structures in Python!

Here’s what to do next:

1. 💻 Practice with the exercises above
2. 🏗️ Build a small project using pyrsistent
3. 📚 Explore other libraries like immutables or pydantic
4. 🌟 Share your functional programming journey with others!

Remember: Every functional programming expert was once a beginner. Keep coding, keep learning, and most importantly, have fun with immutable data! 🚀

Happy coding! 🎉🚀✨