Dictionaries & Hash Maps

def resolve_application_config(
    cli_flags: dict[str, any],
    env_vars: dict[str, any],
    base_defaults: dict[str, any]
) -> dict[str, any]:
    """
    Resolves settings with priority: CLI Flags > Environment Variables > Base Defaults.
    """
    # 1. Start with a clone of base defaults
    resolved = base_defaults.copy()
    
    # 2. Layer environment variables (only for recognized setting keys)
    for key in base_defaults.keys():
        if key in env_vars and env_vars[key] is not None:
            resolved[key] = env_vars[key]
            
    # 3. Layer CLI flags (highest priority; skip None values)
    for key, val in cli_flags.items():
        if key in resolved and val is not None:
            resolved[key] = val
            
    return resolved

# Configuration fixtures:
defaults = {
    "HOST": "127.0.0.1",
    "PORT": 8000,
    "WORKERS": 4,
    "LOG_LEVEL": "INFO",
    "ENABLE_TELEMETRY": True
}

environment = {
    "PORT": 8080,
    "LOG_LEVEL": "DEBUG",
    "HOST": "0.0.0.0"
}

command_line = {
    "PORT": 9000,           # Overrides both env and default!
    "WORKERS": 16,          # Overrides default
    "HOST": None            # Ignored: CLI flag was not supplied
}

print("=" * 65)
print(f"{'HIERARCHICAL CONFIGURATION RESOLVER':^65}")
print("=" * 65)

final_config = resolve_application_config(command_line, environment, defaults)

for setting, value in final_config.items():
    print(f"{setting:<20}: {value}")
print("=" * 65)

# Assert correctness
assert final_config["PORT"] == 9000       # CLI won
assert final_config["HOST"] == "0.0.0.0"  # Env won because CLI was None
assert final_config["WORKERS"] == 16      # CLI won
assert final_config["ENABLE_TELEMETRY"] is True # Default preserved
print("Configuration Cascade Hierarchy Verified!")

def analyze_document_frequencies(corpus: str, top_n: int = 3) -> dict:
    """
    Computes token occurrences and extracts top-N vocabulary terms.
    """
    # 1. Clean and tokenize text
    for char in [",", ".", "!", "?", ";", ":", "\""]:
        corpus = corpus.replace(char, " ")
        
    tokens = corpus.lower().split()
    
    # 2. Tally occurrences using dict.get()
    frequencies = {}
    for token in tokens:
        frequencies[token] = frequencies.get(token, 0) + 1
        
    # 3. Sort by occurrence count descending, then alphabetically ascending
    sorted_items = sorted(frequencies.items(), key=lambda item: (-item[1], item[0]))
    top_terms = sorted_items[:top_n]
    
    # 4. Compute metrics
    total_words = len(tokens)
    unique_words = len(frequencies)
    lexical_diversity = round(unique_words / total_words, 3) if total_words > 0 else 0.0
    
    return {
        "frequencies": frequencies,
        "top_terms": top_terms,
        "total_tokens": total_words,
        "unique_tokens": unique_words,
        "lexical_diversity": lexical_diversity
    }

sample_doc = """
  Python is fast. Python is simple, and Python is productive.
  Engineers love Python because Python makes development productive and robust!
"""

print("=" * 60)
print(f"{'CORPUS TOKEN FREQUENCY ANALYZER':^60}")
print("=" * 60)

stats = analyze_document_frequencies(sample_doc, top_n=3)
print(f"Total Tokens Processed:  {stats['total_tokens']}")
print(f"Unique Vocabulary Terms: {stats['unique_tokens']}")
print(f"Lexical Diversity:       {stats['lexical_diversity']}")
print("-" * 60)
print("Top 3 Most Frequent Terms:")
for word, count in stats["top_terms"]:
    print(f"  '{word:<12}' -> {count} occurrences")
print("=" * 60)

class InMemoryTTLCache:
    def __init__(self, max_capacity: int = 1000):
        self._store = {}
        self.max_capacity = max_capacity
        
    def set(self, key: str, value: any, ttl_seconds: int, current_timestamp: int):
        """Stores key with expiration timestamp. Trims oldest LIFO item if at capacity."""
        # Evict if capacity reached
        if len(self._store) >= self.max_capacity and key not in self._store:
            evicted_key, _ = self._store.popitem()  # LIFO eviction
            print(f"[CAPACITY EVICT] Evicted key '{evicted_key}'")
            
        self._store[key] = {
            "val": value,
            "expires_at": current_timestamp + ttl_seconds
        }
        
    def get(self, key: str, current_timestamp: int) -> any:
        """Retrieves value. Evicts lazily if expired."""
        entry = self._store.get(key)
        if not entry:
            return None
            
        # Check expiration
        if current_timestamp >= entry["expires_at"]:
            del self._store[key]  # Lazy eviction
            print(f"[LAZY EVICT] Key '{key}' expired and purged.")
            return None
            
        return entry["val"]
        
    def purge_expired(self, current_timestamp: int) -> int:
        """Purges all expired keys safely without mutating during iteration."""
        # Snapshot expired keys first to prevent 'RuntimeError: dictionary changed size'
        expired_keys = [
            k for k, entry in self._store.items()
            if current_timestamp >= entry["expires_at"]
        ]
        for k in expired_keys:
            del self._store[k]
        return len(expired_keys)
        
    def __len__(self):
        return len(self._store)

# Test the cache engine:
cache = InMemoryTTLCache(max_capacity=3)
clock = 1000  # Simulated epoch second

print("=" * 65)
print(f"{'IN-MEMORY TTL KEY-VALUE CACHE STORE':^65}")
print("=" * 65)

# Insert 3 keys with varying TTLs
cache.set("session:u101", {"user": "Alice"}, ttl_seconds=30, current_timestamp=clock)
cache.set("session:u102", {"user": "Bob"}, ttl_seconds=60, current_timestamp=clock)
cache.set("session:u103", {"user": "Carol"}, ttl_seconds=10, current_timestamp=clock)

print(f"Cache Size at t={clock}: {len(cache)} entries")
print("-" * 65)

# Advance clock to t=1015 (u103 expired)
clock = 1015
val_u101 = cache.get("session:u101", current_timestamp=clock) # Still valid
val_u103 = cache.get("session:u103", current_timestamp=clock) # Expired!

print(f"Lookup u101 (valid):   {val_u101}")
print(f"Lookup u103 (expired): {val_u103}")
print(f"Cache Size at t={clock}: {len(cache)} entries")
print("-" * 65)

# Bulk purge at t=1070 (u101 and u102 now expired)
clock = 1070
purged = cache.purge_expired(current_timestamp=clock)
print(f"Bulk Purge at t={clock}: Purged {purged} expired entries.")
print(f"Final Cache Size:       {len(cache)} entries")
print("=" * 65)