DEV Community

Thiruvengadam Sakthivel
Thiruvengadam Sakthivel

Posted on

#06 – Advanced Python Features

#ai #programming #productivity #python

Welcome to Day 6! Today we are unlocking Python’s advanced feature set. We will cover writing concise expressions, managing custom streaming data loops, building clean code modifiers using decorators, managing temporary resources with context managers, and designing clean data structures using dataclasses. 🚀

1. Comprehensions 🧠

Comprehensions provide a concise syntax to derive new collections (lists, dictionaries, or sets) from existing iterables. They replace verbose for loops and execute faster because they run at C-speed inside the Python interpreter.

  • Advanced List Comprehensions: Supports filtering conditionals (if) and value transformations (if-else).
  • Dictionary Comprehensions: Dynamically generates key-value map structures ({k: v for ...}).
  • Set Comprehensions: Produces a unique collection of elements, automatically stripping duplicates ({x for ...}).

🌱 Easy Starter Example 

# Advanced List Comprehension with if-else evaluation
numbers = [1, 2, 3, 4, 5, 6]
labels = ["Even" if x % 2 == 0 else "Odd" for x in numbers]
print(labels)  # Output: ['Odd', 'Even', 'Odd', 'Even', 'Odd', 'Even']

# Dictionary & Set Comprehensions
squared_dict = {x: x**2 for x in range(1, 4)}   # Output: {1: 1, 2: 4, 3: 9}
unique_lengths = {len(word) for word in ["apple", "banana", "apple"]} # Output: {5, 6}
Enter fullscreen mode Exit fullscreen mode

🏛️ Real-World Example: HTTP API Response Sanitizer 

def sanitize_api_payload(response_data: list[dict]) -> dict[str, dict]:
    # Extract unique active user identities using a Set Comprehension
    active_users = {row["user_id"] for row in response_data if row.get("is_active")}

    # Clean, transform, and map user payload states via a Dictionary Comprehension
    sanitized_directory = {
        user.lower(): {
            "role": row["role"].upper(),
            "email_domain": row["email"].split("@")[-1],
            "access_clearance": "High" if row["clearance_level"] > 3 else "Standard"
        }
        for row in response_data
        for user in [row.get("user_id", "")]  # Inline variable binding assignment
        if user in active_users
    }
    return sanitized_directory

# Execution Verification
raw_payload = [
    {"user_id": "Alice", "is_active": True, "role": "admin", "email": "alice@company.com", "clearance_level": 5},
    {"user_id": "Bob", "is_active": False, "role": "dev", "email": "bob@company.com", "clearance_level": 2},
    {"user_id": "Charlie", "is_active": True, "role": "analyst", "email": "charlie@partner.org", "clearance_level": 1}
]
print(sanitize_api_payload(raw_payload))
Enter fullscreen mode Exit fullscreen mode

2. Iteration & Generators 🔄

The Iterable Protocol

An Iterable is any object that can return its elements one at a time (e.g., a list or string). Under the hood, the Iterable Protocol relies on two special methods:

  1. __iter__(): Called automatically by Python loops to request an iterator object.
  2. __next__(): Called on the iterator to fetch the next value. It raises a StopIteration exception when no elements are left.

Generators & yield

A Generator is a memory-efficient tool for handling massive datasets. Unlike standard functions that use return to pass back a finished list, a generator uses the yield keyword to output values one at a time. It pauses its state between calls, keeping only a single item in memory rather than storing the entire dataset.

  • Generator Functions: Defined using def but contain yield expressions.
  • Generator Expressions: High-performance, memory-saving alternatives to list comprehensions written inside parentheses (x for x in data).

🌱 Easy Starter Example 

# Generator function yielding elements lazily on-demand
def countdown(start):
    while start > 0:
        yield start
        start -= 1

counter = countdown(3)
print(next(counter))  # Output: 3
print(next(counter))  # Output: 2

# Generator expression (stores zero array data in memory)
squared_gen = (x**2 for x in range(1000000))
print(next(squared_gen))  # Output: 0
Enter fullscreen mode Exit fullscreen mode

🏛️ Real-World Example: Large Log Processing Pipeline 

from pathlib import Path

def stream_huge_log_file(file_path: str):
    """Lazily reads a large file line-by-line to prevent high memory usage."""
    target = Path(file_path)
    if not target.exists():
        return

    with open(target, mode="r", encoding="utf-8") as file:
        for line in file:
            yield line.strip()

def filter_critical_events(log_stream):
    """Filters data streams item-by-item as elements pass through."""
    for log_line in log_stream:
        if "[CRITICAL]" in log_line:
            yield log_line.upper()

# Write fake log file for test execution context
Path("server.log").write_text("line 1\n[CRITICAL] DB Timeout\nline 3\n[CRITICAL] Out of Memory")

# Execution chain: Memory footprint stays tiny regardless of log size
raw_logs = stream_huge_log_file("server.log")
critical_alerts = filter_critical_events(raw_logs)

for alert in critical_alerts:
    print(f"Alert Triggered: {alert}")
Enter fullscreen mode Exit fullscreen mode

3. Decorators 🎨

A Decorator is a structural design pattern that lets you modify or extend the behavior of a function or class without altering its original source code. Decorators wrap an outer closure layer around a target function.

  • Function Decorators: Standard layout syntax wrappers targeting single execution functions.
  • Decorators with Arguments: Advanced layout requiring a three-tier nested function configuration to accept configuration inputs.

🌱 Easy Starter Example 

def simple_decorator(func):
    def wrapper(*args, **kwargs):
        print("Executing code before function call...")
        result = func(*args, **kwargs)
        print("Executing code after function call...")
        return result
    return wrapper

@simple_decorator
def greet(name):
    print(f"Hi {name}")

greet("Alex")
Enter fullscreen mode Exit fullscreen mode

🏛️ Real-World Example: API Request Retry Wrapper with Custom Controls 

import time
from functools import wraps

def retry(attempts: int, delay: float = 1.0):
    """A decorator that retries a function call if it encounters an exception."""
    def decorator(func):
        @wraps(func)  # Preserves target function metadata names and docstrings
        def wrapper(*args, **kwargs):
            last_error = None
            for current_attempt in range(1, attempts + 1):
                try:
                    return func(*args, **kwargs)
                except Exception as error:
                    last_error = error
                    print(f"[Attempt {current_attempt}/{attempts}] Failure encountered: {error}. Retrying in {delay}s...")
                    time.sleep(delay)
            print("All retry threshold options exhausted.")
            raise last_error
        return wrapper
    return decorator

@retry(attempts=3, delay=0.5)
def unstable_network_fetch():
    raise ConnectionResetError("Remote server disconnected.")

try:
    unstable_network_fetch()
except ConnectionResetError:
    print("Caught final execution pipeline crash safely.")
Enter fullscreen mode Exit fullscreen mode

4. Context Managers 🛠️

Context managers cleanly handle resource setup and teardown tasks using the with statement. This guarantees that files close, database locks release, and network connections drop properly, even if your application crashes mid-execution.

To create a custom context manager class, implement these two protocol methods:

  1. __enter__(self): Allocates the target resource and returns it.
  2. __exit__(self, exc_type, exc_val, exc_tb): Handles cleanup tasks automatically. If it returns True, it suppresses any runtime exceptions that occurred inside the with block.

🌱 Easy Starter Example 

class CustomTimer:
    def __enter__(self):
        print("Resource reserved block.")
        return self
    def __exit__(self, exc_type, exc_val, exc_tb):
        print("Resource released block safely.")
        return False # Allow exceptions to bubble up normally

with CustomTimer():
    print("Running operations inside block.")
Enter fullscreen mode Exit fullscreen mode

🏛️ Real-World Example: Production Mock Database Transaction Manager 

class DBTransactionManager:
    def __init__(self, connection_uri: str):
        self.connection_uri = connection_uri
        self.transaction_active = False

    def __enter__(self):
        print(f"Connecting to: {self.connection_uri}")
        print("Starting SQL transaction block...")
        self.transaction_active = True
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        if exc_type is not None:
            # An error occurred inside the with block; trigger a rollback operation
            print(f"🚨 Exception caught: {exc_val}. Rolling back database modifications!")
            self.transaction_active = False
            # Return True to suppress the exception and prevent the app from crashing
            return True 

        # No errors occurred; commit the transaction changes safely
        print("✓ Transaction committed successfully. Closing database locks.")
        self.transaction_active = False
        return True

# Execution Trace A: Successful execution path
with DBTransactionManager("postgresql://localhost:5432/prod_db") as db:
    print("Writing user record info to database row indexes...")

print("---")

# Execution Trace B: Error handling path
with DBTransactionManager("postgresql://localhost:5432/prod_db") as db:
    print("Writing transaction lines...")
    raise IOError("Disk full! Cannot complete write operations.")
print("Application recovered successfully from exception state.")
Enter fullscreen mode Exit fullscreen mode

5. Dataclasses 📊

Python's @dataclass decorator automatically writes standard boilerplate methods for your classes, such as __init__(), __repr__(), and comparison helper methods.

  • Default Values: Assign fields default parameters inline easily.
  • Frozen Dataclasses (frozen=True): Makes instances completely immutable (read-only) and hashable, allowing them to be safely used as dictionary keys or set elements.
  • Comparison Support (order=True): Auto-generates sorting evaluation triggers (<, >, <=, >=) based on the class fields in the order they are defined.

🌱 Easy Starter Example 

from dataclasses import dataclass

@dataclass
class Coordinate:
    x: float
    y: float
    label: str = "Origin" # Default value allocation

pt1 = Coordinate(10.5, 20.0)
print(pt1) # Auto-generated __repr__ output: Coordinate(x=10.5, y=20.0, label='Origin')
Enter fullscreen mode Exit fullscreen mode

🏛️ Real-World Example: High-Frequency Stock Trade Order Tracker 

from dataclasses import dataclass, field
from datetime import datetime

@dataclass(frozen=True, order=True)
class StockTrade:
    # Fields are listed in order of sorting priority (timestamp evaluated first)
    execution_time: datetime = field(compare=True)
    ticker: str = field(compare=False)
    price: float = field(compare=True)
    shares: int = field(compare=False)
    order_type: str = field(default="BUY", compare=False)

# Instantiating read-only trading records
trade1 = StockTrade(datetime(2026, 7, 18, 10, 0), "AAPL", 185.50, 50)
trade2 = StockTrade(datetime(2026, 7, 18, 10, 5), "NVDA", 450.25, 20)
trade3 = StockTrade(datetime(2026, 7, 18, 9, 30), "AAPL", 184.00, 100)

# trade1.price = 190.00 # 🚨 Throws Dataclass AttributeError (Frozen Immutable Guard active)

# Sort collections seamlessly based on execution timelines and price points
trade_history = [trade1, trade2, trade3]
trade_history.sort()

for trade in trade_history:
    print(f"Time: {trade.execution_time} | Ticker: {trade.ticker} | Price: ${trade.price}")
Enter fullscreen mode Exit fullscreen mode

6. Practice Challenge: Advanced Log Analytics Processing Engine 🏋️

Let's combine comprehensions, custom context managers, decorators, generators, and dataclasses into a functional log processing application.

Step 1: Bootstrap Project Infrastructure

Run the following initialization commands in your terminal workspace:

uv init advanced_challenge && cd advanced_challenge
touch log_engine.py
Enter fullscreen mode Exit fullscreen mode

Step 2: Implement Code Base

Open log_engine.py and build out the data pipeline logic:

# log_engine.py
import time
from dataclasses import dataclass
from datetime import datetime
from functools import wraps
from pathlib import Path

# ==========================================
# 1. DATACLASS RESOURCE ARCHITECTURE
# ==========================================
@dataclass(frozen=True, order=True)
class LogMetric:
    severity: str
    timestamp: datetime
    message: str

# ==========================================
# 2. DECORATOR PERF TRACKER
# ==========================================
def monitor_runtime_performance(func):
    @wraps(func)
    def wrapper(*args, **kwargs):
        start_time = time.perf_counter()
        result = func(*args, **kwargs)
        duration = time.perf_counter() - start_time
        print(f"⚡ [Performance Metrics Summary]: Function '{func.__name__}' execution took {duration:.6f} seconds.")
        return result
    return wrapper

# ==========================================
# 3. CUSTOM CONTEXT FILE ENGINE MANAGER
# ==========================================
class ManagedLogSession:
    def __init__(self, filename: str):
        self.file_path = Path(filename)

    def __enter__(self):
        # Create a log file pre-populated with dummy data for this challenge run
        self.file_path.write_text(
            "INFO|2026-07-18 12:00:01|System initialization active.\n"
            "ERROR|2026-07-18 12:05:22|Database authentication failed timeout status.\n"
            "DEBUG|2026-07-18 12:06:00|Verbose memory collection dump completed.\n"
            "ERROR|2026-07-18 12:10:45|Network socket connection dropped abnormally.\n"
        )
        return self.file_path

    def __exit__(self, exc_type, exc_val, exc_tb):
        if self.file_path.exists():
            self.file_path.unlink() # Cleanup system footprint silently on completion
        return False

# ==========================================
# 4. STREAM GENERATOR LOGIC
# ==========================================
def generate_lazy_stream(file_path: Path):
    with open(file_path, mode="r", encoding="utf-8") as file:
        for line in file:
            if line.strip():
                yield line.strip()

# ==========================================
# 5. EXECUTION PIPELINE
# ==========================================
@monitor_runtime_performance
def process_log_pipeline(file_path: Path) -> list[LogMetric]:
    # Stream raw lines lazily
    raw_lines = generate_lazy_stream(file_path)

    # Parse items and construct LogMetric dataclass objects using list comprehensions
    parsed_metrics = [
        LogMetric(
            severity=parts[0],
            timestamp=datetime.strptime(parts[1], "%Y-%m-%d %H:%M:%S"),
            message=parts[2]
        )
        for line in raw_lines
        for parts in [line.split("|")]
        if parts[0] == "ERROR" # Comprehension filter conditional step
    ]

    return parsed_metrics

# ==========================================
# 6. RUNTIME ORCHESTRATION HARNESS
# ==========================================
if __name__ == "__main__":
    print("--- Starting Day 6 Orchestration Run ---")

    with ManagedLogSession("production_telemetry.log") as raw_log_file:
        # Process the log data through our performance-monitored generator pipeline
        error_metrics = process_log_pipeline(raw_log_file)

        print(f"\nFound {len(error_metrics)} critical system errors:")
        for metric in error_metrics:
            print(f"🚨 [{metric.severity}] - {metric.timestamp} -> {metric.message}")

    print("\n--- System cleanup completed successfully. Target log unlinked. ---")
Enter fullscreen mode Exit fullscreen mode

Step 3: Run and Verify Your Project Pipeline

Execute your advanced processing engine via your terminal space:

uv run log_engine.py
Enter fullscreen mode Exit fullscreen mode

Top comments (0)