Python Advanced - Comprehensive Guide

import asyncio
import aiohttp
from typing import List

# Basic coroutine
async def fetch_data(url: str) -> dict:
    """Async function fetches data from URL."""
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            return await response.json()

# Running coroutines
async def main():
    # Single await
    data = await fetch_data('https://api.example.com/data')

    # Concurrent execution
    urls = ['url1', 'url2', 'url3']
    results = await asyncio.gather(*[fetch_data(url) for url in urls])

    # With timeout
    try:
        data = await asyncio.wait_for(
            fetch_data('https://slow-api.com'),
            timeout=5.0
        )
    except asyncio.TimeoutError:
        print('Request timed out')

# Run event loop
if __name__ == '__main__':
    asyncio.run(main())

# Async context manager
class DatabaseConnection:
    async def __aenter__(self):
        self.conn = await create_connection()
        return self.conn

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        await self.conn.close()

async def use_database():
    async with DatabaseConnection() as conn:
        result = await conn.execute('SELECT * FROM users')

# Async iterator
class AsyncRange:
    def __init__(self, start: int, end: int):
        self.start = start
        self.end = end

    def __aiter__(self):
        self.current = self.start
        return self

    async def __anext__(self):
        if self.current >= self.end:
            raise StopAsyncIteration

        await asyncio.sleep(0.1)  # Simulate async work
        self.current += 1
        return self.current - 1

async def iterate_async():
    async for i in AsyncRange(0, 5):
        print(i)

# Task creation and management
async def task_management():
    # Create task
    task = asyncio.create_task(fetch_data('https://api.example.com'))

    # Get result
    result = await task

    # Multiple tasks
    tasks = [
        asyncio.create_task(fetch_data(url))
        for url in urls
    ]

    # Wait for all
    results = await asyncio.gather(*tasks)

    # Wait for first to complete
    done, pending = await asyncio.wait(
        tasks,
        return_when=asyncio.FIRST_COMPLETED
    )

    # Cancel pending
    for task in pending:
        task.cancel()

# Async generator
async def async_generator():
    """Yields values asynchronously."""
    for i in range(5):
        await asyncio.sleep(0.1)
        yield i

async def consume_generator():
    async for value in async_generator():
        print(value)

# Semaphore for rate limiting
async def rate_limited_fetch(urls: List[str], max_concurrent: int = 5):
    """Limit concurrent requests."""
    semaphore = asyncio.Semaphore(max_concurrent)

    async def fetch_with_semaphore(url: str):
        async with semaphore:
            return await fetch_data(url)

    return await asyncio.gather(*[
        fetch_with_semaphore(url) for url in urls
    ])

# Queue for producer-consumer pattern
async def producer_consumer():
    queue = asyncio.Queue(maxsize=10)

    async def producer():
        for i in range(20):
            await queue.put(i)
            print(f'Produced {i}')
            await asyncio.sleep(0.1)

        # Signal completion
        await queue.put(None)

    async def consumer():
        while True:
            item = await queue.get()
            if item is None:
                break

            print(f'Consumed {item}')
            await asyncio.sleep(0.2)
            queue.task_done()

    # Run producer and consumer concurrently
    await asyncio.gather(
        producer(),
        consumer()
    )

# Error handling in async
async def error_handling():
    try:
        result = await fetch_data('https://api.example.com')
    except aiohttp.ClientError as e:
        print(f'Request failed: {e}')
    except asyncio.TimeoutError:
        print('Request timed out')
    finally:
        print('Cleanup')

# Running in thread pool (for CPU-bound work)
import concurrent.futures

async def run_cpu_bound():
    loop = asyncio.get_event_loop()
    executor = concurrent.futures.ProcessPoolExecutor()

    result = await loop.run_in_executor(
        executor,
        cpu_intensive_function,
        arg1, arg2
    )

    return result

# Async with statement (Python 3.10+)
async def async_with_example():
    async with asyncio.TaskGroup() as tg:
        task1 = tg.create_task(fetch_data('url1'))
        task2 = tg.create_task(fetch_data('url2'))

    # Both tasks completed or one raised exception
    results = [task1.result(), task2.result()]

# Async retry decorator
import functools
from typing import TypeVar, Callable

T = TypeVar('T')

def async_retry(
    max_retries: int = 3,
    delay: float = 1.0,
    backoff: float = 2.0
):
    """Retry async function with exponential backoff."""
    def decorator(func: Callable[..., T]) -> Callable[..., T]:
        @functools.wraps(func)
        async def wrapper(*args, **kwargs):
            current_delay = delay

            for attempt in range(max_retries):
                try:
                    return await func(*args, **kwargs)
                except Exception as e:
                    if attempt == max_retries - 1:
                        raise

                    print(f'Attempt {attempt + 1} failed: {e}')
                    await asyncio.sleep(current_delay)
                    current_delay *= backoff

        return wrapper
    return decorator

@async_retry(max_retries=3, delay=1.0)
async def unreliable_api_call():
    async with aiohttp.ClientSession() as session:
        async with session.get('https://api.example.com') as resp:
            return await resp.json()

# Async connection pool
class AsyncConnectionPool:
    def __init__(self, max_size: int = 10):
        self.max_size = max_size
        self.pool: asyncio.Queue = asyncio.Queue(maxsize=max_size)
        self._initialized = False

    async def initialize(self):
        """Create initial connections."""
        for _ in range(self.max_size):
            conn = await self._create_connection()
            await self.pool.put(conn)
        self._initialized = True

    async def _create_connection(self):
        """Create new connection."""
        return await asyncio.sleep(0)  # Placeholder

    async def acquire(self):
        """Get connection from pool."""
        if not self._initialized:
            await self.initialize()

        return await self.pool.get()

    async def release(self, conn):
        """Return connection to pool."""
        await self.pool.put(conn)

    async def __aenter__(self):
        self.conn = await self.acquire()
        return self.conn

    async def __aexit__(self, *args):
        await self.release(self.conn)

# Usage
pool = AsyncConnectionPool(max_size=5)

async def use_pool():
    async with pool as conn:
        # Use connection
        result = await conn.execute('query')

# Async batch processing
async def batch_process(items: List[str], batch_size: int = 10):
    """Process items in batches."""
    for i in range(0, len(items), batch_size):
        batch = items[i:i + batch_size]

        # Process batch concurrently
        results = await asyncio.gather(*[
            process_item(item) for item in batch
        ])

        yield results

async def process_all_batches():
    items = list(range(100))

    async for batch_results in batch_process(items, batch_size=10):
        print(f'Processed batch: {len(batch_results)} items')

# Async caching
class AsyncCache:
    def __init__(self, ttl: float = 60.0):
        self.cache: dict = {}
        self.ttl = ttl
        self.locks: dict = {}

    async def get(self, key: str, fetch_func):
        """Get from cache or fetch."""
        # Check cache
        if key in self.cache:
            value, timestamp = self.cache[key]
            if asyncio.get_event_loop().time() - timestamp < self.ttl:
                return value

        # Acquire lock for key
        if key not in self.locks:
            self.locks[key] = asyncio.Lock()

        async with self.locks[key]:
            # Double-check after acquiring lock
            if key in self.cache:
                value, timestamp = self.cache[key]
                if asyncio.get_event_loop().time() - timestamp < self.ttl:
                    return value

            # Fetch new value
            value = await fetch_func()
            self.cache[key] = (value, asyncio.get_event_loop().time())
            return value

# Usage
cache = AsyncCache(ttl=60.0)

async def get_user(user_id: int):
    return await cache.get(
        f'user:{user_id}',
        lambda: fetch_user_from_db(user_id)
    )

# Async stream processing
async def stream_process(stream):
    """Process items from async stream."""
    async for item in stream:
        # Process item
        result = await process_item(item)

        # Yield result
        yield result

# Combine async generators
async def merge_streams(*streams):
    """Merge multiple async streams."""
    queue = asyncio.Queue()

    async def consume_stream(stream):
        async for item in stream:
            await queue.put(('item', item))
        await queue.put(('done', None))

    tasks = [
        asyncio.create_task(consume_stream(stream))
        for stream in streams
    ]

    done_count = 0
    while done_count < len(streams):
        msg_type, item = await queue.get()

        if msg_type == 'done':
            done_count += 1
        else:
            yield item

    # Wait for all tasks
    await asyncio.gather(*tasks)

# Function decorators
import functools
import time
from typing import Callable, TypeVar

T = TypeVar('T')

# Basic decorator
def timer(func: Callable[..., T]) -> Callable[..., T]:
    """Measure function execution time."""
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        start = time.time()
        result = func(*args, **kwargs)
        end = time.time()
        print(f'{func.__name__} took {end - start:.2f}s')
        return result
    return wrapper

@timer
def slow_function():
    time.sleep(1)

# Decorator with arguments
def retry(max_attempts: int = 3, delay: float = 1.0):
    """Retry function on failure."""
    def decorator(func: Callable[..., T]) -> Callable[..., T]:
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            for attempt in range(max_attempts):
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    if attempt == max_attempts - 1:
                        raise
                    print(f'Attempt {attempt + 1} failed: {e}')
                    time.sleep(delay)
        return wrapper
    return decorator

@retry(max_attempts=3, delay=2.0)
def unreliable_function():
    # Might fail
    pass

# Class decorator
def singleton(cls):
    """Make class a singleton."""
    instances = {}

    @functools.wraps(cls)
    def get_instance(*args, **kwargs):
        if cls not in instances:
            instances[cls] = cls(*args, **kwargs)
        return instances[cls]

    return get_instance

@singleton
class Database:
    def __init__(self):
        self.connection = None

# Method decorators
class property_cached:
    """Cached property decorator."""
    def __init__(self, func):
        self.func = func
        self.name = func.__name__

    def __get__(self, instance, owner):
        if instance is None:
            return self

        # Check cache
        cache_attr = f'_cached_{self.name}'
        if not hasattr(instance, cache_attr):
            # Compute and cache
            value = self.func(instance)
            setattr(instance, cache_attr, value)

        return getattr(instance, cache_attr)

class User:
    def __init__(self, user_id: int):
        self.user_id = user_id

    @property_cached
    def expensive_data(self):
        # Expensive computation
        return fetch_user_data(self.user_id)

# Stacked decorators
@timer
@retry(max_attempts=3)
def important_function():
    # Function is retried, then timed
    pass

# Descriptors
class Validated:
    """Descriptor for validated attributes."""
    def __init__(self, validator):
        self.validator = validator
        self.name = None

    def __set_name__(self, owner, name):
        self.name = name

    def __get__(self, instance, owner):
        if instance is None:
            return self
        return instance.__dict__.get(self.name)

    def __set__(self, instance, value):
        if not self.validator(value):
            raise ValueError(f'Invalid value for {self.name}: {value}')
        instance.__dict__[self.name] = value

# Usage
class Person:
    age = Validated(lambda x: isinstance(x, int) and x >= 0)
    name = Validated(lambda x: isinstance(x, str) and len(x) > 0)

    def __init__(self, name: str, age: int):
        self.name = name
        self.age = age

# Type-specific descriptors
class Integer:
    """Integer descriptor with bounds."""
    def __init__(self, min_value: int = None, max_value: int = None):
        self.min_value = min_value
        self.max_value = max_value
        self.name = None

    def __set_name__(self, owner, name):
        self.name = name

    def __get__(self, instance, owner):
        if instance is None:
            return self
        return instance.__dict__.get(self.name)

    def __set__(self, instance, value):
        if not isinstance(value, int):
            raise TypeError(f'{self.name} must be an integer')

        if self.min_value is not None and value < self.min_value:
            raise ValueError(f'{self.name} must be >= {self.min_value}')

        if self.max_value is not None and value > self.max_value:
            raise ValueError(f'{self.name} must be <= {self.max_value}')

        instance.__dict__[self.name] = value

class Product:
    price = Integer(min_value=0)
    quantity = Integer(min_value=0, max_value=1000)

# Decorator factory
def validate_args(**validators):
    """Validate function arguments."""
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            # Get function signature
            sig = inspect.signature(func)
            bound = sig.bind(*args, **kwargs)
            bound.apply_defaults()

            # Validate arguments
            for arg_name, validator in validators.items():
                if arg_name in bound.arguments:
                    value = bound.arguments[arg_name]
                    if not validator(value):
                        raise ValueError(
                            f'Invalid {arg_name}: {value}'
                        )

            return func(*args, **kwargs)
        return wrapper
    return decorator

@validate_args(
    age=lambda x: x >= 0,
    name=lambda x: len(x) > 0
)
def create_user(name: str, age: int):
    return User(name, age)

# Memoization decorator
def memoize(func):
    """Cache function results."""
    cache = {}

    @functools.wraps(func)
    def wrapper(*args):
        if args not in cache:
            cache[args] = func(*args)
        return cache[args]

    return wrapper

@memoize
def fibonacci(n: int) -> int:
    if n < 2:
        return n
    return fibonacci(n - 1) + fibonacci(n - 2)

# Context manager decorator
from contextlib import contextmanager

@contextmanager
def timer_context(name: str):
    """Time code block."""
    start = time.time()
    try:
        yield
    finally:
        end = time.time()
        print(f'{name} took {end - start:.2f}s')

# Usage
with timer_context('database query'):
    # Code to time
    query_database()

# Basic metaclass
class Meta(type):
    """Custom metaclass."""
    def __new__(mcs, name, bases, attrs):
        # Modify class before creation
        attrs['created_by'] = 'Meta'
        return super().__new__(mcs, name, bases, attrs)

class MyClass(metaclass=Meta):
    pass

print(MyClass.created_by)  # 'Meta'

# Singleton metaclass
class SingletonMeta(type):
    """Metaclass for singleton pattern."""
    _instances = {}

    def __call__(cls, *args, **kwargs):
        if cls not in cls._instances:
            cls._instances[cls] = super().__call__(*args, **kwargs)
        return cls._instances[cls]

class Database(metaclass=SingletonMeta):
    def __init__(self):
        self.connection = None

# Multiple instances return same object
db1 = Database()
db2 = Database()
assert db1 is db2

# Registry metaclass
class RegistryMeta(type):
    """Register all subclasses."""
    _registry = {}

    def __new__(mcs, name, bases, attrs):
        cls = super().__new__(mcs, name, bases, attrs)

        # Register class
        if name != 'Base':  # Don't register base class
            mcs._registry[name] = cls

        return cls

    @classmethod
    def get_registry(mcs):
        return mcs._registry.copy()

class Base(metaclass=RegistryMeta):
    pass

class Handler1(Base):
    pass

class Handler2(Base):
    pass

# Get all registered subclasses
registry = RegistryMeta.get_registry()
print(registry)  # {'Handler1': , 'Handler2': }

# Validation metaclass
class ValidatedMeta(type):
    """Validate class attributes."""
    def __new__(mcs, name, bases, attrs):
        # Check required attributes
        required = ['name', 'version']
        for attr in required:
            if attr not in attrs:
                raise TypeError(f'{name} must define {attr}')

        return super().__new__(mcs, name, bases, attrs)

class Plugin(metaclass=ValidatedMeta):
    name = 'MyPlugin'
    version = '1.0'

# ABCMeta for abstract classes
from abc import ABCMeta, abstractmethod

class Shape(metaclass=ABCMeta):
    @abstractmethod
    def area(self) -> float:
        pass

    @abstractmethod
    def perimeter(self) -> float:
        pass

class Circle(Shape):
    def __init__(self, radius: float):
        self.radius = radius

    def area(self) -> float:
        return 3.14 * self.radius ** 2

    def perimeter(self) -> float:
        return 2 * 3.14 * self.radius

# Cannot instantiate without implementing abstract methods
# shape = Shape()  # TypeError

# __init_subclass__ hook (Python 3.6+)
class PluginBase:
    """Base class with subclass initialization."""
    plugins = []

    def __init_subclass__(cls, plugin_type: str, **kwargs):
        super().__init_subclass__(**kwargs)
        cls.plugin_type = plugin_type
        cls.plugins.append(cls)

class AudioPlugin(PluginBase, plugin_type='audio'):
    pass

class VideoPlugin(PluginBase, plugin_type='video'):
    pass

print(PluginBase.plugins)  # [AudioPlugin, VideoPlugin]

# Dynamic class creation
def create_class(name: str, methods: dict):
    """Create class dynamically."""
    return type(name, (object,), methods)

# Create class
DynamicClass = create_class(
    'DynamicClass',
    {
        'greet': lambda self: print(f'Hello from {self.__class__.__name__}'),
        'value': 42
    }
)

obj = DynamicClass()
obj.greet()

# Modify class after creation
def add_method_to_class(cls, method_name: str, method):
    """Add method to existing class."""
    setattr(cls, method_name, method)

class MyClass:
    pass

def new_method(self):
    return 'New method!'

add_method_to_class(MyClass, 'new_method', new_method)

obj = MyClass()
print(obj.new_method())

# Class decorators vs metaclasses
# Class decorator
def add_repr(cls):
    """Add __repr__ to class."""
    def __repr__(self):
        return f'{cls.__name__}({self.__dict__})'

    cls.__repr__ = __repr__
    return cls

@add_repr
class Person:
    def __init__(self, name: str, age: int):
        self.name = name
        self.age = age

# Metaclass for the same purpose
class ReprMeta(type):
    def __new__(mcs, name, bases, attrs):
        def __repr__(self):
            return f'{name}({self.__dict__})'

        attrs['__repr__'] = __repr__
        return super().__new__(mcs, name, bases, attrs)

class Person2(metaclass=ReprMeta):
    def __init__(self, name: str, age: int):
        self.name = name
        self.age = age

# Attribute access control metaclass
class AttributeControlMeta(type):
    """Control attribute access."""
    def __setattr__(cls, name, value):
        if name.startswith('_protected'):
            raise AttributeError(f'Cannot modify {name}')
        super().__setattr__(name, value)

class Config(metaclass=AttributeControlMeta):
    _protected_value = 'constant'
    normal_value = 'mutable'

# Config._protected_value = 'new'  # AttributeError

# Protocol for structural subtyping
from typing import Protocol

class Drawable(Protocol):
    def draw(self) -> None:
        ...

# Any class with draw() method is considered a Drawable
class Circle:
    def draw(self) -> None:
        print('Drawing circle')

def render(shape: Drawable):
    shape.draw()

# Works without explicit inheritance
render(Circle())

# Basic generator
def count_up_to(n: int):
    """Generate numbers from 0 to n."""
    i = 0
    while i < n:
        yield i
        i += 1

for num in count_up_to(5):
    print(num)

# Generator expression
squares = (x**2 for x in range(10))

# Generator with send()
def echo_generator():
    """Echo back sent values."""
    value = None
    while True:
        value = yield value

gen = echo_generator()
next(gen)  # Prime the generator
print(gen.send('Hello'))  # 'Hello'
print(gen.send('World'))  # 'World'

# Generator pipeline
def read_lines(filename: str):
    """Read lines from file."""
    with open(filename) as f:
        for line in f:
            yield line.strip()

def filter_lines(lines, pattern: str):
    """Filter lines containing pattern."""
    for line in lines:
        if pattern in line:
            yield line

def process_lines(lines):
    """Process each line."""
    for line in lines:
        yield line.upper()

# Chain generators
lines = read_lines('data.txt')
filtered = filter_lines(lines, 'ERROR')
processed = process_lines(filtered)

for line in processed:
    print(line)

# Iterator protocol
class Range:
    """Custom range iterator."""
    def __init__(self, start: int, end: int):
        self.start = start
        self.end = end

    def __iter__(self):
        self.current = self.start
        return self

    def __next__(self):
        if self.current >= self.end:
            raise StopIteration

        value = self.current
        self.current += 1
        return value

for i in Range(0, 5):
    print(i)

# Infinite generator
def infinite_sequence():
    """Generate infinite sequence."""
    num = 0
    while True:
        yield num
        num += 1

# Use with itertools.islice
import itertools
for num in itertools.islice(infinite_sequence(), 10):
    print(num)

# Generator with cleanup
def managed_resource():
    """Generator with resource management."""
    print('Acquiring resource')
    resource = acquire_resource()

    try:
        yield resource
    finally:
        print('Releasing resource')
        resource.close()

# Usage
for resource in managed_resource():
    # Use resource
    pass

# Coroutine (generator-based)
def coroutine(func):
    """Decorator to prime coroutine."""
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        gen = func(*args, **kwargs)
        next(gen)  # Prime
        return gen
    return wrapper

@coroutine
def accumulator():
    """Accumulate sent values."""
    total = 0
    while True:
        value = yield total
        total += value

acc = accumulator()
print(acc.send(10))  # 10
print(acc.send(20))  # 30
print(acc.send(30))  # 60

# Generator delegation with yield from
def sub_generator():
    yield 1
    yield 2
    yield 3

def main_generator():
    yield from sub_generator()
    yield from [4, 5, 6]

for value in main_generator():
    print(value)  # 1, 2, 3, 4, 5, 6

# Recursive generator
def flatten(nested_list):
    """Flatten nested list recursively."""
    for item in nested_list:
        if isinstance(item, list):
            yield from flatten(item)
        else:
            yield item

nested = [1, [2, 3, [4, 5]], 6, [7, 8]]
print(list(flatten(nested)))  # [1, 2, 3, 4, 5, 6, 7, 8]

# Generator with state
class StatefulGenerator:
    """Generator with internal state."""
    def __init__(self):
        self.state = 0

    def __iter__(self):
        return self

    def __next__(self):
        if self.state >= 5:
            raise StopIteration

        value = self.state ** 2
        self.state += 1
        return value

for value in StatefulGenerator():
    print(value)

# Peekable iterator
class PeekableIterator:
    """Iterator with peek capability."""
    def __init__(self, iterable):
        self.iterator = iter(iterable)
        self._next = None
        self._has_next = True

    def __iter__(self):
        return self

    def __next__(self):
        if self._next is not None:
            value = self._next
            self._next = None
            return value

        return next(self.iterator)

    def peek(self):
        """Peek at next value without consuming."""
        if self._next is None:
            try:
                self._next = next(self.iterator)
            except StopIteration:
                self._has_next = False
                raise

        return self._next

    def has_next(self) -> bool:
        try:
            self.peek()
            return True
        except StopIteration:
            return False

# Usage
it = PeekableIterator([1, 2, 3])
print(it.peek())  # 1
print(next(it))   # 1
print(it.peek())  # 2

# Generator for chunking
def chunk(iterable, size: int):
    """Yield chunks of specified size."""
    iterator = iter(iterable)
    while True:
        chunk_data = list(itertools.islice(iterator, size))
        if not chunk_data:
            break
        yield chunk_data

for batch in chunk(range(10), 3):
    print(batch)  # [0, 1, 2], [3, 4, 5], [6, 7, 8], [9]

# Window generator
def window(iterable, size: int):
    """Sliding window over iterable."""
    iterator = iter(iterable)
    window_data = collections.deque(itertools.islice(iterator, size), maxlen=size)

    if len(window_data) == size:
        yield tuple(window_data)

    for item in iterator:
        window_data.append(item)
        yield tuple(window_data)

for w in window([1, 2, 3, 4, 5], 3):
    print(w)  # (1, 2, 3), (2, 3, 4), (3, 4, 5)

from typing import (
    List, Dict, Set, Tuple, Optional, Union, Any,
    Callable, TypeVar, Generic, Protocol, Literal,
    overload, cast, TYPE_CHECKING
)

# Basic type hints
def greet(name: str) -> str:
    return f'Hello, {name}'

# Collection types
def process_items(items: List[int]) -> Dict[str, int]:
    return {'count': len(items), 'sum': sum(items)}

# Optional types
def find_user(user_id: int) -> Optional[dict]:
    """Returns user dict or None."""
    user = database.get(user_id)
    return user if user else None

# Union types
def process_data(data: Union[str, int, float]) -> str:
    return str(data)

# Type aliases
UserId = int
UserData = Dict[str, Union[str, int]]

def get_user(user_id: UserId) -> UserData:
    return {'name': 'John', 'age': 30}

# Generic types
T = TypeVar('T')

def first(items: List[T]) -> Optional[T]:
    """Get first item from list."""
    return items[0] if items else None

# Generic class
class Stack(Generic[T]):
    def __init__(self):
        self.items: List[T] = []

    def push(self, item: T) -> None:
        self.items.append(item)

    def pop(self) -> T:
        return self.items.pop()

# Usage with specific type
stack: Stack[int] = Stack()
stack.push(1)
stack.push(2)

# Callable types
def apply_operation(
    value: int,
    operation: Callable[[int], int]
) -> int:
    return operation(value)

# Function that returns function
def multiplier(factor: int) -> Callable[[int], int]:
    def multiply(x: int) -> int:
        return x * factor
    return multiply

# Protocol for structural subtyping
class Drawable(Protocol):
    def draw(self) -> None:
        ...

def render(shape: Drawable) -> None:
    shape.draw()

# Any class with draw() satisfies protocol
class Circle:
    def draw(self) -> None:
        print('Drawing circle')

render(Circle())  # Works!

# Literal types
def get_config(env: Literal['dev', 'staging', 'prod']) -> dict:
    """Only accepts specific string values."""
    return configs[env]

# TypedDict for structured dictionaries
from typing import TypedDict

class UserDict(TypedDict):
    name: str
    age: int
    email: str

def create_user(data: UserDict) -> int:
    # data is validated as having required keys
    return save_to_db(data)

# Overload for different signatures
@overload
def process(x: int) -> int: ...

@overload
def process(x: str) -> str: ...

def process(x: Union[int, str]) -> Union[int, str]:
    if isinstance(x, int):
        return x * 2
    return x.upper()

# Type narrowing with isinstance
def handle_data(data: Union[str, int, list]):
    if isinstance(data, str):
        # Type narrowed to str
        return data.upper()
    elif isinstance(data, int):
        # Type narrowed to int
        return data * 2
    else:
        # Type narrowed to list
        return len(data)

# Generic with constraints
from typing import TypeVar

Numeric = TypeVar('Numeric', int, float)

def add(a: Numeric, b: Numeric) -> Numeric:
    return a + b

# Type guards
from typing import TypeGuard

def is_str_list(items: List[Any]) -> TypeGuard[List[str]]:
    """Check if all items are strings."""
    return all(isinstance(item, str) for item in items)

def process_strings(items: List[Any]):
    if is_str_list(items):
        # Type narrowed to List[str]
        return [s.upper() for s in items]

# Final types (cannot be subclassed)
from typing import Final

MAX_SIZE: Final = 100

class Base:
    name: Final[str] = 'Base'

# ParamSpec for preserving function signatures
from typing import ParamSpec

P = ParamSpec('P')

def log_calls(func: Callable[P, T]) -> Callable[P, T]:
    """Preserve function signature in decorator."""
    @functools.wraps(func)
    def wrapper(*args: P.args, **kwargs: P.kwargs) -> T:
        print(f'Calling {func.__name__}')
        return func(*args, **kwargs)
    return wrapper

@log_calls
def greet(name: str, age: int) -> str:
    return f'{name} is {age}'

# Type is preserved
result: str = greet('John', 30)

# Conditional imports for type checking
if TYPE_CHECKING:
    from expensive_module import ExpensiveClass

def process(obj: 'ExpensiveClass') -> None:
    # ExpensiveClass not imported at runtime
    pass

# NewType for type safety
from typing import NewType

UserId = NewType('UserId', int)
ProductId = NewType('ProductId', int)

def get_user(user_id: UserId) -> dict:
    return database.get(user_id)

# Type safety
user_id = UserId(123)
product_id = ProductId(456)

get_user(user_id)  # OK
# get_user(product_id)  # Type error!

# Self type for method chaining
from typing import Self  # Python 3.11+

class Builder:
    def with_name(self, name: str) -> Self:
        self.name = name
        return self

    def with_age(self, age: int) -> Self:
        self.age = age
        return self

# Annotated for metadata
from typing import Annotated

def validate_positive(x: int) -> bool:
    return x > 0

Age = Annotated[int, validate_positive]

def set_age(age: Age) -> None:
    # age is int with validation metadata
    pass

# Type narrowing with match (Python 3.10+)
def describe(value: Union[int, str, list]):
    match value:
        case int():
            # Type narrowed to int
            return f'Integer: {value}'
        case str():
            # Type narrowed to str
            return f'String: {value}'
        case list():
            # Type narrowed to list
            return f'List with {len(value)} items'

# Basic context manager
class FileManager:
    def __init__(self, filename: str, mode: str):
        self.filename = filename
        self.mode = mode
        self.file = None

    def __enter__(self):
        self.file = open(self.filename, self.mode)
        return self.file

    def __exit__(self, exc_type, exc_val, exc_tb):
        if self.file:
            self.file.close()

        # Return False to propagate exceptions
        return False

# Usage
with FileManager('data.txt', 'r') as f:
    content = f.read()

# Context manager with contextlib
from contextlib import contextmanager

@contextmanager
def timer(name: str):
    """Time code execution."""
    start = time.time()
    try:
        yield
    finally:
        end = time.time()
        print(f'{name} took {end - start:.2f}s')

with timer('database query'):
    # Code to time
    query_database()

# Database transaction context manager
@contextmanager
def transaction(connection):
    """Database transaction with rollback."""
    try:
        yield connection
        connection.commit()
    except Exception:
        connection.rollback()
        raise

with transaction(db.connection) as conn:
    conn.execute('INSERT INTO users ...')
    conn.execute('UPDATE accounts ...')

# Suppress specific exceptions
from contextlib import suppress

with suppress(FileNotFoundError):
    os.remove('temp.txt')

# Redirect stdout
from contextlib import redirect_stdout
import io

f = io.StringIO()
with redirect_stdout(f):
    print('This goes to StringIO')

output = f.getvalue()

# Multiple context managers
with open('input.txt') as infile, \
     open('output.txt', 'w') as outfile:
    outfile.write(infile.read())

# Or use contextlib.ExitStack
from contextlib import ExitStack

with ExitStack() as stack:
    files = [stack.enter_context(open(f)) for f in filenames]
    # All files automatically closed

# Reentrant context manager
class ReentrantLock:
    """Context manager that can be entered multiple times."""
    def __init__(self):
        self.level = 0
        self.owner = None

    def __enter__(self):
        current_thread = threading.current_thread()

        if self.owner is current_thread:
            self.level += 1
        else:
            # Wait for lock
            while self.owner is not None:
                pass
            self.owner = current_thread
            self.level = 1

        return self

    def __exit__(self, *args):
        self.level -= 1

        if self.level == 0:
            self.owner = None

# Async context manager
class AsyncResource:
    async def __aenter__(self):
        self.resource = await acquire_resource()
        return self.resource

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        await self.resource.close()
        return False

# Usage
async def use_resource():
    async with AsyncResource() as resource:
        await resource.do_work()

# Context manager for temporary changes
@contextmanager
def temporary_change(obj, attr: str, new_value):
    """Temporarily change attribute."""
    old_value = getattr(obj, attr)
    setattr(obj, attr, new_value)

    try:
        yield
    finally:
        setattr(obj, attr, old_value)

# Usage
config.debug = False
with temporary_change(config, 'debug', True):
    # debug is True here
    run_debug_code()
# debug is False again

# Chaining context managers
@contextmanager
def combined_managers(*managers):
    """Combine multiple context managers."""
    with ExitStack() as stack:
        entered = [stack.enter_context(mgr) for mgr in managers]
        yield entered

# Usage
with combined_managers(mgr1, mgr2, mgr3) as (m1, m2, m3):
    # Use all managers
    pass

import gc
import sys
import weakref
from typing import Any

# Check object size
obj = [1, 2, 3, 4, 5]
print(sys.getsizeof(obj))  # Size in bytes

# Reference counting
x = [1, 2, 3]
print(sys.getrefcount(x))  # Reference count

y = x  # Increases ref count
print(sys.getrefcount(x))

del y  # Decreases ref count
print(sys.getrefcount(x))

# Garbage collection
class Circular:
    def __init__(self):
        self.ref = self

# Create circular reference
obj = Circular()
obj = None  # Won't be immediately freed (circular ref)

# Force garbage collection
gc.collect()

# Weak references (don't increase ref count)
class Cache:
    def __init__(self):
        self._cache = weakref.WeakValueDictionary()

    def get(self, key: str) -> Any:
        return self._cache.get(key)

    def set(self, key: str, value: Any):
        self._cache[key] = value

# When value is deleted elsewhere, it's removed from cache
cache = Cache()
obj = SomeObject()
cache.set('key', obj)

del obj  # Also removed from cache

# __slots__ for memory optimization
class Point:
    __slots__ = ['x', 'y']  # No __dict__ created

    def __init__(self, x: float, y: float):
        self.x = x
        self.y = y

# Uses less memory than regular class
p = Point(1.0, 2.0)
# p.z = 3.0  # AttributeError: no dynamic attributes

# Object pooling
class ObjectPool:
    """Reuse objects instead of creating new ones."""
    def __init__(self, factory, max_size: int = 10):
        self.factory = factory
        self.max_size = max_size
        self.pool = []

    def acquire(self):
        if self.pool:
            return self.pool.pop()
        return self.factory()

    def release(self, obj):
        if len(self.pool) < self.max_size:
            # Reset object state
            obj.reset()
            self.pool.append(obj)

# Usage
class ExpensiveObject:
    def reset(self):
        # Reset state for reuse
        pass

pool = ObjectPool(ExpensiveObject, max_size=5)
obj = pool.acquire()
# Use object
pool.release(obj)

# Memory profiling with tracemalloc
import tracemalloc

tracemalloc.start()

# Code to profile
data = [i for i in range(1000000)]

# Get memory snapshot
snapshot = tracemalloc.take_snapshot()
top_stats = snapshot.statistics('lineno')

for stat in top_stats[:10]:
    print(stat)

tracemalloc.stop()

# Find memory leaks
def find_leaks():
    """Detect memory leaks."""
    gc.collect()

    # Get all objects
    objects = gc.get_objects()

    # Count by type
    type_counts = {}
    for obj in objects:
        obj_type = type(obj).__name__
        type_counts[obj_type] = type_counts.get(obj_type, 0) + 1

    # Sort by count
    sorted_types = sorted(
        type_counts.items(),
        key=lambda x: x[1],
        reverse=True
    )

    return sorted_types[:20]

# Generator for memory efficiency
def read_large_file(filename: str):
    """Read large file line by line."""
    with open(filename) as f:
        for line in f:
            yield line.strip()

# Instead of loading entire file
# lines = open('huge.txt').readlines()  # Bad!

# Iterator with __iter__
class LargeDataset:
    """Iterate without loading all data."""
    def __init__(self, filename: str):
        self.filename = filename

    def __iter__(self):
        with open(self.filename) as f:
            for line in f:
                yield process_line(line)

# Use interning for repeated strings
a = sys.intern('hello')
b = sys.intern('hello')
assert a is b  # Same object in memory

# Custom __del__ (destructor)
class Resource:
    def __init__(self, name: str):
        self.name = name
        print(f'Acquiring {name}')

    def __del__(self):
        print(f'Releasing {self.name}')

# Called when object is garbage collected
r = Resource('file')
del r  # Triggers __del__

# Context manager preferred over __del__
class BetterResource:
    def __init__(self, name: str):
        self.name = name

    def __enter__(self):
        print(f'Acquiring {self.name}')
        return self

    def __exit__(self, *args):
        print(f'Releasing {self.name}')

with BetterResource('file'):
    # Guaranteed cleanup
    pass

# Memory-efficient data structures
from collections import deque

# deque for queues (more efficient than list)
queue = deque(maxlen=100)  # Automatic size limit
queue.append(1)

# array for homogeneous data
from array import array
numbers = array('i', [1, 2, 3, 4, 5])  # More compact than list

# Lazy evaluation
class LazyProperty:
    """Computed on first access, then cached."""
    def __init__(self, func):
        self.func = func
        self.name = func.__name__

    def __get__(self, instance, owner):
        if instance is None:
            return self

        value = self.func(instance)
        setattr(instance, self.name, value)
        return value

class DataProcessor:
    @LazyProperty
    def expensive_result(self):
        # Computed only once
        return expensive_computation()

import threading
import multiprocessing
from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor
import queue

# Threading for I/O-bound tasks
def download_file(url: str) -> bytes:
    # I/O-bound operation
    return requests.get(url).content

# Thread pool
urls = ['url1', 'url2', 'url3']

with ThreadPoolExecutor(max_workers=5) as executor:
    results = list(executor.map(download_file, urls))

# Multiprocessing for CPU-bound tasks
def cpu_intensive(data: list) -> int:
    # CPU-bound operation
    return sum(i**2 for i in data)

# Process pool
data_chunks = [range(1000), range(1000, 2000), range(2000, 3000)]

with ProcessPoolExecutor(max_workers=4) as executor:
    results = list(executor.map(cpu_intensive, data_chunks))

# Thread-safe queue
work_queue = queue.Queue()

def worker():
    while True:
        item = work_queue.get()
        if item is None:
            break

        process_item(item)
        work_queue.task_done()

# Start workers
threads = []
for _ in range(5):
    t = threading.Thread(target=worker)
    t.start()
    threads.append(t)

# Add work
for item in items:
    work_queue.put(item)

# Wait for completion
work_queue.join()

# Stop workers
for _ in range(5):
    work_queue.put(None)

for t in threads:
    t.join()

# Lock for thread safety
lock = threading.Lock()
counter = 0

def increment():
    global counter
    with lock:
        counter += 1

# RLock (reentrant lock)
rlock = threading.RLock()

def recursive_function(n: int):
    with rlock:
        if n > 0:
            recursive_function(n - 1)

# Semaphore for rate limiting
semaphore = threading.Semaphore(5)  # Max 5 concurrent

def limited_task():
    with semaphore:
        # Only 5 threads can execute this at once
        perform_task()

# Event for signaling
event = threading.Event()

def wait_for_event():
    print('Waiting for event')
    event.wait()  # Block until set
    print('Event occurred')

def trigger_event():
    time.sleep(2)
    event.set()  # Wake up waiting threads

# Condition variable
condition = threading.Condition()
items = []

def consumer():
    with condition:
        while not items:
            condition.wait()  # Wait for items

        item = items.pop(0)
        process_item(item)

def producer():
    with condition:
        items.append(new_item())
        condition.notify()  # Wake up one consumer

# Barrier for synchronization
barrier = threading.Barrier(3)  # Wait for 3 threads

def synchronized_task():
    # Do independent work
    prepare()

    # Wait for all threads
    barrier.wait()

    # Continue together
    process()

# Process pool with shared memory
from multiprocessing import Value, Array

def worker(shared_value, shared_array):
    with shared_value.get_lock():
        shared_value.value += 1

    with shared_array.get_lock():
        shared_array[0] += 1

shared_num = Value('i', 0)
shared_arr = Array('i', [0] * 10)

processes = []
for _ in range(5):
    p = multiprocessing.Process(
        target=worker,
        args=(shared_num, shared_arr)
    )
    p.start()
    processes.append(p)

for p in processes:
    p.join()

# Manager for shared objects
from multiprocessing import Manager

def worker_with_manager(shared_dict, shared_list):
    shared_dict['key'] = 'value'
    shared_list.append(42)

manager = Manager()
shared_dict = manager.dict()
shared_list = manager.list()

p = multiprocessing.Process(
    target=worker_with_manager,
    args=(shared_dict, shared_list)
)
p.start()
p.join()

# Process pool async
with ProcessPoolExecutor(max_workers=4) as executor:
    # Submit tasks
    futures = [
        executor.submit(cpu_intensive, chunk)
        for chunk in data_chunks
    ]

    # Get results as they complete
    from concurrent.futures import as_completed

    for future in as_completed(futures):
        result = future.result()
        process_result(result)

# Thread-local storage
thread_local = threading.local()

def worker():
    # Each thread has its own value
    thread_local.value = threading.current_thread().name
    print(thread_local.value)

# Daemon threads
def background_task():
    while True:
        perform_maintenance()
        time.sleep(60)

daemon = threading.Thread(target=background_task, daemon=True)
daemon.start()
# Program can exit while daemon is running

# Profiling with cProfile
import cProfile
import pstats

def slow_function():
    # Function to profile
    result = 0
    for i in range(1000000):
        result += i
    return result

# Profile function
profiler = cProfile.Profile()
profiler.enable()

slow_function()

profiler.disable()

# Print stats
stats = pstats.Stats(profiler)
stats.sort_stats('cumulative')
stats.print_stats(10)

# Line profiler (requires line_profiler package)
from line_profiler import LineProfiler

lp = LineProfiler()
lp_wrapper = lp(slow_function)
lp_wrapper()
lp.print_stats()

# Timing with timeit
import timeit

# Time code snippet
time = timeit.timeit(
    'sum(range(100))',
    number=10000
)
print(f'Time: {time:.4f}s')

# Compare implementations
time1 = timeit.timeit(
    '[i for i in range(1000)]',
    number=10000
)

time2 = timeit.timeit(
    'list(range(1000))',
    number=10000
)

print(f'List comprehension: {time1:.4f}s')
print(f'list(range()): {time2:.4f}s')

# Use __slots__ to reduce memory
class RegularClass:
    def __init__(self, x, y):
        self.x = x
        self.y = y

class OptimizedClass:
    __slots__ = ['x', 'y']

    def __init__(self, x, y):
        self.x = x
        self.y = y

# OptimizedClass uses ~40% less memory

# Use local variables (faster than global)
def slow():
    for i in range(1000000):
        len([1, 2, 3])  # Lookup len each time

def fast():
    _len = len  # Cache builtin
    for i in range(1000000):
        _len([1, 2, 3])

# List comprehension vs map
# List comprehension is faster
squares = [x**2 for x in range(1000)]

# Than map
squares = list(map(lambda x: x**2, range(1000)))

# Use sets for membership testing
# Slow (O(n))
items_list = list(range(1000))
if 500 in items_list:
    pass

# Fast (O(1))
items_set = set(range(1000))
if 500 in items_set:
    pass

# Use dict.get() to avoid exceptions
# Slow
try:
    value = my_dict['key']
except KeyError:
    value = default

# Fast
value = my_dict.get('key', default)

# String concatenation
# Slow
s = ''
for i in range(1000):
    s += str(i)

# Fast
s = ''.join(str(i) for i in range(1000))

# Use itertools for efficient iteration
import itertools

# Chain multiple iterables
combined = itertools.chain(list1, list2, list3)

# Slice without creating list
for item in itertools.islice(infinite_generator(), 100):
    process(item)

# Product instead of nested loops
for x, y in itertools.product(range(10), range(10)):
    # Instead of nested for loops
    pass

# Use collections.defaultdict
from collections import defaultdict

# Instead of:
word_count = {}
for word in words:
    if word not in word_count:
        word_count[word] = 0
    word_count[word] += 1

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

# Use collections.Counter
from collections import Counter

# Fast counting
word_count = Counter(words)
most_common = word_count.most_common(10)

# NumPy for numerical operations
import numpy as np

# Slow
result = [x**2 for x in range(1000000)]

# Fast (vectorized)
arr = np.arange(1000000)
result = arr ** 2

# Caching with lru_cache
from functools import lru_cache

@lru_cache(maxsize=128)
def fibonacci(n: int) -> int:
    if n < 2:
        return n
    return fibonacci(n-1) + fibonacci(n-2)

# Much faster for recursive functions

# Use __slots__ in data classes
from dataclasses import dataclass

@dataclass
class Point:
    __slots__ = ['x', 'y']
    x: float
    y: float

# Memory profiler
from memory_profiler import profile

@profile
def memory_intensive():
    # Function to profile
    large_list = [i for i in range(1000000)]
    return sum(large_list)

# Run with: python -m memory_profiler script.py

# Optimization checklist
# 1. Profile first (don't guess)
# 2. Use appropriate data structures
# 3. Avoid premature optimization
# 4. Cache expensive operations
# 5. Use generators for large datasets
# 6. Vectorize with NumPy when possible
# 7. Use __slots__ for many small objects
# 8. Minimize global variable access
# 9. Use C extensions (Cython, numba) for hot loops
# 10. Consider algorithmic improvements first

from collections import (
    namedtuple, deque, Counter, defaultdict,
    OrderedDict, ChainMap
)
from typing import Any
import bisect
import heapq

# namedtuple for lightweight objects
Point = namedtuple('Point', ['x', 'y'])
p = Point(10, 20)
print(p.x, p.y)

# deque for efficient queue operations
queue = deque()
queue.append(1)      # Add to right
queue.appendleft(2)  # Add to left
queue.pop()          # Remove from right
queue.popleft()      # Remove from left

# deque with maxlen (circular buffer)
recent = deque(maxlen=5)
for i in range(10):
    recent.append(i)
# Only last 5 items kept

# Counter for counting
words = ['apple', 'banana', 'apple', 'cherry', 'banana']
counter = Counter(words)
print(counter)  # {'apple': 2, 'banana': 2, 'cherry': 1}
print(counter.most_common(2))  # [('apple', 2), ('banana', 2)]

# defaultdict with default factory
groups = defaultdict(list)
for item in items:
    groups[item.category].append(item)

# Auto-increment counter
counter = defaultdict(int)
for word in words:
    counter[word] += 1

# OrderedDict (maintains insertion order)
ordered = OrderedDict()
ordered['first'] = 1
ordered['second'] = 2
ordered['third'] = 3

# Move to end
ordered.move_to_end('first')

# ChainMap for multiple dicts
dict1 = {'a': 1, 'b': 2}
dict2 = {'b': 3, 'c': 4}
combined = ChainMap(dict1, dict2)
print(combined['b'])  # 2 (from dict1)

# Binary search with bisect
sorted_list = [1, 3, 5, 7, 9]
position = bisect.bisect_left(sorted_list, 6)  # 3
bisect.insort(sorted_list, 6)  # Insert maintaining order

# Heap (priority queue)
heap = []
heapq.heappush(heap, (1, 'priority 1'))
heapq.heappush(heap, (3, 'priority 3'))
heapq.heappush(heap, (2, 'priority 2'))

# Get lowest priority
priority, item = heapq.heappop(heap)  # (1, 'priority 1')

# Get n smallest/largest
numbers = [5, 3, 8, 1, 9, 2]
smallest_3 = heapq.nsmallest(3, numbers)  # [1, 2, 3]
largest_3 = heapq.nlargest(3, numbers)    # [9, 8, 5]

# Trie for prefix matching
class TrieNode:
    def __init__(self):
        self.children = {}
        self.is_end = False

class Trie:
    def __init__(self):
        self.root = TrieNode()

    def insert(self, word: str):
        node = self.root
        for char in word:
            if char not in node.children:
                node.children[char] = TrieNode()
            node = node.children[char]
        node.is_end = True

    def search(self, word: str) -> bool:
        node = self.root
        for char in word:
            if char not in node.children:
                return False
            node = node.children[char]
        return node.is_end

    def starts_with(self, prefix: str) -> bool:
        node = self.root
        for char in prefix:
            if char not in node.children:
                return False
            node = node.children[char]
        return True

# LRU Cache implementation
from collections import OrderedDict

class LRUCache:
    def __init__(self, capacity: int):
        self.cache = OrderedDict()
        self.capacity = capacity

    def get(self, key: str) -> Any:
        if key not in self.cache:
            return None

        # Move to end (most recently used)
        self.cache.move_to_end(key)
        return self.cache[key]

    def put(self, key: str, value: Any):
        if key in self.cache:
            self.cache.move_to_end(key)

        self.cache[key] = value

        if len(self.cache) > self.capacity:
            # Remove least recently used
            self.cache.popitem(last=False)

# Bloom filter for membership testing
import mmh3  # MurmurHash3

class BloomFilter:
    def __init__(self, size: int, hash_count: int):
        self.size = size
        self.hash_count = hash_count
        self.bit_array = [False] * size

    def add(self, item: str):
        for i in range(self.hash_count):
            index = mmh3.hash(item, i) % self.size
            self.bit_array[index] = True

    def contains(self, item: str) -> bool:
        for i in range(self.hash_count):
            index = mmh3.hash(item, i) % self.size
            if not self.bit_array[index]:
                return False
        return True  # Probably in set (false positives possible)

# Union-Find (Disjoint Set)
class UnionFind:
    def __init__(self, n: int):
        self.parent = list(range(n))
        self.rank = [0] * n

    def find(self, x: int) -> int:
        if self.parent[x] != x:
            self.parent[x] = self.find(self.parent[x])  # Path compression
        return self.parent[x]

    def union(self, x: int, y: int):
        px, py = self.find(x), self.find(y)

        if px == py:
            return

        # Union by rank
        if self.rank[px] < self.rank[py]:
            self.parent[px] = py
        elif self.rank[px] > self.rank[py]:
            self.parent[py] = px
        else:
            self.parent[py] = px
            self.rank[px] += 1

# Segment Tree for range queries
class SegmentTree:
    def __init__(self, arr: list):
        self.n = len(arr)
        self.tree = [0] * (4 * self.n)
        self.build(arr, 0, 0, self.n - 1)

    def build(self, arr: list, node: int, start: int, end: int):
        if start == end:
            self.tree[node] = arr[start]
        else:
            mid = (start + end) // 2
            self.build(arr, 2 * node + 1, start, mid)
            self.build(arr, 2 * node + 2, mid + 1, end)
            self.tree[node] = self.tree[2 * node + 1] + self.tree[2 * node + 2]

    def query(self, node: int, start: int, end: int, l: int, r: int) -> int:
        if r < start or end < l:
            return 0

        if l <= start and end <= r:
            return self.tree[node]

        mid = (start + end) // 2
        return (
            self.query(2 * node + 1, start, mid, l, r) +
            self.query(2 * node + 2, mid + 1, end, l, r)
        )

from functools import reduce, partial
from operator import add, mul
from typing import Callable

# map, filter, reduce
numbers = [1, 2, 3, 4, 5]

# map
squares = list(map(lambda x: x**2, numbers))

# filter
evens = list(filter(lambda x: x % 2 == 0, numbers))

# reduce
product = reduce(mul, numbers)  # 1*2*3*4*5 = 120

# Partial application
def power(base: int, exponent: int) -> int:
    return base ** exponent

# Create specialized function
square = partial(power, exponent=2)
cube = partial(power, exponent=3)

print(square(5))  # 25
print(cube(5))    # 125

# Function composition
def compose(*functions):
    """Compose functions right to left."""
    def inner(arg):
        result = arg
        for func in reversed(functions):
            result = func(result)
        return result
    return inner

# Usage
double = lambda x: x * 2
increment = lambda x: x + 1
square = lambda x: x ** 2

f = compose(square, increment, double)
print(f(3))  # ((3 * 2) + 1) ** 2 = 49

# Currying
def curry(func):
    """Convert function to curried version."""
    def curried(*args):
        if len(args) >= func.__code__.co_argcount:
            return func(*args)
        return lambda *more: curried(*(args + more))
    return curried

@curry
def add_three(a: int, b: int, c: int) -> int:
    return a + b + c

# Can call with different argument counts
print(add_three(1)(2)(3))      # 6
print(add_three(1, 2)(3))      # 6
print(add_three(1, 2, 3))      # 6

# Pipe operator simulation
class Pipe:
    def __init__(self, value):
        self.value = value

    def __or__(self, func):
        return Pipe(func(self.value))

    def __call__(self):
        return self.value

# Usage
result = (Pipe(5)
    | (lambda x: x * 2)
    | (lambda x: x + 1)
    | (lambda x: x ** 2)
)()

# Monads (Option/Maybe)
from typing import Optional, TypeVar, Generic

T = TypeVar('T')
U = TypeVar('U')

class Maybe(Generic[T]):
    """Option/Maybe monad."""
    def __init__(self, value: Optional[T]):
        self.value = value

    def bind(self, func: Callable[[T], 'Maybe[U]']) -> 'Maybe[U]':
        if self.value is None:
            return Maybe(None)
        return func(self.value)

    def map(self, func: Callable[[T], U]) -> 'Maybe[U]':
        if self.value is None:
            return Maybe(None)
        return Maybe(func(self.value))

    def get_or_else(self, default: T) -> T:
        return self.value if self.value is not None else default

# Usage
result = (Maybe(5)
    .map(lambda x: x * 2)
    .map(lambda x: x + 1)
    .get_or_else(0)
)

# Lazy evaluation
class Lazy:
    """Lazy evaluation wrapper."""
    def __init__(self, func: Callable):
        self.func = func
        self._value = None
        self._computed = False

    @property
    def value(self):
        if not self._computed:
            self._value = self.func()
            self._computed = True
        return self._value

# Usage
lazy_result = Lazy(lambda: expensive_computation())
# Not computed yet

if condition:
    print(lazy_result.value)  # Computed only if needed

# Recursion with tail call optimization (manual)
def factorial_tail(n: int, acc: int = 1) -> int:
    """Tail-recursive factorial."""
    if n <= 1:
        return acc
    return factorial_tail(n - 1, n * acc)

# Trampolining for tail call optimization
class Thunk:
    def __init__(self, func, *args):
        self.func = func
        self.args = args

def trampoline(func):
    """Execute tail-recursive function iteratively."""
    result = func()
    while isinstance(result, Thunk):
        result = result.func(*result.args)
    return result

def factorial_trampoline(n: int, acc: int = 1):
    if n <= 1:
        return acc
    return Thunk(factorial_trampoline, n - 1, n * acc)

result = trampoline(lambda: factorial_trampoline(5))

# Y Combinator (fixed-point combinator)
Y = lambda f: (lambda x: f(lambda *args: x(x)(*args)))(lambda x: f(lambda *args: x(x)(*args)))

# Usage
factorial = Y(lambda f: lambda n: 1 if n == 0 else n * f(n - 1))
print(factorial(5))  # 120

import unittest
import pytest
from unittest.mock import Mock, patch, MagicMock
from typing import Any

# pytest basic test
def test_addition():
    assert 1 + 1 == 2

def test_string():
    assert 'hello'.upper() == 'HELLO'

# pytest fixtures
@pytest.fixture
def database():
    """Setup database connection."""
    db = connect_to_database()
    yield db
    db.close()

def test_query(database):
    """Test uses fixture."""
    result = database.query('SELECT * FROM users')
    assert len(result) > 0

# pytest parametrize
@pytest.mark.parametrize('input,expected', [
    (1, 2),
    (2, 4),
    (3, 6),
])
def test_double(input, expected):
    assert input * 2 == expected

# Mocking with unittest.mock
def test_api_call():
    with patch('requests.get') as mock_get:
        mock_get.return_value.json.return_value = {'data': 'test'}

        result = fetch_data('https://api.example.com')

        assert result == {'data': 'test'}
        mock_get.assert_called_once()

# Mock object
def test_with_mock():
    mock = Mock()
    mock.method.return_value = 42

    result = mock.method('arg')

    assert result == 42
    mock.method.assert_called_with('arg')

# Spy on real object
def test_spy():
    obj = RealObject()
    obj.method = Mock(side_effect=obj.method)

    obj.method('arg')

    obj.method.assert_called_once_with('arg')

# pytest monkeypatch
def test_env_variable(monkeypatch):
    monkeypatch.setenv('API_KEY', 'test-key')

    result = get_api_key()

    assert result == 'test-key'

# Testing exceptions
def test_exception():
    with pytest.raises(ValueError):
        raise ValueError('error')

def test_exception_message():
    with pytest.raises(ValueError, match='specific error'):
        raise ValueError('specific error message')

# Async testing
@pytest.mark.asyncio
async def test_async_function():
    result = await async_fetch_data()
    assert result is not None

# Fixture scope
@pytest.fixture(scope='module')
def expensive_fixture():
    """Shared across module."""
    resource = expensive_setup()
    yield resource
    expensive_teardown(resource)

# Fixture dependencies
@pytest.fixture
def user():
    return User('[email protected]')

@pytest.fixture
def authenticated_user(user):
    user.authenticate()
    return user

def test_with_auth(authenticated_user):
    assert authenticated_user.is_authenticated

# Marking tests
@pytest.mark.slow
def test_slow_operation():
    time.sleep(2)

@pytest.mark.integration
def test_integration():
    # Integration test
    pass

# Run with: pytest -m slow

# Property-based testing with Hypothesis
from hypothesis import given, strategies as st

@given(st.integers(), st.integers())
def test_addition_commutative(a, b):
    assert a + b == b + a

@given(st.lists(st.integers()))
def test_reverse_twice(lst):
    assert list(reversed(list(reversed(lst)))) == lst

# Test coverage
# pytest --cov=mypackage tests/

# Snapshot testing
def test_snapshot(snapshot):
    data = generate_complex_data()
    snapshot.assert_match(data, 'snapshot_name')

# pyproject.toml (modern Python packaging)
"""
[build-system]
requires = ["setuptools>=61.0"]
build-backend = "setuptools.build_meta"

[project]
name = "mypackage"
version = "1.0.0"
description = "My awesome package"
authors = [
    {name = "John Doe", email = "[email protected]"}
]
dependencies = [
    "requests>=2.28.0",
    "pydantic>=2.0.0",
]

[project.optional-dependencies]
dev = [
    "pytest>=7.0.0",
    "black>=22.0.0",
    "mypy>=1.0.0",
]

[project.scripts]
mycli = "mypackage.cli:main"
"""

# setup.py (legacy, but still used)
from setuptools import setup, find_packages

setup(
    name='mypackage',
    version='1.0.0',
    packages=find_packages(),
    install_requires=[
        'requests>=2.28.0',
    ],
    extras_require={
        'dev': ['pytest>=7.0.0'],
    },
    entry_points={
        'console_scripts': [
            'mycli=mypackage.cli:main',
        ],
    },
)

# Package structure
"""
mypackage/
├── pyproject.toml
├── README.md
├── LICENSE
├── setup.py (optional)
├── src/
│   └── mypackage/
│       ├── __init__.py
│       ├── module.py
│       └── cli.py
├── tests/
│   ├── __init__.py
│   └── test_module.py
└── docs/
    └── index.md
"""

# __init__.py versioning
__version__ = '1.0.0'
__all__ = ['PublicClass', 'public_function']

# Import management
from .module import PublicClass, public_function

# Version checking
import sys

if sys.version_info < (3, 8):
    raise RuntimeError('Python 3.8+ required')

# Build and publish
"""
# Install build tools
pip install build twine

# Build distribution
python -m build

# Check package
twine check dist/*

# Upload to PyPI
twine upload dist/*

# Upload to Test PyPI
twine upload --repository testpypi dist/*
"""

# requirements.txt vs pyproject.toml
"""
requirements.txt - for applications
- Pinned versions: requests==2.28.1
- For reproducible environments

pyproject.toml - for libraries
- Version ranges: requests>=2.28.0
- Allows flexibility for users
"""

# Development workflow
"""
# Create virtual environment
python -m venv venv
source venv/bin/activate  # or venv\Scripts\activate on Windows

# Install in editable mode
pip install -e .

# Install with dev dependencies
pip install -e ".[dev]"
"""

# Versioning (Semantic Versioning)
"""
MAJOR.MINOR.PATCH
1.2.3

MAJOR: Breaking changes
MINOR: New features (backward compatible)
PATCH: Bug fixes
"""

# Manifest file (MANIFEST.in)
"""
include README.md
include LICENSE
include requirements.txt
recursive-include src/mypackage *.py
recursive-exclude tests *
"""

# Type stubs (_stub files)
# mypackage-stubs/__init__.pyi
"""
def public_function(arg: str) -> int: ...

class PublicClass:
    def method(self) -> None: ...
"""