Python collections Modul: Counter, defaultdict, deque, namedtuple – Leitfaden

import collections
 
# See all available classes
print([name for name in dir(collections) if not name.startswith('_')])
# ['ChainMap', 'Counter', 'OrderedDict', 'UserDict', 'UserList',
#  'UserString', 'abc', 'defaultdict', 'deque', 'namedtuple']

from collections import Counter
 
# From an iterable
words = ['apple', 'banana', 'apple', 'cherry', 'banana', 'apple']
word_count = Counter(words)
print(word_count)
# Counter({'apple': 3, 'banana': 2, 'cherry': 1})
 
# From a string
letter_count = Counter('mississippi')
print(letter_count)
# Counter({'s': 4, 'i': 4, 'p': 2, 'm': 1})
 
# From a dictionary
inventory = Counter({'shirts': 25, 'pants': 15, 'hats': 10})
 
# From keyword arguments
stock = Counter(laptops=5, monitors=12)

from collections import Counter
 
text = "to be or not to be that is the question"
words = Counter(text.split())
 
# Get the 3 most common words
print(words.most_common(3))
# [('to', 2), ('be', 2), ('or', 1)]
 
# Get all elements sorted by frequency
print(words.most_common())
# [('to', 2), ('be', 2), ('or', 1), ('not', 1), ('that', 1), ('is', 1), ('the', 1), ('question', 1)]
 
# Least common: reverse the list or slice from the end
print(words.most_common()[-3:])
# [('is', 1), ('the', 1), ('question', 1)]

from collections import Counter
 
a = Counter(x=4, y=2, z=1)
b = Counter(x=1, y=3, z=5)
 
# Addition: combine counts
print(a + b)  # Counter({'z': 6, 'y': 5, 'x': 5})
 
# Subtraction: drops zero and negative results
print(a - b)  # Counter({'x': 3})
 
# Intersection (min of each)
print(a & b)  # Counter({'y': 2, 'x': 1, 'z': 1})
 
# Union (max of each)
print(a | b)  # Counter({'z': 5, 'x': 4, 'y': 3})

from collections import Counter
 
# Word frequency analysis
log_entries = [
    "ERROR: disk full",
    "WARNING: high memory",
    "ERROR: disk full",
    "ERROR: timeout",
    "WARNING: high memory",
    "ERROR: disk full",
    "INFO: backup complete",
]
error_types = Counter(entry.split(":")[0].strip() for entry in log_entries)
print(error_types)
# Counter({'ERROR': 4, 'WARNING': 2, 'INFO': 1})
 
# Find unique elements (count == 1)
data = [1, 2, 3, 2, 1, 4, 5, 4]
unique = [item for item, count in Counter(data).items() if count == 1]
print(unique)  # [3, 5]
 
# Check if one collection is a subset of another (anagram check)
def is_anagram(word1, word2):
    return Counter(word1.lower()) == Counter(word2.lower())
 
print(is_anagram("listen", "silent"))  # True
print(is_anagram("hello", "world"))    # False

from collections import defaultdict
 
# int factory: default is 0
counter = defaultdict(int)
counter['apples'] += 1
counter['oranges'] += 3
print(dict(counter))  # {'apples': 1, 'oranges': 3}
 
# list factory: default is []
groups = defaultdict(list)
pairs = [('fruit', 'apple'), ('veggie', 'carrot'), ('fruit', 'banana'), ('veggie', 'pea')]
for category, item in pairs:
    groups[category].append(item)
print(dict(groups))
# {'fruit': ['apple', 'banana'], 'veggie': ['carrot', 'pea']}
 
# set factory: default is set()
index = defaultdict(set)
words = [('file1', 'python'), ('file2', 'python'), ('file1', 'java'), ('file3', 'python')]
for filename, lang in words:
    index[lang].add(filename)
print(dict(index))
# {'python': {'file1', 'file2', 'file3'}, 'java': {'file1'}}

from collections import defaultdict
 
students = [
    ('Math', 'Alice'), ('Science', 'Bob'), ('Math', 'Charlie'),
    ('Science', 'Diana'), ('Math', 'Eve'), ('History', 'Frank'),
]
 
# Without defaultdict -- verbose and error-prone
groups_manual = {}
for subject, name in students:
    if subject not in groups_manual:
        groups_manual[subject] = []
    groups_manual[subject].append(name)
 
# With defaultdict -- clean and direct
groups = defaultdict(list)
for subject, name in students:
    groups[subject].append(name)
 
print(dict(groups))
# {'Math': ['Alice', 'Charlie', 'Eve'], 'Science': ['Bob', 'Diana'], 'History': ['Frank']}

from collections import defaultdict
 
# Two-level nested defaultdict
def nested_dict():
    return defaultdict(int)
 
sales = defaultdict(nested_dict)
sales['2025']['Q1'] = 150000
sales['2025']['Q2'] = 175000
sales['2026']['Q1'] = 200000
print(sales['2025']['Q1'])  # 150000
print(sales['2024']['Q3'])  # 0 (auto-created, no KeyError)
 
# Arbitrary depth nesting with a recursive factory
def deep_dict():
    return defaultdict(deep_dict)
 
config = deep_dict()
config['database']['primary']['host'] = 'localhost'
config['database']['primary']['port'] = 5432
config['database']['replica']['host'] = 'replica.local'
print(config['database']['primary']['host'])  # localhost

from collections import defaultdict
 
# Lambda for custom defaults
scores = defaultdict(lambda: 100)  # Every student starts with 100
scores['Alice'] -= 5
scores['Bob'] -= 10
print(scores['Charlie'])  # 100 (new student gets default)
print(dict(scores))  # {'Alice': 95, 'Bob': 90, 'Charlie': 100}
 
# Named function for complex defaults
def default_user():
    return {'role': 'viewer', 'active': True, 'login_count': 0}
 
users = defaultdict(default_user)
users['alice']['role'] = 'admin'
print(users['bob'])  # {'role': 'viewer', 'active': True, 'login_count': 0}

from collections import deque
 
d = deque([1, 2, 3, 4, 5])
 
# O(1) operations on both ends
d.append(6)         # Add to right: [1, 2, 3, 4, 5, 6]
d.appendleft(0)     # Add to left:  [0, 1, 2, 3, 4, 5, 6]
 
right = d.pop()     # Remove from right: 6
left = d.popleft()  # Remove from left:  0
print(d)  # deque([1, 2, 3, 4, 5])
 
# Extend from both sides
d.extend([6, 7])          # Right extend: [1, 2, 3, 4, 5, 6, 7]
d.extendleft([-1, 0])     # Left extend (reversed): [0, -1, 1, 2, 3, 4, 5, 6, 7]

from collections import deque
 
# Keep only the last 5 items
recent = deque(maxlen=5)
for i in range(10):
    recent.append(i)
 
print(recent)  # deque([5, 6, 7, 8, 9], maxlen=5)
 
# Sliding window average
def moving_average(iterable, window_size):
    window = deque(maxlen=window_size)
    for value in iterable:
        window.append(value)
        if len(window) == window_size:
            yield sum(window) / window_size
 
data = [10, 20, 30, 40, 50, 60, 70]
print(list(moving_average(data, 3)))
# [20.0, 30.0, 40.0, 50.0, 60.0]

from collections import deque
 
d = deque([1, 2, 3, 4, 5])
 
d.rotate(2)   # Rotate right by 2
print(d)  # deque([4, 5, 1, 2, 3])
 
d.rotate(-3)  # Rotate left by 3
print(d)  # deque([2, 3, 4, 5, 1])

from collections import deque
import time
 
# Benchmark: append/pop from left side
n = 100_000
 
# List: O(n) for each insert at position 0
start = time.perf_counter()
lst = []
for i in range(n):
    lst.insert(0, i)
list_time = time.perf_counter() - start
 
# Deque: O(1) for appendleft
start = time.perf_counter()
dq = deque()
for i in range(n):
    dq.appendleft(i)
deque_time = time.perf_counter() - start
 
print(f"List insert(0, x): {list_time:.4f}s")
print(f"Deque appendleft:  {deque_time:.4f}s")
print(f"Deque is {list_time / deque_time:.0f}x faster")
# Typical output:
# List insert(0, x): 1.2340s
# Deque appendleft:  0.0065s
# Deque is 190x faster

from collections import namedtuple
 
# Define a type
Point = namedtuple('Point', ['x', 'y'])
p = Point(3, 4)
 
# Access by name or index
print(p.x)     # 3
print(p[1])    # 4
print(p)       # Point(x=3, y=4)
 
# Alternative field definition styles
Color = namedtuple('Color', 'red green blue')        # Space-separated string
Config = namedtuple('Config', 'host, port, database')  # Comma-separated string

from collections import namedtuple
 
# Plain tuple: which index is what?
employee_tuple = ('Alice', 'Engineering', 95000, True)
print(employee_tuple[2])  # 95000 -- but what does index 2 mean?
 
# namedtuple: self-documenting
Employee = namedtuple('Employee', 'name department salary active')
employee = Employee('Alice', 'Engineering', 95000, True)
print(employee.salary)     # 95000 -- immediately clear
print(employee.department) # Engineering

from collections import namedtuple
 
Employee = namedtuple('Employee', 'name department salary')
emp = Employee('Alice', 'Engineering', 95000)
 
# _replace: create a new instance with some fields changed (immutable)
promoted = emp._replace(salary=110000)
print(promoted)  # Employee(name='Alice', department='Engineering', salary=110000)
print(emp)       # Employee(name='Alice', department='Engineering', salary=95000)  -- unchanged
 
# _asdict: convert to OrderedDict (Python 3.8+ returns regular dict)
print(emp._asdict())
# {'name': 'Alice', 'department': 'Engineering', 'salary': 95000}
 
# _fields: get field names
print(Employee._fields)  # ('name', 'department', 'salary')
 
# _make: create from an iterable
data = ['Bob', 'Marketing', 85000]
emp2 = Employee._make(data)
print(emp2)  # Employee(name='Bob', department='Marketing', salary=85000)

from collections import namedtuple
 
# defaults parameter (Python 3.6.1+)
Connection = namedtuple('Connection', 'host port timeout', defaults=[5432, 30])
conn1 = Connection('localhost')               # port=5432, timeout=30
conn2 = Connection('db.example.com', 3306)    # timeout=30
conn3 = Connection('db.example.com', 3306, 60)
 
print(conn1)  # Connection(host='localhost', port=5432, timeout=30)
print(conn2)  # Connection(host='db.example.com', port=3306, timeout=30)

from typing import NamedTuple
 
class Point(NamedTuple):
    x: float
    y: float
    label: str = "origin"
 
p = Point(3.0, 4.0, "A")
print(p.x, p.label)  # 3.0 A
 
# Still a tuple -- supports unpacking, indexing, iteration
x, y, label = p
print(f"({x}, {y})")  # (3.0, 4.0)

from collections import OrderedDict
 
# 1. Equality considers order
d1 = {'a': 1, 'b': 2}
d2 = {'b': 2, 'a': 1}
print(d1 == d2)  # True -- regular dicts ignore order in comparison
 
od1 = OrderedDict([('a', 1), ('b', 2)])
od2 = OrderedDict([('b', 2), ('a', 1)])
print(od1 == od2)  # False -- OrderedDict considers order
 
# 2. move_to_end() for reordering
od = OrderedDict([('a', 1), ('b', 2), ('c', 3)])
od.move_to_end('a')           # Move 'a' to the end
print(list(od.keys()))  # ['b', 'c', 'a']
 
od.move_to_end('c', last=False)  # Move 'c' to the beginning
print(list(od.keys()))  # ['c', 'b', 'a']

from collections import OrderedDict
 
class LRUCache:
    def __init__(self, capacity):
        self.cache = OrderedDict()
        self.capacity = capacity
 
    def get(self, key):
        if key not in self.cache:
            return -1
        self.cache.move_to_end(key)  # Mark as recently used
        return self.cache[key]
 
    def put(self, key, value):
        if key in self.cache:
            self.cache.move_to_end(key)
        self.cache[key] = value
        if len(self.cache) > self.capacity:
            self.cache.popitem(last=False)  # Remove oldest
 
cache = LRUCache(3)
cache.put('a', 1)
cache.put('b', 2)
cache.put('c', 3)
cache.get('a')       # Access 'a', moves it to end
cache.put('d', 4)    # Evicts 'b' (least recently used)
print(list(cache.cache.keys()))  # ['c', 'a', 'd']

from collections import ChainMap
 
defaults = {'theme': 'light', 'language': 'en', 'timeout': 30}
user_prefs = {'theme': 'dark'}
session = {'language': 'fr'}
 
config = ChainMap(session, user_prefs, defaults)
 
# Lookup searches session -> user_prefs -> defaults
print(config['theme'])     # 'dark'    (from user_prefs)
print(config['language'])  # 'fr'      (from session)
print(config['timeout'])   # 30        (from defaults)

from collections import ChainMap
import os
 
# Real-world config pattern: CLI args > env vars > config file > defaults
defaults = {
    'debug': False,
    'log_level': 'WARNING',
    'port': 8080,
    'host': '0.0.0.0',
}
 
config_file = {
    'log_level': 'INFO',
    'port': 9090,
}
 
env_vars = {
    k.lower(): v for k, v in os.environ.items()
    if k.lower() in defaults
}
 
cli_args = {'debug': True}  # Parsed from argparse
 
config = ChainMap(cli_args, env_vars, config_file, defaults)
print(config['debug'])      # True (from cli_args)
print(config['log_level'])  # 'INFO' (from config_file)
print(config['host'])       # '0.0.0.0' (from defaults)

from collections import ChainMap
 
# Simulating variable scoping (like nested function scopes)
global_scope = {'x': 1, 'y': 2}
local_scope = ChainMap(global_scope)
 
# Enter a new scope
inner_scope = local_scope.new_child()
inner_scope['x'] = 10  # Shadows global x
inner_scope['z'] = 30  # New local variable
 
print(inner_scope['x'])  # 10 (local)
print(inner_scope['y'])  # 2  (falls through to global)
print(inner_scope['z'])  # 30 (local)
 
# Exit scope -- original is unchanged
print(local_scope['x'])  # 1 (global still intact)

from collections import Counter, defaultdict
import time
 
data = list(range(1000)) * 1000  # 1 million items, 1000 unique
 
# Method 1: Counter
start = time.perf_counter()
c = Counter(data)
counter_time = time.perf_counter() - start
 
# Method 2: defaultdict(int)
start = time.perf_counter()
dd = defaultdict(int)
for item in data:
    dd[item] += 1
dd_time = time.perf_counter() - start
 
# Method 3: Manual dict
start = time.perf_counter()
manual = {}
for item in data:
    manual[item] = manual.get(item, 0) + 1
manual_time = time.perf_counter() - start
 
print(f"Counter:         {counter_time:.4f}s")
print(f"defaultdict(int):{dd_time:.4f}s")
print(f"dict.get():      {manual_time:.4f}s")
# Typical: Counter ~0.03s, defaultdict ~0.07s, dict.get() ~0.09s

from collections import deque
import time
 
n = 100_000
 
# Simulate a FIFO queue: append right, pop left
# List
start = time.perf_counter()
q = list(range(n))
while q:
    q.pop(0)
list_queue_time = time.perf_counter() - start
 
# Deque
start = time.perf_counter()
q = deque(range(n))
while q:
    q.popleft()
deque_queue_time = time.perf_counter() - start
 
print(f"List pop(0):     {list_queue_time:.4f}s")
print(f"Deque popleft(): {deque_queue_time:.4f}s")
print(f"Deque is {list_queue_time / deque_queue_time:.0f}x faster")
# Typical: List ~2.5s, Deque ~0.004s -> ~600x faster

from collections import Counter
from datetime import datetime
 
# Parse and analyze server logs
log_lines = [
    "2026-02-18 10:15:03 GET /api/users 200",
    "2026-02-18 10:15:04 POST /api/login 401",
    "2026-02-18 10:15:05 GET /api/users 200",
    "2026-02-18 10:15:06 GET /api/products 500",
    "2026-02-18 10:15:07 POST /api/login 200",
    "2026-02-18 10:15:08 GET /api/users 200",
    "2026-02-18 10:15:09 GET /api/products 500",
    "2026-02-18 10:15:10 POST /api/login 401",
]
 
# Count status codes
status_codes = Counter(line.split()[-1] for line in log_lines)
print("Status codes:", status_codes.most_common())
# [('200', 4), ('401', 2), ('500', 2)]
 
# Count endpoints
endpoints = Counter(line.split()[3] for line in log_lines)
print("Top endpoints:", endpoints.most_common(2))
# [('/api/users', 3), ('/api/login', 3)]
 
# Count error endpoints (status >= 400)
errors = Counter(
    line.split()[3] for line in log_lines
    if int(line.split()[-1]) >= 400
)
print("Error endpoints:", errors)
# Counter({'/api/login': 2, '/api/products': 2})

from collections import ChainMap
import json
 
# Multi-layer config system for a web application
def load_config(config_path=None, cli_overrides=None):
    # Layer 1: Hard-coded defaults
    defaults = {
        'host': '127.0.0.1',
        'port': 8000,
        'debug': False,
        'db_pool_size': 5,
        'log_level': 'WARNING',
        'cors_origins': ['http://localhost:3000'],
    }
 
    # Layer 2: Config file
    file_config = {}
    if config_path:
        with open(config_path) as f:
            file_config = json.load(f)
 
    # Layer 3: CLI overrides (highest priority)
    cli = cli_overrides or {}
 
    # ChainMap searches cli -> file_config -> defaults
    return ChainMap(cli, file_config, defaults)
 
# Usage
config = load_config(cli_overrides={'debug': True, 'port': 9000})
print(config['debug'])        # True (CLI override)
print(config['port'])         # 9000 (CLI override)
print(config['db_pool_size']) # 5    (default)
print(config['log_level'])    # WARNING (default)

from collections import deque
 
class RecentItemsTracker:
    """Track the N most recent unique items."""
 
    def __init__(self, max_items=10):
        self.items = deque(maxlen=max_items)
        self.seen = set()
 
    def add(self, item):
        if item in self.seen:
            # Move to front by removing and re-adding
            self.items.remove(item)
            self.items.append(item)
        else:
            if len(self.items) == self.items.maxlen:
                # Remove the oldest item from the set too
                oldest = self.items[0]
                self.seen.discard(oldest)
            self.items.append(item)
            self.seen.add(item)
 
    def get_recent(self):
        return list(reversed(self.items))
 
# Track recently viewed products
tracker = RecentItemsTracker(max_items=5)
for product in ['shoes', 'shirt', 'hat', 'shoes', 'jacket', 'belt', 'hat']:
    tracker.add(product)
 
print(tracker.get_recent())
# ['hat', 'belt', 'jacket', 'shoes', 'shirt']

from collections import namedtuple, Counter, defaultdict
 
# Define structured records
Transaction = namedtuple('Transaction', 'id customer product amount date')
 
transactions = [
    Transaction(1, 'Alice', 'Widget', 29.99, '2026-02-01'),
    Transaction(2, 'Bob', 'Gadget', 49.99, '2026-02-01'),
    Transaction(3, 'Alice', 'Widget', 29.99, '2026-02-03'),
    Transaction(4, 'Charlie', 'Gadget', 49.99, '2026-02-05'),
    Transaction(5, 'Alice', 'Gizmo', 19.99, '2026-02-07'),
    Transaction(6, 'Bob', 'Widget', 29.99, '2026-02-08'),
]
 
# Most popular products
product_count = Counter(t.product for t in transactions)
print("Popular products:", product_count.most_common())
# [('Widget', 3), ('Gadget', 2), ('Gizmo', 1)]
 
# Revenue by customer
revenue = defaultdict(float)
for t in transactions:
    revenue[t.customer] += t.amount
print("Revenue:", dict(revenue))
# {'Alice': 79.97, 'Bob': 79.98, 'Charlie': 49.99}
 
# Convert to DataFrame for visualization
import pandas as pd
df = pd.DataFrame(transactions, columns=Transaction._fields)
print(df.groupby('customer')['amount'].sum())

from collections import Counter
import pandas as pd
import pygwalker as pyg
 
# Process data with collections
log_data = ["ERROR", "WARNING", "ERROR", "INFO", "ERROR", "WARNING", "INFO", "INFO"]
counts = Counter(log_data)
 
# Convert to DataFrame
df = pd.DataFrame(counts.items(), columns=['Level', 'Count'])
 
# Launch interactive visualization
walker = pyg.walk(df)

from collections import Counter, defaultdict, namedtuple, deque
 
# Named record type
LogEntry = namedtuple('LogEntry', 'timestamp level message')
 
# Simulated log stream
log_stream = deque([
    LogEntry('10:01', 'ERROR', 'Connection timeout'),
    LogEntry('10:02', 'INFO', 'Request processed'),
    LogEntry('10:03', 'ERROR', 'Connection timeout'),
    LogEntry('10:04', 'WARNING', 'High memory'),
    LogEntry('10:05', 'ERROR', 'Disk full'),
    LogEntry('10:06', 'INFO', 'Request processed'),
    LogEntry('10:07', 'ERROR', 'Connection timeout'),
], maxlen=100)
 
# Count error types
error_counts = Counter(
    entry.message for entry in log_stream if entry.level == 'ERROR'
)
print("Error types:", error_counts.most_common())
# [('Connection timeout', 3), ('Disk full', 1)]
 
# Group entries by level
by_level = defaultdict(list)
for entry in log_stream:
    by_level[entry.level].append(entry)
 
for level, entries in by_level.items():
    print(f"{level}: {len(entries)} entries")
# ERROR: 4 entries
# INFO: 2 entries
# WARNING: 1 entries