07 - Static Typing with Python

# Basic variable annotations (Python 3.6+)
name: str = "Alice"
age: int = 30
height: float = 5.9
is_active: bool = True

print(f"name is {name}, type: {type(name)}")
print(f"age is {age}, type: {type(age)}")
print(f"height is {height}, type: {type(height)}")
print(f"is_active is {is_active}, type: {type(is_active)}")

# Output:
# name is Alice, type: <class 'str'>
# age is 30, type: <class 'int'>
# height is 5.9, type: <class 'float'>
# is_active is True, type: <class 'bool'>

# Function with parameter and return type annotations (Python 3.5+)
def calculate_total_price(quantity: int, price: float) -> float:
    """Calculate the total price for a purchase."""
    return quantity * price

total = calculate_total_price(5, 19.99)
print(f"Total price: ${total:.2f}")
# Output:
# Total price: $99.95

# We can see the type annotations using the __annotations__ attribute
print(f"Function annotations: {calculate_total_price.__annotations__}")
# Output:
# Function annotations: {'quantity': <class 'int'>, 'price': <class 'float'>, 'return': <class 'float'>}

# Type comments (for Python 3.4 or older)
name = "Bob"  # type: str
age = 25  # type: int

def calculate_discount(price, percentage):
    # type: (float, float) -> float
    """Calculate the discount amount."""
    return price * (percentage / 100)

discount = calculate_discount(99.99, 20)
print(f"Discount amount: ${discount:.2f}")
# Output:
# Discount amount: $20.00

# Import the typing module to access more types (Python 3.5+)
from typing import Any, List, Dict, Tuple, Set

# Basic type examples
user_id: int = 12345
username: str = "data_engineer"
temperature: float = 72.8
is_available: bool = False
generic_data: Any = "This could be any type"  # Any type was available since Python 3.5

print(f"User ID: {user_id}")
print(f"Username: {username}")
print(f"Temperature: {temperature}")
print(f"Available: {is_available}")
print(f"Generic data: {generic_data}")

# Output:
# User ID: 12345
# Username: data_engineer
# Temperature: 72.8
# Available: False
# Generic data: This could be any type

from typing import List, Dict, Tuple, Set  # Python 3.5+

# List of integers
numbers: List[int] = [1, 2, 3, 4, 5]

# Dictionary mapping strings to floats
prices: Dict[str, float] = {
    "apple": 0.99,
    "banana": 0.59,
    "orange": 1.29
}

# Tuple with specific types for each position
person: Tuple[str, int, float] = ("Alice", 30, 5.8)

# Set of strings
unique_tags: Set[str] = {"python", "data", "engineering"}

print(f"Numbers: {numbers}")
print(f"Prices: {prices}")
print(f"Person: {person}")
print(f"Unique tags: {unique_tags}")

# Output:
# Numbers: [1, 2, 3, 4, 5]
# Prices: {'apple': 0.99, 'banana': 0.59, 'orange': 1.29}
# Person: ('Alice', 30, 5.8)
# Unique tags: {'python', 'data', 'engineering'}

# Python 3.9+ simplified syntax
numbers: list[int] = [1, 2, 3, 4, 5]
prices: dict[str, float] = {"apple": 0.99}
person: tuple[str, int, float] = ("Alice", 30, 5.8)

from typing import List, Dict, Tuple, Any  # Python 3.5+

# List of dictionaries (common for data processing)
users: List[Dict[str, Any]] = [
    {"id": 1, "name": "Alice", "active": True},
    {"id": 2, "name": "Bob", "active": False}
]

# Dictionary with tuple keys and list values
coordinates_map: Dict[Tuple[int, int], List[str]] = {
    (0, 0): ["origin", "center"],
    (10, 10): ["top-right"],
    (-10, 10): ["top-left"]
}

print(f"Users: {users}")
print(f"Coordinates map: {coordinates_map}")

# Output:
# Users: [{'id': 1, 'name': 'Alice', 'active': True}, {'id': 2, 'name': 'Bob', 'active': False}]
# Coordinates map: {(0, 0): ['origin', 'center'], (10, 10): ['top-right'], (-10, 10): ['top-left']}

from typing import Callable  # Python 3.5+

# Define a function type: takes two floats and returns a float
MathOperation = Callable[[float, float], float]

def apply_operation(x: float, y: float, operation: MathOperation) -> float:
    """Apply a mathematical operation to two numbers."""
    return operation(x, y)

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

def multiply(a: float, b: float) -> float:
    return a * b

result1 = apply_operation(5.0, 3.0, add)
result2 = apply_operation(5.0, 3.0, multiply)

print(f"5 + 3 = {result1}")
print(f"5 * 3 = {result2}")

# Output:
# 5 + 3 = 8.0
# 5 * 3 = 15.0

from typing import Optional  # Python 3.5+

# A function that returns a string or None
def get_user_name(user_id: int) -> Optional[str]:
    """Get a user's name from their ID, or None if user doesn't exist."""
    user_database = {
        1: "Alice",
        2: "Bob",
        3: "Charlie"
    }
    return user_database.get(user_id)  # Returns None if key doesn't exist

# Test with existing and non-existing users
for user_id in [1, 4]:
    name = get_user_name(user_id)
    if name is None:
        print(f"User {user_id} not found")
    else:
        print(f"User {user_id}: {name}")

# Output:
# User 1: Alice
# User 4 not found

from typing import Union, List, Dict, Any  # Python 3.5+

# A value that can be either an int or a string
user_identifier: Union[int, str] = "user_abc"
print(f"User identifier: {user_identifier}")

# Later in the code, we might assign a different type to the same variable
user_identifier = 12345
print(f"User identifier (changed): {user_identifier}")

# Function that accepts different types of data
def process_data(data: Union[List[Dict[str, Any]], Dict[str, Any]]) -> int:
    """Process either a single record or a list of records."""
    if isinstance(data, list):
        # Process a list of records
        record_count = len(data)
        print(f"Processing {record_count} records")
        return record_count
    else:
        # Process a single record
        print("Processing a single record")
        return 1

# Try both types of input
single_record = {"id": 1, "name": "Alice"}
multiple_records = [
    {"id": 1, "name": "Alice"},
    {"id": 2, "name": "Bob"}
]

count1 = process_data(single_record)
count2 = process_data(multiple_records)

print(f"Processed records: {count1}")
print(f"Processed records: {count2}")

# Output:
# User identifier: user_abc
# User identifier (changed): 12345
# Processing a single record
# Processed records: 1
# Processing 2 records
# Processed records: 2

# Python 3.10+ simplified union syntax
user_identifier: int | str = "user_abc"  # Instead of Union[int, str]

from typing import Union  # Python 3.5+

def display_length(value: Union[str, list]) -> None:
    """Display the length of a string or list."""
    # Type narrowing through isinstance check
    if isinstance(value, str):
        # Inside this block, the type checker knows value is a string
        print(f"String length: {len(value)}")
    else:
        # Inside this block, the type checker knows value is a list
        print(f"List length: {len(value)}")

# Test with different types
display_length("Hello, world!")
display_length([1, 2, 3, 4])

# Output:
# String length: 13
# List length: 4

from typing import TypedDict, List  # TypedDict requires Python 3.8+

# Define a structured dictionary type
class UserDict(TypedDict):
    id: int
    name: str
    email: str
    active: bool
    tags: List[str]

# Create a user with the expected structure
user: UserDict = {
    "id": 1,
    "name": "Alice Smith",
    "email": "alice@example.com",
    "active": True,
    "tags": ["admin", "developer"]
}

def format_user(user: UserDict) -> str:
    """Format a user for display."""
    status = "Active" if user["active"] else "Inactive"
    tags = ", ".join(user["tags"]) if user["tags"] else "No tags"
    return f"User {user['id']}: {user['name']} ({user['email']}) - {status} - Tags: {tags}"

print(format_user(user))

# Output:
# User 1: Alice Smith (alice@example.com) - Active - Tags: admin, developer

from typing_extensions import TypedDict  # For Python 3.5-3.7

from typing import NamedTuple, List  # Typed NamedTuple requires Python 3.6+

# Define a named tuple type
class DataPoint(NamedTuple):
    timestamp: str
    value: float
    tags: List[str]

# Create a data point
temperature = DataPoint(
    timestamp="2023-04-15T12:30:00",
    value=72.5,
    tags=["temperature", "indoor"]
)

# Access fields by name instead of index
print(f"Time: {temperature.timestamp}")
print(f"Value: {temperature.value}°F")
print(f"Tags: {', '.join(temperature.tags)}")

# Named tuples are still tuples and support unpacking
time, value, tags = temperature
print(f"Unpacked - Time: {time}, Value: {value}, Tags: {tags}")

# Output:
# Time: 2023-04-15T12:30:00
# Value: 72.5°F
# Tags: temperature, indoor
# Unpacked - Time: 2023-04-15T12:30:00, Value: 72.5, Tags: ['temperature', 'indoor']

# Install pyright using pip
pip install pyright

# Save this in a file named type_errors.py
from typing import List, Dict

def add_numbers(a: int, b: int) -> int:
    return a + b

def get_user_names(users: List[Dict[str, str]]) -> List[str]:
    return [user["name"] for user in users]

# Type error: passing a string instead of an int
result = add_numbers(5, "10")
print(f"Result: {result}")

# Type error: missing the 'name' key in one of the dictionaries
users = [
    {"name": "Alice", "email": "alice@example.com"},
    {"email": "bob@example.com"}  # Missing 'name' key
]
names = get_user_names(users)
print(f"Names: {names}")

# Run pyright on our file
pyright type_errors.py

# Save this in a file named type_errors_fixed.py
from typing import List, Dict, Any

def add_numbers(a: int, b: int) -> int:
    return a + b

def get_user_names(users: List[Dict[str, Any]]) -> List[str]:
    return [user["name"] if "name" in user else "Unknown" for user in users]

# Fixed: convert string to int before passing
result = add_numbers(5, int("10"))
print(f"Result: {result}")

# Fixed: handle missing 'name' key
users = [
    {"name": "Alice", "email": "alice@example.com"},
    {"email": "bob@example.com"}  # Missing 'name' key
]
names = get_user_names(users)
print(f"Names: {names}")

# Output:
# Result: 15
# Names: ['Alice', 'Unknown']

pyright type_errors_fixed.py

{
  "include": ["src"],
  "exclude": ["**/node_modules", "**/__pycache__"],
  "ignore": ["src/legacy"],
  "reportMissingImports": true,
  "reportMissingTypeStubs": false,
  "pythonVersion": "3.9",
  "typeCheckingMode": "basic"
}

# Start by adding types to function signatures
def process_data(data):  # No types yet
    result = []
    for item in data:
        result.append(transform_item(item))
    return result

# Then gradually add more specific types
from typing import List, Dict, Any

def transform_item(item: Dict[str, Any]) -> Dict[str, Any]:
    """Transform a single data item."""
    return {
        "id": item.get("id", 0),
        "name": item.get("name", "").upper(),
        "processed": True
    }

def process_data(data: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
    """Process a list of data items."""
    result = []
    for item in data:
        result.append(transform_item(item))
    return result

# Test with sample data
sample_data = [
    {"id": 1, "name": "Alice"},
    {"id": 2, "name": "Bob"}
]

processed = process_data(sample_data)
print(f"Processed data: {processed}")

# Output:
# Processed data: [{'id': 1, 'name': 'ALICE', 'processed': True}, {'id': 2, 'name': 'BOB', 'processed': True}]

from typing import Any, Dict, List

def parse_data(raw_data: Any) -> List[Dict[str, Any]]:
    """Parse data that could be in various formats."""
    if isinstance(raw_data, list):
        return raw_data  # Already a list
    elif isinstance(raw_data, dict):
        return [raw_data]  # Single dictionary
    elif isinstance(raw_data, str):
        try:
            import json
            parsed = json.loads(raw_data)
            return parse_data(parsed)  # Recursively handle the parsed data
        except json.JSONDecodeError:
            return [{"value": raw_data}]  # Treat as a single value
    else:
        return [{"value": str(raw_data)}]  # Convert to string

# Test with different input types
test_inputs = [
    [{"id": 1}, {"id": 2}],  # List of dictionaries
    {"id": 3},  # Single dictionary
    '{"id": 4}',  # JSON string
    "plain text",  # Plain string
    42  # Number
]

for input_data in test_inputs:
    result = parse_data(input_data)
    print(f"Input: {input_data} -> Result: {result}")

# Output:
# Input: [{'id': 1}, {'id': 2}] -> Result: [{'id': 1}, {'id': 2}]
# Input: {'id': 3} -> Result: [{'id': 3}]
# Input: {"id": 4} -> Result: [{'id': 4}]
# Input: plain text -> Result: [{'value': 'plain text'}]
# Input: 42 -> Result: [{'value': '42'}]

from typing import Dict, List, Union, Optional, TypedDict  # Python 3.8+ for TypedDict

# For Python 3.5-3.7, use:
# from typing import Dict, List, Union, Optional
# from typing_extensions import TypedDict

# Define types for database schema
class ColumnSchema(TypedDict):
    name: str
    data_type: str
    nullable: bool
    description: Optional[str]

class TableSchema(TypedDict):
    name: str
    columns: List[ColumnSchema]
    primary_key: List[str]

# Create a sample table schema
user_table: TableSchema = {
    "name": "users",
    "columns": [
        {"name": "id", "data_type": "integer", "nullable": False, "description": "Unique user ID"},
        {"name": "username", "data_type": "varchar(50)", "nullable": False, "description": "User's login name"},
        {"name": "email", "data_type": "varchar(100)", "nullable": False, "description": "User's email address"},
        {"name": "active", "data_type": "boolean", "nullable": False, "description": "Whether the user account is active"}
    ],
    "primary_key": ["id"]
}

def generate_create_table_sql(schema: TableSchema) -> str:
    """Generate SQL CREATE TABLE statement from schema."""
    columns_sql = []

    for column in schema["columns"]:
        parts = [f"{column['name']} {column['data_type']}"]

        if not column["nullable"]:
            parts.append("NOT NULL")

        if "description" in column and column["description"]:
            # Escape single quotes in the description
            escaped_desc = column["description"].replace("'", "''")
            parts.append(f"COMMENT '{escaped_desc}'")

        columns_sql.append(" ".join(parts))

    primary_key = f"PRIMARY KEY ({', '.join(schema['primary_key'])})"
    columns_sql.append(primary_key)

    sql = f"CREATE TABLE {schema['name']} (\n  " + ",\n  ".join(columns_sql) + "\n);"
    return sql

# Generate and print the SQL
sql = generate_create_table_sql(user_table)
print(sql)

# Output:
# CREATE TABLE users (
#   id integer NOT NULL COMMENT 'Unique user ID',
#   username varchar(50) NOT NULL COMMENT 'User''s login name',
#   email varchar(100) NOT NULL COMMENT 'User''s email address',
#   active boolean NOT NULL COMMENT 'Whether the user account is active',
#   PRIMARY KEY (id)
# );

# Standard library imports (typing was added to stdlib in Python 3.5)
from typing import List, Dict, Any, Callable, Optional

# Define types for our ETL pipeline
RawData = List[Dict[str, Any]]
TransformedData = List[Dict[str, Any]]
Extractor = Callable[[], RawData]
Transformer = Callable[[RawData], TransformedData]
Loader = Callable[[TransformedData], None]

def create_etl_pipeline(
    extract: Extractor,
    transform: Transformer,
    load: Loader
) -> Callable[[], None]:
    """Create an ETL pipeline from extract, transform, and load functions."""
    def pipeline() -> None:
        """Execute the ETL pipeline."""
        print("Starting ETL pipeline")

        # Extract
        print("Extracting data...")
        raw_data = extract()
        print(f"Extracted {len(raw_data)} records")

        # Transform
        print("Transforming data...")
        transformed_data = transform(raw_data)
        print(f"Transformed data into {len(transformed_data)} records")

        # Load
        print("Loading data...")
        load(transformed_data)
        print("Loading complete")

        print("ETL pipeline complete")

    return pipeline

# Example implementation of extract, transform, and load functions
def extract_sample_data() -> RawData:
    """Extract sample data (in real life, this would come from a database or API)."""
    return [
        {"id": 1, "name": "alice", "age": "30"},
        {"id": 2, "name": "bob", "age": "25"},
        {"id": 3, "name": "charlie", "age": None}
    ]

def transform_sample_data(data: RawData) -> TransformedData:
    """Transform the data by standardizing types and formats."""
    result = []
    for record in data:
        transformed = {
            "user_id": record["id"],
            "full_name": record["name"].title(),
            "age": int(record["age"]) if record["age"] is not None else None,
            "is_adult": True if record["age"] is not None and int(record["age"]) >= 18 else False
        }
        result.append(transformed)
    return result

def load_to_console(data: TransformedData) -> None:
    """Load the data (in this case, just print it)."""
    for record in data:
        print(f"Loaded: {record}")

# Create and run our pipeline
etl_pipeline = create_etl_pipeline(
    extract_sample_data,
    transform_sample_data,
    load_to_console
)

etl_pipeline()

# Output:
# Starting ETL pipeline
# Extracting data...
# Extracted 3 records
# Transforming data...
# Transformed data into 3 records
# Loading data...
# Loaded: {'user_id': 1, 'full_name': 'Alice', 'age': 30, 'is_adult': True}
# Loaded: {'user_id': 2, 'full_name': 'Bob', 'age': 25, 'is_adult': True}
# Loaded: {'user_id': 3, 'full_name': 'Charlie', 'age': None, 'is_adult': False}
# Loading complete
# ETL pipeline complete

# data_processor.py - Before adding type annotations

def parse_csv_line(line):
    """Parse a CSV line into a list of values."""
    return line.strip().split(',')

def process_record(record):
    """Process a single data record."""
    if len(record) < 3:
        return None

    # Extract and transform fields
    try:
        id_value = int(record[0])
        name = record[1].strip()
        value = float(record[2])
        tags = record[3].split(';') if len(record) > 3 else []

        return {
            "id": id_value,
            "name": name,
            "value": value,
            "tags": tags,
            "is_valid": value > 0
        }
    except (ValueError, IndexError):
        return None

def process_data(data_lines):
    """Process multiple lines of data."""
    records = []
    error_count = 0

    for i, line in enumerate(data_lines):
        record_data = parse_csv_line(line)
        processed_record = process_record(record_data)

        if processed_record:
            records.append(processed_record)
        else:
            error_count += 1
            print(f"Error processing line {i+1}: {line.strip()}")

    return {
        "records": records,
        "record_count": len(records),
        "error_count": error_count
    }

def calculate_statistics(processed_data):
    """Calculate statistics from processed data."""
    records = processed_data["records"]

    if not records:
        return {
            "count": 0,
            "total": 0,
            "average": 0,
            "min": None,
            "max": None
        }

    values = [r["value"] for r in records]

    return {
        "count": len(values),
        "total": sum(values),
        "average": sum(values) / len(values),
        "min": min(values),
        "max": max(values)
    }

def generate_report(processed_data, statistics):
    """Generate a report from processed data and statistics."""
    report_lines = []

    report_lines.append("DATA PROCESSING REPORT")
    report_lines.append("=====================")
    report_lines.append(f"Processed {processed_data['record_count']} records with {processed_data['error_count']} errors")
    report_lines.append("")
    report_lines.append("STATISTICS:")
    report_lines.append(f"- Total: {statistics['total']:.2f}")
    report_lines.append(f"- Average: {statistics['average']:.2f}")
    report_lines.append(f"- Min: {statistics['min']}")
    report_lines.append(f"- Max: {statistics['max']}")
    report_lines.append("")
    report_lines.append("RECORDS:")

    for record in processed_data["records"]:
        tags = f" [Tags: {', '.join(record['tags'])}]" if record["tags"] else ""
        status = "VALID" if record["is_valid"] else "INVALID"
        report_lines.append(f"- {record['id']}: {record['name']} = {record['value']} ({status}){tags}")

    return "\n".join(report_lines)

def main():
    # Sample data for testing
    sample_data = [
        "1,Temperature,23.5,sensor;outdoor",
        "2,Humidity,45.2,sensor;indoor",
        "3,Pressure,-1.0,sensor",
        "4,Invalid,abc,test",
        "5,Partial",
        "6,Wind Speed,15.7"
    ]

    # Process the data
    processed_data = process_data(sample_data)
    statistics = calculate_statistics(processed_data)
    report = generate_report(processed_data, statistics)

    # Print the report
    print(report)

if __name__ == "__main__":
    main()

# data_processor_typed.py - With type annotations
from typing import List, Dict, Optional, Union, TypedDict, Any, Tuple

# Custom type definitions
class Record(TypedDict):
    """Structure for a processed data record."""
    id: int
    name: str
    value: float
    tags: List[str]
    is_valid: bool

class Statistics(TypedDict):
    """Structure for statistical calculations."""
    count: int
    total: float
    average: float
    min: Optional[float]
    max: Optional[float]

class ProcessedData(TypedDict):
    """Structure for the results of data processing."""
    records: List[Record]
    record_count: int
    error_count: int

def parse_csv_line(line: str) -> List[str]:
    """Parse a CSV line into a list of values."""
    return line.strip().split(',')

def process_record(record: List[str]) -> Optional[Record]:
    """Process a single data record."""
    if len(record) < 3:
        return None

    # Extract and transform fields
    try:
        id_value = int(record[0])
        name = record[1].strip()
        value = float(record[2])
        tags = record[3].split(';') if len(record) > 3 else []

        return {
            "id": id_value,
            "name": name,
            "value": value,
            "tags": tags,
            "is_valid": value > 0
        }
    except (ValueError, IndexError):
        return None

def process_data(data_lines: List[str]) -> ProcessedData:
    """Process multiple lines of data."""
    records: List[Record] = []
    error_count: int = 0

    for i, line in enumerate(data_lines):
        record_data = parse_csv_line(line)
        processed_record = process_record(record_data)

        if processed_record:
            records.append(processed_record)
        else:
            error_count += 1
            print(f"Error processing line {i+1}: {line.strip()}")

    return {
        "records": records,
        "record_count": len(records),
        "error_count": error_count
    }

def calculate_statistics(processed_data: ProcessedData) -> Statistics:
    """Calculate statistics from processed data."""
    records = processed_data["records"]

    if not records:
        return {
            "count": 0,
            "total": 0.0,
            "average": 0.0,
            "min": None,
            "max": None
        }

    values: List[float] = [r["value"] for r in records]

    return {
        "count": len(values),
        "total": sum(values),
        "average": sum(values) / len(values),
        "min": min(values),
        "max": max(values)
    }

def generate_report(processed_data: ProcessedData, statistics: Statistics) -> str:
    """Generate a report from processed data and statistics."""
    report_lines: List[str] = []

    report_lines.append("DATA PROCESSING REPORT")
    report_lines.append("=====================")
    report_lines.append(f"Processed {processed_data['record_count']} records with {processed_data['error_count']} errors")
    report_lines.append("")
    report_lines.append("STATISTICS:")
    report_lines.append(f"- Total: {statistics['total']:.2f}")
    report_lines.append(f"- Average: {statistics['average']:.2f}")
    report_lines.append(f"- Min: {statistics['min']}")
    report_lines.append(f"- Max: {statistics['max']}")
    report_lines.append("")
    report_lines.append("RECORDS:")

    for record in processed_data["records"]:
        tags = f" [Tags: {', '.join(record['tags'])}]" if record["tags"] else ""
        status = "VALID" if record["is_valid"] else "INVALID"
        report_lines.append(f"- {record['id']}: {record['name']} = {record['value']} ({status}){tags}")

    return "\n".join(report_lines)

def main() -> None:
    # Sample data for testing
    sample_data: List[str] = [
        "1,Temperature,23.5,sensor;outdoor",
        "2,Humidity,45.2,sensor;indoor",
        "3,Pressure,-1.0,sensor",
        "4,Invalid,abc,test",
        "5,Partial",
        "6,Wind Speed,15.7"
    ]

    # Process the data
    processed_data: ProcessedData = process_data(sample_data)
    statistics: Statistics = calculate_statistics(processed_data)
    report: str = generate_report(processed_data, statistics)

    # Print the report
    print(report)

if __name__ == "__main__":
    main()

def calculate_discounted_price(price, discount_percentage, tax_rate):
    """Calculate the final price after discount and tax."""
    discounted_price = price * (1 - discount_percentage / 100)
    final_price = discounted_price * (1 + tax_rate / 100)
    return round(final_price, 2)

def group_by_category(products):
    """Group products by their category."""
    result = {}
    for product in products:
        category = product.get("category", "Uncategorized")
        if category not in result:
            result[category] = []
        result[category].append(product)
    return result

def find_user(user_id=None, email=None):
    """Find a user by ID or email."""
    if user_id is None and email is None:
        return None

    # Simulated database
    users = [
        {"id": 1, "email": "user1@example.com", "name": "User One"},
        {"id": 2, "email": "user2@example.com", "name": "User Two"}
    ]

    if user_id is not None:
        for user in users:
            if user["id"] == user_id:
                return user

    if email is not None:
        for user in users:
            if user["email"] == email:
                return user

    return None

def format_book_citation(book):
    """Format a book for citation."""
    authors = ", ".join(book["authors"])
    return f"{authors}. ({book['year']}). {book['title']}. {book['publisher']}."

def create_data_validator(validation_rules):
    """Create a validator function based on rules."""
    def validator(data):
        """Validate data against rules."""
        errors = []
        for field, rule in validation_rules.items():
            if field not in data:
                errors.append(f"Missing field: {field}")
            elif not rule(data[field]):
                errors.append(f"Invalid value for {field}: {data[field]}")
        return errors

    return validator

def calculate_discounted_price(
    price: float,
    discount_percentage: float,
    tax_rate: float
) -> float:
    """Calculate the final price after discount and tax."""
    discounted_price = price * (1 - discount_percentage / 100)
    final_price = discounted_price * (1 + tax_rate / 100)
    return round(final_price, 2)

# Test the function
result = calculate_discounted_price(100.0, 20.0, 10.0)
print(f"Final price: ${result}")
# Output:
# Final price: $88.0

from typing import List, Dict, Any

def group_by_category(products: List[Dict[str, Any]]) -> Dict[str, List[Dict[str, Any]]]:
    """Group products by their category."""
    result: Dict[str, List[Dict[str, Any]]] = {}
    for product in products:
        category = product.get("category", "Uncategorized")
        if category not in result:
            result[category] = []
        result[category].append(product)
    return result

# Test the function
sample_products = [
    {"id": 1, "name": "Laptop", "category": "Electronics"},
    {"id": 2, "name": "Mouse", "category": "Electronics"},
    {"id": 3, "name": "Coffee Mug", "category": "Kitchen"},
    {"id": 4, "name": "T-shirt", "category": "Clothing"}
]

grouped = group_by_category(sample_products)
print("Products grouped by category:")
for category, items in grouped.items():
    names = [item["name"] for item in items]
    print(f"- {category}: {', '.join(names)}")

# Output:
# Products grouped by category:
# - Electronics: Laptop, Mouse
# - Kitchen: Coffee Mug
# - Clothing: T-shirt

from typing import Dict, Optional, Any, Union

def find_user(
    user_id: Optional[int] = None,
    email: Optional[str] = None
) -> Optional[Dict[str, Union[int, str]]]:
    """Find a user by ID or email."""
    if user_id is None and email is None:
        return None

    # Simulated database
    users: List[Dict[str, Union[int, str]]] = [
        {"id": 1, "email": "user1@example.com", "name": "User One"},
        {"id": 2, "email": "user2@example.com", "name": "User Two"}
    ]

    if user_id is not None:
        for user in users:
            if user["id"] == user_id:
                return user

    if email is not None:
        for user in users:
            if user["email"] == email:
                return user

    return None

# Test the function
user1 = find_user(user_id=1)
user2 = find_user(email="user2@example.com")
nonexistent = find_user(user_id=999)

print(f"User 1: {user1}")
print(f"User 2: {user2}")
print(f"Nonexistent user: {nonexistent}")

# Output:
# User 1: {'id': 1, 'email': 'user1@example.com', 'name': 'User One'}
# User 2: {'id': 2, 'email': 'user2@example.com', 'name': 'User Two'}
# Nonexistent user: None

from typing import List, TypedDict

class Book(TypedDict):
    """Structure for book information."""
    title: str
    authors: List[str]
    year: int
    publisher: str

def format_book_citation(book: Book) -> str:
    """Format a book for citation."""
    authors = ", ".join(book["authors"])
    return f"{authors}. ({book['year']}). {book['title']}. {book['publisher']}."

# Test the function
python_book: Book = {
    "title": "Python Programming: A Modern Approach",
    "authors": ["Smith, John", "Doe, Jane"],
    "year": 2023,
    "publisher": "Tech Books Publishing"
}

citation = format_book_citation(python_book)
print(f"Citation: {citation}")

# Output:
# Citation: Smith, John, Doe, Jane. (2023). Python Programming: A Modern Approach. Tech Books Publishing.

from typing import Dict, List, Callable, Any, TypeVar

T = TypeVar('T')  # Generic type for values

ValidationRule = Callable[[Any], bool]
Validator = Callable[[Dict[str, Any]], List[str]]

def create_data_validator(
    validation_rules: Dict[str, ValidationRule]
) -> Validator:
    """Create a validator function based on rules."""
    def validator(data: Dict[str, Any]) -> List[str]:
        """Validate data against rules."""
        errors: List[str] = []
        for field, rule in validation_rules.items():
            if field not in data:
                errors.append(f"Missing field: {field}")
            elif not rule(data[field]):
                errors.append(f"Invalid value for {field}: {data[field]}")
        return errors

    return validator

# Test the function
def is_positive_number(value: Any) -> bool:
    """Check if value is a positive number."""
    return isinstance(value, (int, float)) and value > 0

def is_valid_email(value: Any) -> bool:
    """Simple check if value looks like an email."""
    return isinstance(value, str) and "@" in value

# Create a validator for a user record
user_validator = create_data_validator({
    "id": is_positive_number,
    "email": is_valid_email
})

# Test with valid data
valid_user = {"id": 1, "email": "user@example.com"}
print(f"Valid user errors: {user_validator(valid_user)}")

# Test with invalid data
invalid_user = {"id": -1, "email": "not_an_email"}
print(f"Invalid user errors: {user_validator(invalid_user)}")

# Test with missing field
incomplete_user = {"id": 1}
print(f"Incomplete user errors: {user_validator(incomplete_user)}")

# Output:
# Valid user errors: []
# Invalid user errors: ['Invalid value for id: -1', 'Invalid value for email: not_an_email']
# Incomplete user errors: ['Missing field: email']