Python 3.4: asyncio, enum, pathlib, and Bundled pip

import asyncio

# Coroutine that simulates fetching data
@asyncio.coroutine
def fetch_data(source, delay):
    """Simulate a network request with a variable delay."""
    print(f"Starting request to {source}...")
    yield from asyncio.sleep(delay)
    data = {"source": source, "results": delay * 100}
    print(f"Response received from {source}.")
    return data

# Coroutine that orchestrates multiple requests in parallel
@asyncio.coroutine
def aggregate_results():
    """Launch multiple requests simultaneously and aggregate results."""
    tasks = [
        fetch_data("users_api", 2),
        fetch_data("products_api", 1),
        fetch_data("orders_api", 3),
    ]
    # gather launches all tasks in parallel
    results = yield from asyncio.gather(*tasks)
    total = sum(r["results"] for r in results)
    print(f"Aggregated total: {total}")
    return results

# Running the event loop
loop = asyncio.get_event_loop()
results = loop.run_until_complete(aggregate_results())
loop.close()

# All 3 requests execute in ~3s instead of 6s sequentially
for r in results:
    print(f"  {r['source']} -> {r['results']} records")

from enum import Enum, IntEnum

# State machine for an order system
class OrderStatus(Enum):
    DRAFT = "draft"
    CONFIRMED = "confirmed"
    PREPARING = "preparing"
    SHIPPED = "shipped"
    DELIVERED = "delivered"
    CANCELLED = "cancelled"

# Allowed transitions between states
TRANSITIONS = {
    OrderStatus.DRAFT: [OrderStatus.CONFIRMED, OrderStatus.CANCELLED],
    OrderStatus.CONFIRMED: [OrderStatus.PREPARING, OrderStatus.CANCELLED],
    OrderStatus.PREPARING: [OrderStatus.SHIPPED],
    OrderStatus.SHIPPED: [OrderStatus.DELIVERED],
    OrderStatus.DELIVERED: [],
    OrderStatus.CANCELLED: [],
}

class Order:
    def __init__(self, number):
        self.number = number
        self.status = OrderStatus.DRAFT

    def transition(self, new_status):
        """Change the order status if the transition is valid."""
        if new_status in TRANSITIONS[self.status]:
            old = self.status
            self.status = new_status
            print(f"Order {self.number}: {old.value} -> {new_status.value}")
        else:
            print(f"Forbidden transition: {self.status.value} -> {new_status.value}")

order = Order("ORD-2014-001")
order.transition(OrderStatus.CONFIRMED)   # OK
order.transition(OrderStatus.PREPARING)   # OK
order.transition(OrderStatus.CANCELLED)   # Forbidden transition

# IntEnum for HTTP status codes
class HTTPCode(IntEnum):
    OK = 200
    CREATED = 201
    NOT_FOUND = 404
    SERVER_ERROR = 500

# Can be used directly as an integer
print(HTTPCode.NOT_FOUND == 404)  # True
print(HTTPCode.OK + 1)            # 201

from pathlib import Path

# Comparison with os.path
import os.path

# Old way (os.path)
old_path = os.path.join(os.path.expanduser("~"), "projects", "my_app")
exists = os.path.isdir(old_path)

# New way (pathlib)
path = Path.home() / "projects" / "my_app"
exists = path.is_dir()

# Navigating a project tree
root = Path("/var/www/my_site")
config = root / "config" / "settings.toml"
logs = root / "logs"

print(config.name)      # settings.toml
print(config.stem)      # settings
print(config.suffix)    # .toml
print(config.parent)    # /var/www/my_site/config

# List all Python files in a project
project = Path(".")  # current directory
py_files = sorted(project.rglob("*.py"))
for f in py_files[:5]:
    size = f.stat().st_size
    print(f"  {f} ({size} bytes)")

# Simplified reading and writing
output_file = Path("result.txt")
output_file.write_text("Processing completed successfully.\n", encoding="utf-8")
content = output_file.read_text(encoding="utf-8")
print(content)  # Processing completed successfully.

# Creating nested directories
folder = Path("output") / "reports" / "2024"
folder.mkdir(parents=True, exist_ok=True)

import statistics

# Analyzing API response times (in milliseconds)
response_times = [45, 52, 48, 120, 47, 51, 49, 200, 46, 53, 50, 48]

print(f"Mean        : {statistics.mean(response_times):.1f} ms")
print(f"Median      : {statistics.median(response_times):.1f} ms")
print(f"Std dev     : {statistics.stdev(response_times):.1f} ms")
print(f"Variance    : {statistics.variance(response_times):.1f}")

# The median is more resistant to outliers than the mean
# (120 ms and 200 ms pull the mean upward)

# Analyzing the grade distribution of an exam
grades = [8, 10, 10, 11, 12, 12, 12, 13, 14, 14, 15, 16, 18]

print(f"\nExam grades:")
print(f"Mean        : {statistics.mean(grades):.1f}")
print(f"Median      : {statistics.median(grades):.1f}")
print(f"Mode        : {statistics.mode(grades)}")
print(f"Low median  : {statistics.median_low(grades)}")
print(f"High median : {statistics.median_high(grades)}")

# Verify that pip is available
# python -m ensurepip --upgrade

# Standard pip usage
# python -m pip install requests
# python -m pip install --upgrade pip
# python -m pip list

# Deployment script that ensures pip is present
import subprocess
import sys

def ensure_pip():
    """Check and install pip if necessary."""
    try:
        import pip
        print(f"pip {pip.__version__} is already installed.")
    except ImportError:
        print("pip missing, installing via ensurepip...")
        import ensurepip
        ensurepip.bootstrap(upgrade=True)
        print("pip installed successfully.")

def install_dependencies(packages):
    """Install a list of packages via pip."""
    for package in packages:
        subprocess.check_call([
            sys.executable, "-m", "pip", "install",
            "--quiet", package
        ])
        print(f"  {package} installed.")

ensure_pip()
# install_dependencies(["requests", "click", "rich"])