The `concurrent.futures` module provides a high-level interface for asynchronously executing callables. It simplifies running tasks in separate threads (`ThreadPoolExecutor`) or processes (`ProcessPoolExecutor`), making it easier to parallelize I/O-bound or CPU-bound operations without managing threads or processes directly.
import concurrent.futures
import time
import requests # Needs 'pip install requests'
URLS = [
'https://httpbin.org/delay/2', # Simulates 2 second delay
'https://httpbin.org/delay/1', # Simulates 1 second delay
'https://httpbin.org/delay/3', # Simulates 3 second delay
'https://httpbin.org/delay/1', # Simulates 1 second delay
]
def fetch_url(url):
"""Fetches a URL and returns its status code and content length."""
start = time.time()
try:
response = requests.get(url, timeout=5)
duration = time.time() - start
print(f"Fetched {url} in {duration:.2f}s - Status: {response.status_code}")
return url, response.status_code, len(response.content)
except requests.exceptions.RequestException as e:
duration = time.time() - start
print(f"Failed {url} in {duration:.2f}s - Error: {e}")
return url, "Error", e
# Use ThreadPoolExecutor for I/O-bound tasks like network requests
start_total = time.time()
with concurrent.futures.ThreadPoolExecutor(max_workers=4) as executor:
# Use executor.map to apply fetch_url to each URL concurrently
# Results will be in the order the tasks were submitted
results = list(executor.map(fetch_url, URLS))
print("\n--- Results ---")
for url, status, data in results:
if status == "Error":
print(f"{url} -> Error: {data}")
else:
print(f"{url} -> Status: {status}, Length: {data}")
end_total = time.time()
# Total time should be close to the longest single request (3s) + overhead
print(f"\nTotal time using ThreadPoolExecutor: {end_total - start_total:.2f}s")
The `functools` module provides higher-order functions and operations on callable objects. Key tools include `partial` for creating specialized versions of functions with pre-filled arguments, and `lru_cache` for memoizing function calls (caching results) to speed up expensive computations with repeated inputs.
import functools
import time
# --- Using functools.partial ---
def power(base, exponent):
"""Calculates base raised to the power of exponent."""
print(f"Calculating {base}^{exponent}")
return base ** exponent
# Create a specialized function 'square' where exponent is always 2
square = functools.partial(power, exponent=2)
# Create a specialized function 'cube' where exponent is always 3
cube = functools.partial(power, exponent=3)
print("Using partial functions:")
print(f"Square of 5: {square(5)}") # Only need to provide 'base'
print(f"Cube of 4: {cube(4)}") # Only need to provide 'base'
print("-" * 20)
# --- Using functools.lru_cache ---
@functools.lru_cache(maxsize=128) # Cache up to 128 most recent unique calls
def expensive_computation(n):
"""Simulates an expensive computation, like Fibonacci."""
print(f"Computing for n={n}...")
time.sleep(0.5) # Simulate work
if n < 2:
return n
return expensive_computation(n-1) + expensive_computation(n-2)
print("Using lru_cache:")
start = time.time()
result1 = expensive_computation(10)
duration1 = time.time() - start
print(f"First call for 10 took {duration1:.2f}s, result: {result1}")
start = time.time()
result2 = expensive_computation(10) # Should be much faster due to cache
duration2 = time.time() - start
print(f"Second call for 10 took {duration2:.2f}s, result: {result2}")
# Check cache info
print(f"Cache info: {expensive_computation.cache_info()}")
# Clear the cache if needed
# expensive_computation.cache_clear()
Beyond basic `try...except`, Python allows creating custom exception classes for more specific error signaling. Exception chaining (`raise NewException from original_exception`) preserves the original error context, aiding debugging. The `try...except...else...finally` structure provides complete control over execution flow during potential errors.
# --- Custom Exception Class ---
class DataProcessingError(Exception):
"""Custom exception for errors during data processing."""
def __init__(self, message, source_data):
super().__init__(message)
self.source_data = source_data
self.message = message
def __str__(self):
return f"{self.__class__.__name__}: {self.message} (Source: {self.source_data})"
# --- Function that might raise errors ---
def process_data(data):
"""Processes data, potentially raising errors."""
print(f"\nProcessing: {data}")
try:
if not isinstance(data, dict):
# Raise a standard TypeError first
raise TypeError("Input must be a dictionary.")
required_key = 'value'
if required_key not in data:
# Raise our custom error
raise DataProcessingError(f"Missing required key: '{required_key}'", data)
result = int(data[required_key]) * 2
# This return happens only if no exceptions above occurred
return result
except ValueError as ve:
# Catch specific standard error (e.g., int() conversion fails)
# Chain the original ValueError to our custom error
raise DataProcessingError(f"Invalid value for key '{required_key}'", data) from ve
# Note: The initial TypeError is not caught here, it will propagate outwards.
# --- Using try...except...else...finally ---
inputs = [
{'value': '10'},
{'value': 'abc'},
{'other': 20},
"not a dict",
{'value': 5}
]
for item in inputs:
resource_acquired = False
try:
print("Attempting to acquire resource...")
resource_acquired = True # Simulate acquiring a resource
print("Resource acquired.")
processed_result = process_data(item)
except DataProcessingError as dpe:
print(f"Caught expected error: {dpe}")
# Access original exception if chained
if dpe.__cause__:
print(f" Caused by: {type(dpe.__cause__).__name__}: {dpe.__cause__}")
except TypeError as te:
print(f"Caught standard error: {te}")
except Exception as e:
# Catch any other unexpected exceptions
print(f"Caught unexpected error: {type(e).__name__}: {e}")
else:
# This block runs ONLY if NO exceptions occurred in the 'try' block
print(f"Processing successful! Result: {processed_result}")
finally:
# This block ALWAYS runs, whether an exception occurred or not
if resource_acquired:
print("Releasing resource.")
else:
print("No resource to release.")
print("-" * 30)
The `ctypes` module allows calling functions in shared libraries (like C `.dll` or `.so` files) directly from Python. It enables Python code to leverage existing C codebases or access low-level system functions without writing C extension modules. You define C data types and function prototypes in Python.
import ctypes
import os
import platform
# --- Example: Calling standard C library functions ---
# Load the C standard library (libc) based on the operating system
if platform.system() == "Windows":
libc = ctypes.CDLL('msvcrt') # Microsoft Visual C Runtime
elif platform.system() == "Darwin": # macOS
libc = ctypes.CDLL('libc.dylib')
else: # Linux and other Unix-like
libc = ctypes.CDLL('libc.so.6')
# --- Calling printf ---
# Define the argument types and return type for printf
# printf returns an int (number of characters printed)
# It takes a const char* (format string) and variable arguments (...)
# We'll use a simple case: printf(const char*)
libc.printf.restype = ctypes.c_int
libc.printf.argtypes = [ctypes.c_char_p]
# Call printf from Python
message = b"Hello from C's printf via ctypes!\n" # Note: Must be bytes
libc.printf(message)
# --- Calling time ---
# time_t time(time_t *tloc); -> returns current calendar time
# We call it as time(NULL) to just get the return value
libc.time.restype = ctypes.c_long # time_t is often a long integer
libc.time.argtypes = [ctypes.c_void_p] # Argument is time_t*, use void* for NULL
current_c_time = libc.time(None)
print(f"\nCurrent time from C's time(): {current_c_time}")
# --- Using C data structures ---
# Example: Using struct utsname from sys/utsname.h (Unix-like systems)
if platform.system() != "Windows":
# Define the structure in Python based on its C definition
# struct utsname {
# char sysname[]; /* Operating system name (e.g., "Linux") */
# char nodename[]; /* Name within "some implementation-defined network" */
# char release[]; /* Operating system release (e.g., "2.6.28") */
# char version[]; /* Operating system version */
# char machine[]; /* Hardware identifier */
# };
# The actual size might vary, common sizes used here
class Utsname(ctypes.Structure):
_fields_ = [
("sysname", ctypes.c_char * 65),
("nodename", ctypes.c_char * 65),
("release", ctypes.c_char * 65),
("version", ctypes.c_char * 65),
("machine", ctypes.c_char * 65)
]
# Define the uname function prototype: int uname(struct utsname *buf);
libc.uname.restype = ctypes.c_int
libc.uname.argtypes = [ctypes.POINTER(Utsname)]
# Create an instance of the structure
uname_buffer = Utsname()
# Call uname, passing a pointer to the buffer
result = libc.uname(ctypes.byref(uname_buffer))
if result == 0:
print("\nSystem information from C's uname():")
# Decode bytes to strings for printing
print(f" OS Name: {uname_buffer.sysname.decode()}")
print(f" Node Name: {uname_buffer.nodename.decode()}")
print(f" Release: {uname_buffer.release.decode()}")
# Version might contain null bytes, clean it
version_str = uname_buffer.version.split(b'\x00', 1)[0].decode()
print(f" Version: {version_str}")
print(f" Machine: {uname_buffer.machine.decode()}")
else:
print("\nFailed to call uname().")
else:
print("\nuname() example skipped on Windows.")
`cProfile` is Python's built-in profiler for gathering performance statistics. It measures how much time is spent in different functions during your program's execution. Analyzing this data helps identify bottlenecks and optimize critical sections of code. The results can be viewed directly or analyzed further using the `pstats` module.
import cProfile
import pstats
import io # To capture output in memory
import time
# --- Functions to profile ---
def slow_function(n):
"""A function that takes some time."""
time.sleep(0.1)
result = 0
for i in range(n):
result += i*i
return result
def helper_function(x):
"""A helper called by another function."""
time.sleep(0.05)
return x * x
def main_task():
"""The main task coordinating calls."""
print("Starting main task...")
total = 0
for i in range(3):
total += slow_function(10000)
for j in range(5):
total += helper_function(j)
print(f"Main task finished. Total: {total}")
# --- Method 1: Running cProfile directly on a function call ---
print("--- Profiling using cProfile.run() ---")
profiler = cProfile.Profile()
profiler.enable() # Start profiling
# Run the code to be profiled
main_task()
profiler.disable() # Stop profiling
print("Profiling complete.")
# --- Analyzing the results using pstats ---
# Create a stream to capture stats output
s = io.StringIO()
# Sort stats by cumulative time spent in the function and its sub-functions
ps = pstats.Stats(profiler, stream=s).sort_stats('cumulative')
print("\n--- Profiling Stats (sorted by cumulative time) ---")
ps.print_stats() # Print all stats to the stream
print(s.getvalue()) # Print the captured stats output
# Print only the top 5 functions by cumulative time
s = io.StringIO()
ps = pstats.Stats(profiler, stream=s).sort_stats('cumulative')
print("\n--- Top 5 Functions by Cumulative Time ---")
ps.print_stats(5)
print(s.getvalue())
# Print stats sorted by internal time (time spent in the function itself)
s = io.StringIO()
ps = pstats.Stats(profiler, stream=s).sort_stats('time') # 'time' or 'tottime'
print("\n--- Top 5 Functions by Internal Time ---")
ps.print_stats(5)
print(s.getvalue())
# --- Method 2: Saving profile data to a file ---
profile_file = "profile_output.prof"
print(f"\n--- Profiling and saving to {profile_file} ---")
cProfile.run('main_task()', profile_file)
print(f"Profiling data saved to {profile_file}.")
# You can later load and analyze this file:
# loaded_profiler = pstats.Stats(profile_file)
# loaded_profiler.sort_stats('cumulative').print_stats(10)
```
The `dataclasses` module provides a decorator and functions for automatically adding special methods like `__init__()`, `__repr__()`, `__eq__()`, and more to user-defined classes. This significantly reduces boilerplate code when creating classes that primarily store data.
from dataclasses import dataclass, field, FrozenInstanceError
from typing import List, Optional
# --- Basic Dataclass ---
@dataclass
class Point:
"""Represents a point in 2D space."""
x: float
y: float
# Automatically generated: __init__, __repr__, __eq__
p1 = Point(1.0, 2.5)
p2 = Point(1.0, 2.5)
p3 = Point(3.0, 4.0)
print("--- Basic Dataclass ---")
print(f"p1: {p1}") # Uses the auto-generated __repr__
print(f"p1 == p2: {p1 == p2}") # Uses the auto-generated __eq__
print(f"p1 == p3: {p1 == p3}")
# --- Dataclass with Default Values and Field Customization ---
@dataclass(order=True) # Add order=True to generate __lt__, __le__, __gt__, __ge__
class InventoryItem:
"""Represents an item in inventory."""
name: str
unit_price: float
# Use field() for default values or other customizations
quantity_on_hand: int = field(default=0)
# Example: A list field needs a default_factory to avoid sharing the same list instance
tags: List[str] = field(default_factory=list)
# Private attribute intended for internal use (repr=False hides it from default repr)
_internal_id: Optional[str] = field(default=None, repr=False)
# You can still define custom methods
def total_cost(self) -> float:
return self.unit_price * self.quantity_on_hand
item1 = InventoryItem("Laptop", 1200.50, tags=["electronics", "computer"])
item2 = InventoryItem("Mouse", 25.00, quantity_on_hand=10)
item3 = InventoryItem("Keyboard", 75.00, quantity_on_hand=5, tags=["accessories"])
item4 = InventoryItem("Laptop", 1100.00) # Different price
print("\n--- Dataclass with Defaults and Order ---")
print(item1)
print(item2)
print(f"Item 1 Total Cost: ${item1.total_cost():.2f}") # quantity defaults to 0
print(f"Item 2 Total Cost: ${item2.total_cost():.2f}")
# Comparison using order=True (compares fields sequentially: name, unit_price, etc.)
print(f"item2 < item3: {item2 < item3}") # Mouse < Keyboard (name)
print(f"item1 > item4: {item1 > item4}") # 1200.50 > 1100.00 (unit_price, as name is equal)
# --- Frozen Dataclasses (Immutable) ---
@dataclass(frozen=True)
class ImmutableConfig:
"""An immutable configuration object."""
setting_a: str
setting_b: int
config = ImmutableConfig("value1", 123)
print("\n--- Frozen Dataclass ---")
print(config)
try:
config.setting_a = "new_value" # This will raise an error
except FrozenInstanceError as e:
print(f"Error modifying frozen instance: {e}")
# Frozen instances are hashable if their fields are hashable, useful for dict keys or set members
config_set = {config, ImmutableConfig("value2", 456)}
print(f"Set of configs: {config_set}")
```
Unlock a wealth of tutorials and software downloads available across our platform.