Advanced Python Data Structures: Collections and Custom Implementations

Explore advanced data structures in Python beyond the basic list, dict, and tuple. This comprehensive guide covers the collections module, custom implementations, and performance optimization techniques.

Table of Contents #

Collections Module Overview
Advanced Dictionary Structures
Specialized Lists and Queues
Custom Data Structure Implementations
Performance Analysis
Memory Optimization
Real-World Applications

Collections Module Overview #

defaultdict: Dictionaries with Default Values #

Eliminate KeyError with automatic default values:

🐍 Try it yourself

from collections import defaultdict

# Traditional dictionary approach
traditional_dict = {}
words = ['apple', 'banana', 'apple', 'cherry', 'banana', 'apple']

for word in words:
    if word in traditional_dict:
        traditional_dict[word] += 1
    else:
        traditional_dict[word] = 1

print("Traditional approach:", traditional_dict)

# defaultdict approach
word_count = defaultdict(int)
for word in words:
    word_count[word] += 1

print("defaultdict approach:", dict(word_count))

# Complex nested structures
nested_data = defaultdict(lambda: defaultdict(list))

# Adding data without initialization
nested_data['fruits']['red'].append('apple')
nested_data['fruits']['red'].append('strawberry')
nested_data['fruits']['yellow'].append('banana')
nested_data['vegetables']['green'].append('lettuce')

print("\nNested defaultdict:")
for category, items in nested_data.items():
    print(f"  {category}:")
    for color, products in items.items():
        print(f"    {color}: {products}")

# Group data by multiple criteria
student_grades = [
    ('Alice', 'Math', 85), ('Alice', 'Science', 92), ('Alice', 'English', 88),
    ('Bob', 'Math', 78), ('Bob', 'Science', 85), ('Bob', 'English', 82),
    ('Charlie', 'Math', 92), ('Charlie', 'Science', 88), ('Charlie', 'English', 90)
]

grades_by_student = defaultdict(lambda: defaultdict(list))

for student, subject, grade in student_grades:
    grades_by_student[student][subject].append(grade)

print("\nStudent grades structure:")
for student, subjects in grades_by_student.items():
    avg_grades = {subj: sum(grades)/len(grades) for subj, grades in subjects.items()}
    print(f"  {student}: {dict(avg_grades)}")

Output:

Click "Run Code" to see the output

Counter: Frequency Counting Made Easy #

Powerful counting and frequency analysis:

🐍 Try it yourself

from collections import Counter

# Basic counting
text = "hello world hello python world programming"
words = text.split()

word_counter = Counter(words)
print("Word frequencies:", word_counter)
print("Most common words:", word_counter.most_common(3))

# Mathematical operations with counters
counter1 = Counter(['a', 'b', 'c', 'a', 'b'])
counter2 = Counter(['a', 'b', 'b', 'd'])

print(f"\nCounter 1: {counter1}")
print(f"Counter 2: {counter2}")
print(f"Addition: {counter1 + counter2}")
print(f"Subtraction: {counter1 - counter2}")
print(f"Intersection: {counter1 & counter2}")
print(f"Union: {counter1 | counter2}")

# Character frequency analysis
def analyze_text(text):
    char_count = Counter(text.lower())
    
    # Remove whitespace from analysis
    char_count.pop(' ', None)
    
    total_chars = sum(char_count.values())
    
    analysis = {
        'total_characters': total_chars,
        'unique_characters': len(char_count),
        'most_common': char_count.most_common(5),
        'frequency_distribution': {
            char: count / total_chars * 100
            for char, count in char_count.most_common(5)
        }
    }
    
    return analysis

sample_text = "Python is a powerful programming language"
analysis = analyze_text(sample_text)

print(f"\nText analysis for: '{sample_text}'")
print(f"Total characters: {analysis['total_characters']}")
print(f"Unique characters: {analysis['unique_characters']}")
print("Most common characters:")
for char, freq in analysis['frequency_distribution'].items():
    print(f"  '{char}': {freq:.1f}%")

# Counter with custom objects
class Product:
    def __init__(self, name, category):
        self.name = name
        self.category = category
    
    def __hash__(self):
        return hash((self.name, self.category))
    
    def __eq__(self, other):
        return isinstance(other, Product) and \
               self.name == other.name and self.category == other.category
    
    def __repr__(self):
        return f"Product('{self.name}', '{self.category}')"

products = [
    Product('iPhone', 'Electronics'),
    Product('MacBook', 'Electronics'),
    Product('iPhone', 'Electronics'),
    Product('Shirt', 'Clothing'),
    Product('iPhone', 'Electronics')
]

product_counter = Counter(products)
print(f"\nProduct sales count:")
for product, count in product_counter.most_common():
    print(f"  {product}: {count} sales")

Output:

Click "Run Code" to see the output

deque: Double-Ended Queue #

Efficient operations at both ends:

🐍 Try it yourself

from collections import deque

# Basic deque operations
dq = deque([1, 2, 3, 4, 5])
print(f"Initial deque: {dq}")

# Efficient operations at both ends
dq.appendleft(0)
dq.append(6)
print(f"After appendleft(0) and append(6): {dq}")

left_item = dq.popleft()
right_item = dq.pop()
print(f"Popped {left_item} from left and {right_item} from right: {dq}")

# Rotating elements
dq.rotate(2)  # Rotate right by 2
print(f"After rotate(2): {dq}")

dq.rotate(-3)  # Rotate left by 3
print(f"After rotate(-3): {dq}")

# Fixed-size deque (circular buffer)
circular_buffer = deque(maxlen=5)

for i in range(8):
    circular_buffer.append(i)
    print(f"After adding {i}: {list(circular_buffer)}")

# Sliding window implementation
def sliding_window_max(arr, window_size):
    """Find maximum in each sliding window using deque"""
    if not arr or window_size <= 0:
        return []
    
    dq = deque()  # Store indices
    result = []
    
    for i in range(len(arr)):
        # Remove indices outside current window
        while dq and dq[0] <= i - window_size:
            dq.popleft()
        
        # Remove indices of smaller elements
        while dq and arr[dq[-1]] <= arr[i]:
            dq.pop()
        
        dq.append(i)
        
        # Add maximum of current window to result
        if i >= window_size - 1:
            result.append(arr[dq[0]])
    
    return result

# Test sliding window
test_array = [1, 3, -1, -3, 5, 3, 6, 7]
window_size = 3

max_values = sliding_window_max(test_array, window_size)
print(f"\nSliding window maximum:")
print(f"Array: {test_array}")
print(f"Window size: {window_size}")
print(f"Maximum values: {max_values}")

# Task queue simulation
task_queue = deque()

# Add tasks
tasks = ['task1', 'task2', 'task3', 'urgent_task', 'task4']
for task in tasks:
    if 'urgent' in task:
        task_queue.appendleft(task)  # Priority task
    else:
        task_queue.append(task)  # Regular task

print(f"\nTask queue: {list(task_queue)}")

# Process tasks
print("Processing tasks:")
while task_queue:
    current_task = task_queue.popleft()
    print(f"  Processing: {current_task}")

Output:

Click "Run Code" to see the output

Advanced Dictionary Structures #

OrderedDict: Maintaining Insertion Order #

Preserve order in dictionary operations:

🐍 Try it yourself

from collections import OrderedDict

# OrderedDict maintains insertion order
regular_dict = {}
ordered_dict = OrderedDict()

items = [('c', 3), ('a', 1), ('b', 2), ('d', 4)]

for key, value in items:
    regular_dict[key] = value
    ordered_dict[key] = value

print("Regular dict:", regular_dict)
print("Ordered dict:", ordered_dict)

# Moving items to end
ordered_dict.move_to_end('a')
print("After moving 'a' to end:", ordered_dict)

ordered_dict.move_to_end('c', last=False)
print("After moving 'c' to beginning:", ordered_dict)

# LRU Cache implementation using OrderedDict
class LRUCache:
    def __init__(self, capacity):
        self.capacity = capacity
        self.cache = OrderedDict()
    
    def get(self, key):
        if key in self.cache:
            # Move to end (most recently used)
            self.cache.move_to_end(key)
            return self.cache[key]
        return None
    
    def put(self, key, value):
        if key in self.cache:
            # Update existing key
            self.cache[key] = value
            self.cache.move_to_end(key)
        else:
            # Add new key
            if len(self.cache) >= self.capacity:
                # Remove least recently used (first item)
                self.cache.popitem(last=False)
            self.cache[key] = value
    
    def __repr__(self):
        return f"LRUCache({dict(self.cache)})"

# Test LRU Cache
lru = LRUCache(3)

print(f"\nLRU Cache testing (capacity: 3):")
lru.put('a', 1)
print(f"After put('a', 1): {lru}")

lru.put('b', 2)
print(f"After put('b', 2): {lru}")

lru.put('c', 3)
print(f"After put('c', 3): {lru}")

print(f"Get 'a': {lru.get('a')}")
print(f"Cache after get('a'): {lru}")

lru.put('d', 4)  # Should evict 'b'
print(f"After put('d', 4): {lru}")

# Configuration management with ordered settings
class ConfigManager:
    def __init__(self):
        self.settings = OrderedDict()
        self.descriptions = {}
    
    def add_setting(self, key, value, description=""):
        self.settings[key] = value
        self.descriptions[key] = description
    
    def update_setting(self, key, value):
        if key in self.settings:
            # Maintain order by moving to end
            old_value = self.settings[key]
            self.settings[key] = value
            self.settings.move_to_end(key)
            return old_value
        else:
            raise KeyError(f"Setting '{key}' not found")
    
    def get_config_summary(self):
        summary = []
        for key, value in self.settings.items():
            desc = self.descriptions.get(key, "No description")
            summary.append(f"{key}: {value} ({desc})")
        return summary

# Test configuration manager
config = ConfigManager()
config.add_setting('debug', True, "Enable debug mode")
config.add_setting('host', 'localhost', "Server host")
config.add_setting('port', 8080, "Server port")
config.add_setting('timeout', 30, "Request timeout in seconds")

print(f"\nConfiguration settings:")
for setting in config.get_config_summary():
    print(f"  {setting}")

config.update_setting('debug', False)
print(f"\nAfter updating debug setting:")
for setting in config.get_config_summary():
    print(f"  {setting}")

Output:

Click "Run Code" to see the output

ChainMap: Combining Multiple Mappings #

Work with multiple dictionaries as a single mapping:

🐍 Try it yourself

from collections import ChainMap

# Basic ChainMap usage
default_config = {'debug': False, 'host': 'localhost', 'port': 8080}
user_config = {'debug': True, 'port': 3000}
env_config = {'host': '0.0.0.0'}

# Chain maps with priority: env_config > user_config > default_config
config = ChainMap(env_config, user_config, default_config)

print("Configuration lookup (with priority):")
print(f"  debug: {config['debug']}")  # From user_config
print(f"  host: {config['host']}")    # From env_config
print(f"  port: {config['port']}")    # From user_config

# View all mappings
print(f"\nAll mappings: {config.maps}")

# Add new mapping at runtime
runtime_config = {'timeout': 30}
config = config.new_child(runtime_config)

print(f"After adding runtime config: {config['timeout']}")
print(f"Updated maps: {len(config.maps)} mappings")

# Configuration inheritance system
class ConfigurationManager:
    def __init__(self):
        self.global_config = {'log_level': 'INFO', 'max_connections': 100}
        self.app_configs = {}
    
    def create_app_config(self, app_name, app_specific_config=None):
        if app_specific_config is None:
            app_specific_config = {}
        
        # Create chain: app_specific > global
        self.app_configs[app_name] = ChainMap(app_specific_config, self.global_config)
        return self.app_configs[app_name]
    
    def update_global_config(self, updates):
        self.global_config.update(updates)
    
    def get_app_config(self, app_name):
        return self.app_configs.get(app_name)
    
    def list_all_configs(self):
        return {
            'global': dict(self.global_config),
            'apps': {name: dict(config) for name, config in self.app_configs.items()}
        }

# Test configuration manager
config_mgr = ConfigurationManager()

# Create app-specific configurations
web_config = config_mgr.create_app_config('web_app', {
    'port': 8080,
    'debug': True
})

api_config = config_mgr.create_app_config('api_service', {
    'port': 3000,
    'rate_limit': 1000
})

worker_config = config_mgr.create_app_config('background_worker', {
    'workers': 4,
    'log_level': 'DEBUG'  # Override global setting
})

print(f"\nConfiguration for web_app:")
for key, value in web_config.items():
    print(f"  {key}: {value}")

print(f"\nConfiguration for background_worker:")
for key, value in worker_config.items():
    print(f"  {key}: {value}")

# Update global configuration
config_mgr.update_global_config({'max_connections': 200, 'retry_attempts': 3})

print(f"\nAfter global config update - web_app config:")
for key, value in web_config.items():
    print(f"  {key}: {value}")

# Environment variable simulation
def create_env_aware_config(base_config):
    # Simulated environment variables
    env_vars = {
        'DB_HOST': 'production-db.example.com',
        'DB_PORT': '5432',
        'DEBUG': 'false'
    }
    
    # Convert env vars to config format
    env_config = {}
    for key, value in env_vars.items():
        config_key = key.lower().replace('_', '.')
        
        # Type conversion
        if value.lower() in ('true', 'false'):
            env_config[config_key] = value.lower() == 'true'
        elif value.isdigit():
            env_config[config_key] = int(value)
        else:
            env_config[config_key] = value
    
    return ChainMap(env_config, base_config)

base_app_config = {
    'db.host': 'localhost',
    'db.port': 3306,
    'debug': True,
    'app.name': 'MyApp'
}

final_config = create_env_aware_config(base_app_config)

print(f"\nEnvironment-aware configuration:")
for key, value in final_config.items():
    print(f"  {key}: {value}")

Output:

Click "Run Code" to see the output

Specialized Lists and Queues #

Named Tuples: Structured Data with Named Fields #

Create lightweight, immutable data structures:

🐍 Try it yourself

from collections import namedtuple

# Basic namedtuple creation
Point = namedtuple('Point', ['x', 'y'])
point1 = Point(10, 20)

print(f"Point: {point1}")
print(f"X coordinate: {point1.x}")
print(f"Y coordinate: {point1.y}")

# Namedtuple with methods
Person = namedtuple('Person', ['name', 'age', 'email'])

def create_person(name, age, email):
    # Validation before creation
    if age < 0:
        raise ValueError("Age cannot be negative")
    if '@' not in email:
        raise ValueError("Invalid email format")
    
    return Person(name, age, email)

# Create person instances
people = [
    create_person('Alice Johnson', 28, 'alice@example.com'),
    create_person('Bob Smith', 35, 'bob@example.com'),
    create_person('Charlie Brown', 42, 'charlie@example.com')
]

print(f"\nPeople database:")
for person in people:
    print(f"  {person.name} ({person.age}) - {person.email}")

# Advanced namedtuple with defaults and methods
from collections import namedtuple

# Create namedtuple with defaults using _fields and _field_defaults
Employee = namedtuple('Employee', ['name', 'department', 'salary', 'start_date'])
Employee.__new__.__defaults__ = (None,) * len(Employee._fields)

# Create Employee with partial data
emp1 = Employee('Alice', 'Engineering', 85000, '2024-01-15')
emp2 = Employee('Bob', 'Marketing', 75000)  # No start_date

print(f"\nEmployees:")
print(f"  {emp1}")
print(f"  {emp2}")

# Extending namedtuple with methods
class EnhancedEmployee(namedtuple('Employee', ['name', 'department', 'salary', 'years_experience'])):
    
    def annual_bonus(self):
        """Calculate annual bonus based on salary and experience"""
        base_bonus = self.salary * 0.1
        experience_bonus = self.years_experience * 1000
        return base_bonus + experience_bonus
    
    def promotion_eligible(self):
        """Check if employee is eligible for promotion"""
        return self.years_experience >= 2 and self.salary < 100000
    
    def __str__(self):
        return f"{self.name} ({self.department}) - ${self.salary:,} with {self.years_experience} years experience"

# Test enhanced employee
enhanced_employees = [
    EnhancedEmployee('Diana', 'Engineering', 90000, 3),
    EnhancedEmployee('Eve', 'Sales', 70000, 1),
    EnhancedEmployee('Frank', 'Engineering', 95000, 5)
]

print(f"\nEnhanced employee analysis:")
for emp in enhanced_employees:
    bonus = emp.annual_bonus()
    eligible = emp.promotion_eligible()
    print(f"  {emp}")
    print(f"    Annual bonus: ${bonus:,.2f}")
    print(f"    Promotion eligible: {eligible}")

# Database record simulation
DatabaseRecord = namedtuple('DatabaseRecord', ['id', 'table', 'operation', 'timestamp', 'data'])

# Simulate database operations
import datetime

records = [
    DatabaseRecord(1, 'users', 'INSERT', datetime.datetime.now(), {'name': 'Alice', 'email': 'alice@example.com'}),
    DatabaseRecord(2, 'users', 'UPDATE', datetime.datetime.now(), {'id': 1, 'email': 'alice.johnson@example.com'}),
    DatabaseRecord(3, 'orders', 'INSERT', datetime.datetime.now(), {'user_id': 1, 'product': 'laptop', 'amount': 1299.99})
]

print(f"\nDatabase operation log:")
for record in records:
    print(f"  ID {record.id}: {record.operation} on {record.table}")
    print(f"    Time: {record.timestamp.strftime('%Y-%m-%d %H:%M:%S')}")
    print(f"    Data: {record.data}")

# Configuration tuple
ServerConfig = namedtuple('ServerConfig', ['host', 'port', 'debug', 'workers', 'timeout'])

# Create configurations for different environments
configs = {
    'development': ServerConfig('localhost', 8080, True, 1, 30),
    'staging': ServerConfig('staging.example.com', 8080, False, 2, 60),
    'production': ServerConfig('api.example.com', 443, False, 8, 120)
}

print(f"\nServer configurations:")
for env, config in configs.items():
    print(f"  {env.title()}: {config.host}:{config.port} "
          f"(debug={config.debug}, workers={config.workers})")

Output:

Click "Run Code" to see the output

Custom Data Structure Implementations #

Binary Search Tree #

Implement a custom BST for sorted data:

🐍 Try it yourself

# Binary Search Tree implementation
class TreeNode:
    def __init__(self, value):
        self.value = value
        self.left = None
        self.right = None
    
    def __repr__(self):
        return f"TreeNode({self.value})"

class BinarySearchTree:
    def __init__(self):
        self.root = None
    
    def insert(self, value):
        if self.root is None:
            self.root = TreeNode(value)
        else:
            self._insert_recursive(self.root, value)
    
    def _insert_recursive(self, node, value):
        if value <= node.value:
            if node.left is None:
                node.left = TreeNode(value)
            else:
                self._insert_recursive(node.left, value)
        else:
            if node.right is None:
                node.right = TreeNode(value)
            else:
                self._insert_recursive(node.right, value)
    
    def search(self, value):
        return self._search_recursive(self.root, value)
    
    def _search_recursive(self, node, value):
        if node is None:
            return False
        
        if value == node.value:
            return True
        elif value < node.value:
            return self._search_recursive(node.left, value)
        else:
            return self._search_recursive(node.right, value)
    
    def inorder_traversal(self):
        result = []
        self._inorder_recursive(self.root, result)
        return result
    
    def _inorder_recursive(self, node, result):
        if node is not None:
            self._inorder_recursive(node.left, result)
            result.append(node.value)
            self._inorder_recursive(node.right, result)
    
    def find_min(self):
        if self.root is None:
            return None
        
        current = self.root
        while current.left is not None:
            current = current.left
        return current.value
    
    def find_max(self):
        if self.root is None:
            return None
        
        current = self.root
        while current.right is not None:
            current = current.right
        return current.value
    
    def height(self):
        return self._height_recursive(self.root)
    
    def _height_recursive(self, node):
        if node is None:
            return -1
        
        left_height = self._height_recursive(node.left)
        right_height = self._height_recursive(node.right)
        
        return max(left_height, right_height) + 1

# Test BST
bst = BinarySearchTree()
values = [50, 30, 70, 20, 40, 60, 80, 10, 25, 35, 45]

print("Inserting values into BST:", values)
for value in values:
    bst.insert(value)

print(f"Inorder traversal (sorted): {bst.inorder_traversal()}")
print(f"Minimum value: {bst.find_min()}")
print(f"Maximum value: {bst.find_max()}")
print(f"Tree height: {bst.height()}")

# Search operations
search_values = [25, 45, 100, 30]
print(f"\nSearch results:")
for val in search_values:
    found = bst.search(val)
    print(f"  Search {val}: {'Found' if found else 'Not found'}")

Output:

Click "Run Code" to see the output

Priority Queue with Heap #

Implement a priority queue using heap operations:

🐍 Try it yourself

import heapq

# Priority Queue implementation
class PriorityQueue:
    def __init__(self):
        self._queue = []
        self._index = 0  # For tie-breaking
    
    def push(self, item, priority):
        # Use negative priority for max-heap behavior (if needed)
        heapq.heappush(self._queue, (priority, self._index, item))
        self._index += 1
    
    def pop(self):
        if self._queue:
            priority, index, item = heapq.heappop(self._queue)
            return item, priority
        else:
            raise IndexError("Priority queue is empty")
    
    def peek(self):
        if self._queue:
            priority, index, item = self._queue[0]
            return item, priority
        else:
            return None, None
    
    def is_empty(self):
        return len(self._queue) == 0
    
    def size(self):
        return len(self._queue)
    
    def __len__(self):
        return len(self._queue)
    
    def __repr__(self):
        items = [(priority, item) for priority, _, item in self._queue]
        return f"PriorityQueue({items})"

# Test priority queue
pq = PriorityQueue()

# Add tasks with priorities (lower number = higher priority)
tasks = [
    ("Send email", 3),
    ("Fix critical bug", 1),
    ("Update documentation", 5),
    ("Review code", 2),
    ("Deploy to production", 1)
]

print("Adding tasks to priority queue:")
for task, priority in tasks:
    pq.push(task, priority)
    print(f"  Added: {task} (priority {priority})")

print(f"\nPriority queue state: {pq}")

print(f"\nProcessing tasks by priority:")
while not pq.is_empty():
    task, priority = pq.pop()
    print(f"  Processing: {task} (priority {priority})")

# Advanced priority queue with custom objects
class Task:
    def __init__(self, name, priority, estimated_time, dependencies=None):
        self.name = name
        self.priority = priority
        self.estimated_time = estimated_time
        self.dependencies = dependencies or []
        self.completed = False
    
    def __repr__(self):
        return f"Task('{self.name}', priority={self.priority}, time={self.estimated_time}h)"

class TaskScheduler:
    def __init__(self):
        self.queue = PriorityQueue()
        self.completed_tasks = []
        self.task_registry = {}
    
    def add_task(self, task):
        self.task_registry[task.name] = task
        # Calculate effective priority considering dependencies
        effective_priority = self._calculate_effective_priority(task)
        self.queue.push(task, effective_priority)
    
    def _calculate_effective_priority(self, task):
        # Lower number = higher priority
        # Consider dependencies and estimated time
        dependency_penalty = len(task.dependencies) * 0.5
        time_factor = task.estimated_time * 0.1
        return task.priority + dependency_penalty + time_factor
    
    def can_execute(self, task):
        # Check if all dependencies are completed
        for dep_name in task.dependencies:
            dep_task = self.task_registry.get(dep_name)
            if dep_task and not dep_task.completed:
                return False
        return True
    
    def get_next_task(self):
        temp_storage = []
        
        while not self.queue.is_empty():
            task, priority = self.queue.pop()
            
            if self.can_execute(task) and not task.completed:
                # Restore temporarily removed tasks
                for stored_task, stored_priority in temp_storage:
                    self.queue.push(stored_task, stored_priority)
                return task
            else:
                temp_storage.append((task, priority))
        
        # Restore all tasks if none can be executed
        for stored_task, stored_priority in temp_storage:
            self.queue.push(stored_task, stored_priority)
        
        return None
    
    def complete_task(self, task):
        task.completed = True
        self.completed_tasks.append(task)
    
    def execute_schedule(self):
        print("Executing task schedule:")
        total_time = 0
        
        while True:
            next_task = self.get_next_task()
            if next_task is None:
                break
            
            print(f"  Executing: {next_task.name} "
                  f"(priority {next_task.priority}, {next_task.estimated_time}h)")
            
            total_time += next_task.estimated_time
            self.complete_task(next_task)
        
        print(f"\nSchedule completed in {total_time} hours")
        print(f"Completed tasks: {len(self.completed_tasks)}")
        
        if not self.queue.is_empty():
            print(f"Remaining tasks (blocked by dependencies): {self.queue.size()}")

# Test task scheduler
scheduler = TaskScheduler()

# Create tasks with dependencies
project_tasks = [
    Task("Design database schema", 2, 4),
    Task("Set up development environment", 1, 2),
    Task("Implement user authentication", 3, 6, ["Design database schema", "Set up development environment"]),
    Task("Create user interface", 3, 8, ["Set up development environment"]),
    Task("Write unit tests", 4, 4, ["Implement user authentication"]),
    Task("Deploy to staging", 2, 2, ["Implement user authentication", "Create user interface"]),
    Task("User acceptance testing", 3, 6, ["Deploy to staging"]),
    Task("Deploy to production", 1, 1, ["User acceptance testing", "Write unit tests"])
]

for task in project_tasks:
    scheduler.add_task(task)

scheduler.execute_schedule()

Output:

Click "Run Code" to see the output

Performance Analysis #

Time Complexity Comparison #

Compare performance of different data structures:

🐍 Try it yourself

import time
import random
from collections import deque

# Performance testing framework
def time_operation(operation, *args, **kwargs):
    start_time = time.perf_counter()
    result = operation(*args, **kwargs)
    end_time = time.perf_counter()
    return result, end_time - start_time

# Test data structures for different operations
def test_insertion_performance():
    sizes = [1000, 5000, 10000]
    
    print("Insertion Performance Comparison:")
    print("Size\t\tList\t\tDeque\t\tSet\t\tDict")
    print("-" * 60)
    
    for size in sizes:
        data = list(range(size))
        random.shuffle(data)
        
        # List insertion at end
        test_list = []
        _, list_time = time_operation(lambda: [test_list.append(x) for x in data])
        
        # Deque insertion at end
        test_deque = deque()
        _, deque_time = time_operation(lambda: [test_deque.append(x) for x in data])
        
        # Set insertion
        test_set = set()
        _, set_time = time_operation(lambda: [test_set.add(x) for x in data])
        
        # Dict insertion
        test_dict = {}
        _, dict_time = time_operation(lambda: [test_dict.update({x: x}) for x in data])
        
        print(f"{size}\t\t{list_time:.6f}\t{deque_time:.6f}\t{set_time:.6f}\t{dict_time:.6f}")

test_insertion_performance()

# Search performance comparison
def test_search_performance():
    size = 10000
    data = list(range(size))
    search_items = random.sample(data, 100)
    
    # Prepare data structures
    test_list = data.copy()
    test_set = set(data)
    test_dict = {x: x for x in data}
    
    print(f"\nSearch Performance (searching for 100 items in {size} elements):")
    
    # List search (linear)
    _, list_time = time_operation(lambda: [x in test_list for x in search_items])
    
    # Set search (hash-based)
    _, set_time = time_operation(lambda: [x in test_set for x in search_items])
    
    # Dict search (hash-based)
    _, dict_time = time_operation(lambda: [x in test_dict for x in search_items])
    
    print(f"List search time: {list_time:.6f} seconds")
    print(f"Set search time: {set_time:.6f} seconds")
    print(f"Dict search time: {dict_time:.6f} seconds")
    print(f"Set is {list_time/set_time:.1f}x faster than list")
    print(f"Dict is {list_time/dict_time:.1f}x faster than list")

test_search_performance()

# Memory usage analysis
def analyze_memory_usage():
    import sys
    
    size = 1000
    data = list(range(size))
    
    # Create different data structures
    test_list = data.copy()
    test_tuple = tuple(data)
    test_set = set(data)
    test_dict = {x: x for x in data}
    test_deque = deque(data)
    
    structures = [
        ("List", test_list),
        ("Tuple", test_tuple),
        ("Set", test_set),
        ("Dict", test_dict),
        ("Deque", test_deque)
    ]
    
    print(f"\nMemory Usage Comparison ({size} elements):")
    print("Structure\tMemory (bytes)\tPer Element")
    print("-" * 45)
    
    for name, structure in structures:
        memory = sys.getsizeof(structure)
        per_element = memory / size
        print(f"{name}\t\t{memory}\t\t{per_element:.2f}")

analyze_memory_usage()

# Big O complexity demonstration
def demonstrate_complexity():
    print(f"\nTime Complexity Demonstration:")
    
    sizes = [1000, 2000, 4000, 8000]
    
    print("\nO(1) - Dictionary lookup:")
    test_dict = {i: i for i in range(max(sizes))}
    
    for size in sizes:
        search_key = size // 2
        _, lookup_time = time_operation(lambda: test_dict.get(search_key))
        print(f"  Size {size}: {lookup_time:.8f} seconds")
    
    print("\nO(n) - List search:")
    for size in sizes:
        test_list = list(range(size))
        search_item = size // 2
        _, search_time = time_operation(lambda: search_item in test_list)
        print(f"  Size {size}: {search_time:.8f} seconds")
    
    print("\nO(n log n) - Sorting:")
    for size in sizes:
        test_data = random.sample(range(size * 2), size)
        _, sort_time = time_operation(lambda: sorted(test_data))
        print(f"  Size {size}: {sort_time:.8f} seconds")

demonstrate_complexity()

Output:

Click "Run Code" to see the output

Memory Optimization #

Space-Efficient Data Structures #

Optimize memory usage for large datasets:

🐍 Try it yourself

import sys
from array import array

# Memory-efficient alternatives
def compare_memory_efficiency():
    size = 10000
    
    # Regular list of integers
    regular_list = list(range(size))
    
    # Array of integers (more memory efficient)
    int_array = array('i', range(size))
    
    # Generator (most memory efficient for one-time iteration)
    int_generator = (x for x in range(size))
    
    print("Memory Efficiency Comparison:")
    print(f"Regular list: {sys.getsizeof(regular_list)} bytes")
    print(f"Array: {sys.getsizeof(int_array)} bytes")
    print(f"Generator: {sys.getsizeof(int_generator)} bytes")
    
    memory_savings = sys.getsizeof(regular_list) / sys.getsizeof(int_array)
    print(f"Array is {memory_savings:.1f}x more memory efficient")

compare_memory_efficiency()

# Slots for memory-efficient classes
class RegularPerson:
    def __init__(self, name, age, email):
        self.name = name
        self.age = age
        self.email = email

class SlottedPerson:
    __slots__ = ['name', 'age', 'email']
    
    def __init__(self, name, age, email):
        self.name = name
        self.age = age
        self.email = email

# Compare memory usage of classes
def compare_class_memory():
    print(f"\nClass Memory Comparison:")
    
    regular = RegularPerson("Alice", 30, "alice@example.com")
    slotted = SlottedPerson("Alice", 30, "alice@example.com")
    
    print(f"Regular class instance: {sys.getsizeof(regular)} bytes")
    print(f"Slotted class instance: {sys.getsizeof(slotted)} bytes")
    
    # Test with multiple instances
    regular_instances = [RegularPerson(f"Person{i}", 20+i, f"person{i}@example.com") 
                        for i in range(1000)]
    slotted_instances = [SlottedPerson(f"Person{i}", 20+i, f"person{i}@example.com") 
                        for i in range(1000)]
    
    regular_total = sum(sys.getsizeof(obj) for obj in regular_instances)
    slotted_total = sum(sys.getsizeof(obj) for obj in slotted_instances)
    
    print(f"\n1000 regular instances: {regular_total} bytes")
    print(f"1000 slotted instances: {slotted_total} bytes")
    print(f"Memory savings: {(regular_total - slotted_total) / regular_total * 100:.1f}%")

compare_class_memory()

# Efficient data storage patterns
class EfficientDataStore:
    def __init__(self):
        # Store related data in separate arrays for better memory layout
        self.names = []
        self.ages = array('i')  # Integer array
        self.salaries = array('f')  # Float array
        self.active_flags = array('b')  # Boolean array (as bytes)
    
    def add_person(self, name, age, salary, active=True):
        self.names.append(name)
        self.ages.append(age)
        self.salaries.append(salary)
        self.active_flags.append(1 if active else 0)
    
    def get_person(self, index):
        return {
            'name': self.names[index],
            'age': self.ages[index],
            'salary': self.salaries[index],
            'active': bool(self.active_flags[index])
        }
    
    def get_memory_usage(self):
        return {
            'names': sys.getsizeof(self.names),
            'ages': sys.getsizeof(self.ages),
            'salaries': sys.getsizeof(self.salaries),
            'active_flags': sys.getsizeof(self.active_flags),
            'total': (sys.getsizeof(self.names) + 
                     sys.getsizeof(self.ages) + 
                     sys.getsizeof(self.salaries) + 
                     sys.getsizeof(self.active_flags))
        }
    
    def __len__(self):
        return len(self.names)

# Compare with traditional approach
class TraditionalDataStore:
    def __init__(self):
        self.people = []
    
    def add_person(self, name, age, salary, active=True):
        self.people.append({
            'name': name,
            'age': age,
            'salary': salary,
            'active': active
        })
    
    def get_person(self, index):
        return self.people[index]
    
    def get_memory_usage(self):
        return sys.getsizeof(self.people)
    
    def __len__(self):
        return len(self.people)

# Test data stores
def test_data_stores():
    print(f"\nData Store Memory Comparison:")
    
    # Create test data
    efficient_store = EfficientDataStore()
    traditional_store = TraditionalDataStore()
    
    # Add 1000 people
    for i in range(1000):
        name = f"Person{i}"
        age = 20 + (i % 50)
        salary = 50000 + (i * 100)
        active = i % 3 != 0
        
        efficient_store.add_person(name, age, salary, active)
        traditional_store.add_person(name, age, salary, active)
    
    efficient_memory = efficient_store.get_memory_usage()
    traditional_memory = traditional_store.get_memory_usage()
    
    print(f"Efficient store total memory: {efficient_memory['total']} bytes")
    print(f"Traditional store memory: {traditional_memory} bytes")
    
    savings = (traditional_memory - efficient_memory['total']) / traditional_memory * 100
    print(f"Memory savings: {savings:.1f}%")
    
    print(f"\nEfficient store breakdown:")
    for component, size in efficient_memory.items():
        if component != 'total':
            print(f"  {component}: {size} bytes")

test_data_stores()

Output:

Click "Run Code" to see the output

Real-World Applications #

Web Request Processing System #

Use advanced data structures for a web server simulation:

🐍 Try it yourself

from collections import defaultdict, deque, Counter, namedtuple
import heapq
import time
import random

# Web request processing system
Request = namedtuple('Request', ['id', 'client_ip', 'method', 'url', 'timestamp', 'size'])

class WebServer:
    def __init__(self, max_connections=100):
        self.max_connections = max_connections
        
        # Active connections (client_ip -> deque of recent requests)
        self.active_connections = defaultdict(lambda: deque(maxlen=10))
        
        # Request queue (priority queue: priority, timestamp, request)
        self.request_queue = []
        self.request_counter = 0
        
        # Rate limiting (client_ip -> request times)
        self.rate_limit_window = defaultdict(lambda: deque(maxlen=100))
        
        # Statistics
        self.method_stats = Counter()
        self.url_stats = Counter()
        self.client_stats = Counter()
        
        # Cache for frequently requested URLs
        self.cache = {}
        self.cache_hits = 0
        self.cache_misses = 0
    
    def calculate_priority(self, request):
        """Calculate request priority (lower = higher priority)"""
        base_priority = 5
        
        # High priority for certain methods
        if request.method == 'POST':
            base_priority -= 2
        elif request.method == 'GET':
            base_priority -= 1
        
        # High priority for API endpoints
        if '/api/' in request.url:
            base_priority -= 1
        
        # Lower priority for large requests
        if request.size > 1024 * 1024:  # > 1MB
            base_priority += 2
        
        return base_priority
    
    def is_rate_limited(self, client_ip, current_time, limit_per_minute=60):
        """Check if client exceeds rate limit"""
        client_requests = self.rate_limit_window[client_ip]
        
        # Remove requests older than 1 minute
        cutoff_time = current_time - 60
        while client_requests and client_requests[0] < cutoff_time:
            client_requests.popleft()
        
        return len(client_requests) >= limit_per_minute
    
    def add_request(self, request):
        """Add request to processing queue"""
        current_time = time.time()
        
        # Check rate limiting
        if self.is_rate_limited(request.client_ip, current_time):
            print(f"Rate limited: {request.client_ip}")
            return False
        
        # Add to rate limit tracking
        self.rate_limit_window[request.client_ip].append(current_time)
        
        # Calculate priority and add to queue
        priority = self.calculate_priority(request)
        heapq.heappush(self.request_queue, (priority, self.request_counter, request))
        self.request_counter += 1
        
        # Update connection tracking
        self.active_connections[request.client_ip].append(request)
        
        return True
    
    def process_request(self, request):
        """Process a single request"""
        # Check cache first
        cache_key = f"{request.method}:{request.url}"
        
        if cache_key in self.cache:
            self.cache_hits += 1
            return f"Cache hit for {request.url}"
        else:
            self.cache_misses += 1
            
            # Simulate processing time
            processing_time = random.uniform(0.01, 0.1)
            
            # Add to cache if GET request
            if request.method == 'GET':
                self.cache[cache_key] = {
                    'response': f"Response for {request.url}",
                    'timestamp': time.time()
                }
            
            # Update statistics
            self.method_stats[request.method] += 1
            self.url_stats[request.url] += 1
            self.client_stats[request.client_ip] += 1
            
            return f"Processed {request.method} {request.url} in {processing_time:.3f}s"
    
    def process_requests(self, max_requests=10):
        """Process requests from queue"""
        processed = 0
        
        while self.request_queue and processed < max_requests:
            priority, counter, request = heapq.heappop(self.request_queue)
            result = self.process_request(request)
            print(f"  {result}")
            processed += 1
        
        return processed
    
    def get_statistics(self):
        """Get server statistics"""
        total_requests = sum(self.method_stats.values())
        cache_hit_rate = (self.cache_hits / (self.cache_hits + self.cache_misses) * 100 
                         if (self.cache_hits + self.cache_misses) > 0 else 0)
        
        return {
            'total_requests': total_requests,
            'active_connections': len(self.active_connections),
            'queue_size': len(self.request_queue),
            'cache_hit_rate': cache_hit_rate,
            'top_methods': self.method_stats.most_common(3),
            'top_urls': self.url_stats.most_common(5),
            'top_clients': self.client_stats.most_common(3)
        }
    
    def cleanup_old_connections(self, max_age_seconds=300):
        """Remove old inactive connections"""
        current_time = time.time()
        to_remove = []
        
        for client_ip, requests in self.active_connections.items():
            if requests and (current_time - requests[-1].timestamp) > max_age_seconds:
                to_remove.append(client_ip)
        
        for client_ip in to_remove:
            del self.active_connections[client_ip]
        
        return len(to_remove)

# Test the web server
def simulate_web_traffic():
    server = WebServer()
    
    # Generate sample requests
    client_ips = ['192.168.1.1', '192.168.1.2', '192.168.1.3', '10.0.0.1', '10.0.0.2']
    methods = ['GET', 'POST', 'PUT', 'DELETE']
    urls = ['/home', '/api/users', '/api/orders', '/login', '/dashboard', 
            '/products', '/api/products', '/checkout', '/profile', '/search']
    
    print("Simulating web traffic...")
    
    # Generate and add requests
    for i in range(50):
        request = Request(
            id=i + 1,
            client_ip=random.choice(client_ips),
            method=random.choice(methods),
            url=random.choice(urls),
            timestamp=time.time(),
            size=random.randint(1024, 1024*1024)  # 1KB to 1MB
        )
        
        if server.add_request(request):
            if i % 10 == 0:
                print(f"Added request {i+1}")
    
    # Process requests in batches
    print(f"\nProcessing requests...")
    while server.request_queue:
        processed = server.process_requests(5)
        if processed == 0:
            break
    
    # Show statistics
    stats = server.get_statistics()
    
    print(f"\nServer Statistics:")
    print(f"Total requests processed: {stats['total_requests']}")
    print(f"Cache hit rate: {stats['cache_hit_rate']:.1f}%")
    print(f"Active connections: {stats['active_connections']}")
    
    print(f"\nTop HTTP methods:")
    for method, count in stats['top_methods']:
        print(f"  {method}: {count}")
    
    print(f"\nTop URLs:")
    for url, count in stats['top_urls']:
        print(f"  {url}: {count}")
    
    print(f"\nTop clients:")
    for client, count in stats['top_clients']:
        print(f"  {client}: {count}")
    
    # Cleanup demo
    removed = server.cleanup_old_connections(1)  # Very short timeout for demo
    print(f"\nCleaned up {removed} old connections")

simulate_web_traffic()

Output:

Click "Run Code" to see the output

Conclusion #

Advanced Python data structures provide:

Specialized functionality for specific use cases
Performance optimizations for different operations
Memory efficiency for large datasets
Custom implementations for unique requirements
Real-world applications in system design

Key takeaways:

Choose the right data structure for your specific needs
Consider time and space complexity trade-offs
Use the collections module for specialized requirements
Implement custom structures when built-ins aren't sufficient
Always profile and measure performance in real applications

Master these advanced data structures to write more efficient and scalable Python applications!

PyGuide

PyGuide

Advanced Python Data Structures: Collections and Custom Implementations

Table of Contents #

Collections Module Overview #

defaultdict: Dictionaries with Default Values #

🐍 Try it yourself

Counter: Frequency Counting Made Easy #

🐍 Try it yourself

deque: Double-Ended Queue #

🐍 Try it yourself

Advanced Dictionary Structures #

OrderedDict: Maintaining Insertion Order #

🐍 Try it yourself

ChainMap: Combining Multiple Mappings #

🐍 Try it yourself

Specialized Lists and Queues #

Named Tuples: Structured Data with Named Fields #

🐍 Try it yourself

Custom Data Structure Implementations #

Binary Search Tree #

🐍 Try it yourself

Priority Queue with Heap #

🐍 Try it yourself

Performance Analysis #

Time Complexity Comparison #

🐍 Try it yourself

Memory Optimization #

Space-Efficient Data Structures #

🐍 Try it yourself

Real-World Applications #

Web Request Processing System #

🐍 Try it yourself

Conclusion #