Python 3.5: async/await, Type Hints, and the @ Operator

import asyncio

# Comparison Python 3.4 vs 3.5

# Python 3.4:
# @asyncio.coroutine
# def fetch(url):
#     response = yield from asyncio.sleep(1)
#     return response

# Python 3.5: native syntax
async def fetch_page(url, session_id):
    """Simulate fetching a web page."""
    print(f"[Session {session_id}] Fetching {url}...")
    await asyncio.sleep(0.5)  # simulate network latency
    return {"url": url, "size": len(url) * 100, "status": 200}

async def process_page(url, session_id):
    """Fetch and process a page."""
    result = await fetch_page(url, session_id)
    print(f"[Session {session_id}] {url} -> {result['size']} bytes")
    return result

async def crawler(urls):
    """Crawl multiple URLs in parallel."""
    tasks = [
        process_page(url, idx)
        for idx, url in enumerate(urls, 1)
    ]
    results = await asyncio.gather(*tasks)
    total = sum(r["size"] for r in results)
    print(f"\nTotal fetched: {total} bytes from {len(results)} pages")
    return results

# Entry point
urls = [
    "https://blog.example.com/article-1",
    "https://blog.example.com/article-2",
    "https://blog.example.com/article-3",
    "https://blog.example.com/article-4",
]
asyncio.run(crawler(urls))

from typing import Optional, Union, List, Dict, Tuple

# Function with type annotations
def find_user(
    identifier: Union[int, str],
    active_only: bool = True
) -> Optional[Dict[str, Union[str, int]]]:
    """Look up a user by ID or username."""
    database = {
        1: {"name": "Dupont", "first_name": "Alice", "age": 32, "active": True},
        2: {"name": "Martin", "first_name": "Bob", "age": 28, "active": False},
    }
    for uid, info in database.items():
        if uid == identifier or info["name"] == identifier:
            if active_only and not info["active"]:
                return None
            return info
    return None

# Function that transforms data with types
def compute_statistics(values: List[float]) -> Dict[str, float]:
    """Compute basic statistics on a list of numbers."""
    if not values:
        return {"mean": 0.0, "minimum": 0.0, "maximum": 0.0}
    return {
        "mean": sum(values) / len(values),
        "minimum": min(values),
        "maximum": max(values),
    }

# Generic function with Tuple
def parse_coordinates(text: str) -> Tuple[float, float]:
    """Parse a 'lat,lon' string into a coordinate tuple."""
    parts = text.split(",")
    return float(parts[0].strip()), float(parts[1].strip())

# Usage
user = find_user(1)
print(user)  # {'name': 'Dupont', 'first_name': 'Alice', ...}

stats = compute_statistics([23.5, 19.8, 31.2, 27.1])
print(stats)  # {'mean': 25.4, 'minimum': 19.8, 'maximum': 31.2}

lat, lon = parse_coordinates("48.8566, 2.3522")
print(f"Paris: {lat}, {lon}")  # Paris: 48.8566, 2.3522

class Matrix:
    """Simple 2x2 matrix with @ operator support."""

    def __init__(self, a, b, c, d):
        # | a  b |
        # | c  d |
        self.a, self.b = a, b
        self.c, self.d = c, d

    def __matmul__(self, other):
        """Matrix multiplication with the @ operator."""
        return Matrix(
            self.a * other.a + self.b * other.c,
            self.a * other.b + self.b * other.d,
            self.c * other.a + self.d * other.c,
            self.c * other.b + self.d * other.d,
        )

    def __repr__(self):
        return f"Matrix({self.a}, {self.b}, {self.c}, {self.d})"

    def display(self):
        print(f"| {self.a:>6.2f}  {self.b:>6.2f} |")
        print(f"| {self.c:>6.2f}  {self.d:>6.2f} |")

# 90-degree rotation
import math
angle = math.radians(90)
rotation = Matrix(
    math.cos(angle), -math.sin(angle),
    math.sin(angle),  math.cos(angle),
)

# Scale x2
scale = Matrix(2, 0, 0, 2)

# Composing transformations with @
transform = rotation @ scale
print("Rotation 90\u00b0 then scale x2:")
transform.display()
# |   0.00  -2.00 |
# |   2.00   0.00 |

# Double rotation
double_rotation = rotation @ rotation
print("\nDouble rotation 90\u00b0 (= 180\u00b0):")
double_rotation.display()
# |  -1.00   0.00 |
# |   0.00  -1.00 |

import os
import time

# Comparison os.listdir vs os.scandir

def list_with_listdir(directory):
    """List files with os.listdir + os.stat (old method)."""
    results = []
    for name in os.listdir(directory):
        path = os.path.join(directory, name)
        info = os.stat(path)  # additional system call!
        if os.path.isfile(path):
            results.append((name, info.st_size))
    return results

def list_with_scandir(directory):
    """List files with os.scandir (fast method)."""
    results = []
    with os.scandir(directory) as entries:
        for entry in entries:
            if entry.is_file():  # no extra system call
                info = entry.stat()  # often already cached
                results.append((entry.name, info.st_size))
    return results

# Benchmark on a directory
directory = "/usr/lib"

start = time.perf_counter()
files_v1 = list_with_listdir(directory)
listdir_time = time.perf_counter() - start

start = time.perf_counter()
files_v2 = list_with_scandir(directory)
scandir_time = time.perf_counter() - start

print(f"os.listdir: {len(files_v1)} files in {listdir_time:.4f}s")
print(f"os.scandir: {len(files_v2)} files in {scandir_time:.4f}s")
if listdir_time > 0:
    ratio = listdir_time / scandir_time
    print(f"scandir is {ratio:.1f}x faster")

# Advanced usage: find the 5 largest files
def top_files(directory, n=5):
    """Find the n largest files in a directory."""
    files = []
    with os.scandir(directory) as entries:
        for entry in entries:
            if entry.is_file(follow_symlinks=False):
                files.append((entry.name, entry.stat().st_size))
    files.sort(key=lambda f: f[1], reverse=True)
    return files[:n]

print(f"\nTop 5 files in {directory}:")
for name, size in top_files(directory):
    print(f"  {name:40s} {size:>10,} bytes")