Python 3.8: The Walrus Operator, Positional-Only Parameters, and f-string =

# While loop with assignment in the condition
import re

# Before Python 3.8: temporary variable before the loop
# line = input("Command: ")
# while line != "quit":
#     process(line)
#     line = input("Command: ")

# With the walrus operator: more concise
# while (line := input("Command: ")) != "quit":
#     process(line)

# List comprehension with filtering
# Before: the computation is performed twice
raw_data = ["42", "abc", "17", "", "99", "xyz", "8"]

def validate(value):
    """Return the integer if valid, None otherwise."""
    try:
        return int(value)
    except ValueError:
        return None

# With := we avoid calling validate() twice
valid_items = [v for x in raw_data if (v := validate(x)) is not None]
print(valid_items)  # [42, 17, 99, 8]

# Practical example: log file parsing
log_lines = [
    "2024-01-15 INFO  Server started on port 8080",
    "2024-01-15 DEBUG Health check OK",
    "2024-01-15 ERROR Database connection refused",
    "2024-01-15 WARN  Memory usage at 85%",
    "2024-01-15 ERROR Timeout on request /api/users",
]

error_pattern = re.compile(r"(\d{4}-\d{2}-\d{2})\s+ERROR\s+(.+)")

errors = [
    (m.group(1), m.group(2))
    for line in log_lines
    if (m := error_pattern.match(line))
]
print(errors)
# [('2024-01-15', 'Database connection refused'),
#  ('2024-01-15', 'Timeout on request /api/users')]

# Reading a file in chunks
# with open("large_file.bin", "rb") as f:
#     while (chunk := f.read(8192)):
#         process_chunk(chunk)

# The / separates positional-only parameters from others
def power(base, exponent, /, *, modulo=None):
    """Compute base ** exponent, like the built-in pow()."""
    result = base ** exponent
    if modulo is not None:
        result %= modulo
    return result

# Works
print(power(2, 10))              # 1024
print(power(2, 10, modulo=100))   # 24

# Does NOT work (positional-only parameters)
# power(base=2, exponent=10)  # TypeError!

# Practical API design
def search(query, /, *, limit=10, sort="relevance",
           filters=None):
    """Search with a free-form positional first parameter."""
    print(f"Searching for '{query}' (limit={limit}, sort={sort})")
    if filters:
        print(f"  Filters: {filters}")

# The user cannot write query=...
# So we can rename 'query' to 'term' without breaking the API
search("python asyncio", limit=5, sort="date")
# Searching for 'python asyncio' (limit=5, sort=date)

# Combining all parameter types
def full_format(positional_only, /, normal, *, keyword_only):
    """Demonstrates the three parameter categories."""
    print(f"{positional_only=}, {normal=}, {keyword_only=}")

full_format(1, 2, keyword_only=3)          # OK
full_format(1, normal=2, keyword_only=3)    # OK
# full_format(positional_only=1, normal=2, keyword_only=3)  # TypeError!

# Self-documenting expressions
x = 42
y = 3.14
name = "Python"
print(f"{x=}")     # x=42
print(f"{y=}")     # y=3.14
print(f"{name=}")  # name='Python'

# Works with complex expressions
items = ["apple", "banana", "cherry", "date"]
print(f"{len(items)=}")           # len(items)=4
print(f"{items[0].upper()=}")     # items[0].upper()='APPLE'
print(f"{sum(range(10))=}")       # sum(range(10))=45

# Compatible with formatting
import math
print(f"{math.pi=:.4f}")   # math.pi=3.1416
print(f"{1000000=:_}")      # 1000000=1_000_000

# Practical debugging workflow
def calculate_discount(price, quantity, promo_code=None):
    """Calculate a discount with built-in debugging."""
    subtotal = price * quantity
    discount = 0.0

    if quantity >= 10:
        discount += 0.05  # 5% for bulk orders
    if promo_code == "PROMO20":
        discount += 0.20

    final_amount = subtotal * (1 - discount)

    # Quick debugging: display each step
    print(f"  {price=}, {quantity=}, {promo_code=}")
    print(f"  {subtotal=:.2f}, {discount=:.0%}")
    print(f"  {final_amount=:.2f}")

    return final_amount

calculate_discount(29.99, 12, "PROMO20")
#   price=29.99, quantity=12, promo_code='PROMO20'
#   subtotal=359.88, discount=25%
#   final_amount=269.91

from typing import TypedDict, List, Optional

# Define the structure of an API response
class Address(TypedDict):
    street: str
    city: str
    postal_code: str
    country: str

class User(TypedDict):
    id: int
    name: str
    email: str
    age: Optional[int]
    address: Address
    roles: List[str]

# Usage with decoded JSON data
def display_profile(user: User) -> None:
    """Display a typed user profile."""
    print(f"Name: {user['name']}")
    print(f"Email: {user['email']}")
    print(f"City: {user['address']['city']}")
    print(f"Roles: {', '.join(user['roles'])}")

profile: User = {
    "id": 1,
    "name": "Marie Curie",
    "email": "marie@example.com",
    "age": 35,
    "address": {
        "street": "12 Science Street",
        "city": "Paris",
        "postal_code": "75005",
        "country": "France",
    },
    "roles": ["admin", "researcher"],
}

display_profile(profile)
# Name: Marie Curie
# Email: marie@example.com
# City: Paris
# Roles: admin, researcher

# TypedDict with total=False (optional keys)
class SearchOptions(TypedDict, total=False):
    limit: int
    page: int
    sort: str
    filters: dict

# All keys are optional
opts: SearchOptions = {"limit": 20}
print(opts)  # {'limit': 20}

from functools import cached_property
import time

# Simulating a database connection pool
class DatabaseService:
    """Service with lazy database connection."""

    def __init__(self, host, port, db_name):
        self.host = host
        self.port = port
        self.db_name = db_name
        print(f"Service created (no connection yet)")

    @cached_property
    def connection_pool(self):
        """Create the connection pool (called only once)."""
        print(f"Creating pool to {self.host}:{self.port}...")
        time.sleep(0.1)  # Simulate connection delay
        return {
            "host": self.host,
            "port": self.port,
            "database": self.db_name,
            "size": 5,
            "active": True,
        }

    @cached_property
    def schema(self):
        """Load the database schema (called only once)."""
        print("Loading schema...")
        return ["users", "orders", "products", "logs"]

    def query(self, sql):
        """Execute a query using the pool."""
        pool = self.connection_pool  # Created on first access
        print(f"Executing on {pool['database']}: {sql}")

service = DatabaseService("db.example.com", 5432, "production")
# Service created (no connection yet)

service.query("SELECT count(*) FROM users")
# Creating pool to db.example.com:5432...
# Executing on production: SELECT count(*) FROM users

service.query("SELECT * FROM orders LIMIT 10")
# Executing on production: SELECT * FROM orders LIMIT 10
# (no pool recreation!)

# Comparison with classic @property
class Old:
    @property
    def expensive(self):
        print("Expensive computation...")  # Called on every access!
        return 42

class New:
    @cached_property
    def expensive(self):
        print("Expensive computation...")  # Called only once
        return 42