Prompt Editor

Here are three distinct, comprehensive, and practical Python coding challenges designed to test deep knowledge of core Python programming, system design, and algorithmic efficiency. --- ### Challenge 1: Transactional Key-Value Store with Nested Rollbacks #### 1. Challenge Title: Transactional Key-Value Store with Nested Rollbacks #### 2. Problem Description: Implement an in-memory, thread-safe transactional key-value store. The store must support standard CRUD operations (`SET`, `GET`, `DELETE`) along with tracking operations (`COUNT`). Most importantly, it must support nested transaction blocks (`BEGIN`, `COMMIT`, `ROLLBACK`). When a transaction is started (`BEGIN`), all subsequent write operations occur within that transaction's local scope. A transaction can be committed (`COMMIT`), making its changes permanent and merging them into the parent scope (or the global store if it is the root transaction). Alternatively, a transaction can be rolled back (`ROLLBACK`), discarding all changes made within its scope and restoring the state to what it was prior to the matching `BEGIN`. #### 3. Input Specification: * **Keys and Values:** Strings only. Valid keys match the regex `^[a-zA-Z0-9_\-]{1,64}$`. * **Operations:** Methods exposed via a robust class interface: * `set(key: str, value: str) -> None` * `get(key: str) -> Optional[str]` * `delete(key: str) -> None` * `count(value: str) -> int` (Returns the number of keys currently set to the specified value). * `begin() -> None` (Starts a new nested transaction transaction). * `commit() -> None` (Commits the current transaction block). * `rollback() -> None` (Rolls back the current transaction block). #### 4. Output Specification: * `get` returns the string value if found, or `None` if the key does not exist (or was deleted) in the current active scope. * `count` returns a non-negative integer representing the frequency of a value in the current active scope. * Transactions methods do not return values but must raise descriptive, custom exceptions on invalid operations (e.g., committing or rolling back without an active transaction). #### 5. Core Requirements & Constraints: * **Time Complexity:** * `get`, `set`, and `delete` must run in amortized $\mathcal{O}(1)$ time. * `count` must run in $\mathcal{O}(1)$ time. (Hint: Maintain a secondary value-frequency mapping that is also transaction-aware; scanning the entire store to count values is unacceptable). * `begin` must run in $\mathcal{O}(1)$ time. * **Memory Efficiency:** Deep-copying the entire database state upon starting a transaction is unacceptable. Implement write-ahead tracking or copy-on-write mechanisms for keys modified within transactions. * **No External Libraries:** Standard library only. #### 6. Edge Cases to Consider: * **Nested Modification and Rollback:** Modifying a key in transaction level 2, rolling back level 2, and ensuring the key retains its level 1 (or global) value. * **Deleting a Key Inside a Transaction:** Deleting a key in an active transaction that exists in a parent transaction/global state. A `get` call within the active transaction must return `None`, but rolling back must restore the key's previous value. * **No Active Transaction:** Calling `commit()` or `rollback()` when no transactions have been started must raise a custom exception (e.g., `NoActiveTransactionError`). #### 7. Recommended Best Practices: * **Type Hinting:** Use strict type annotations (`typing.Optional`, `typing.Dict`, etc.) complying with PEP 484. * **Custom Exceptions:** Define a clear exception hierarchy derived from a base `KeyValueStoreError`. * **Thread Safety:** Use reentrant locks (`threading.RLock`) to ensure the store is thread-safe. * **Clean Design:** Structure transaction states using a stack of dictionaries (representing delta changes) or a parent-pointer tree structure. #### 8. Example Test Case: python # Setup db = TransactionalKVStore() db.set("a", "foo") db.set("b", "foo") db.begin() # Transaction 1 db.set("a", "bar") assert db.get("a") == "bar" db.begin() # Transaction 2 (Nested) db.delete("a") assert db.get("a") is None db.rollback() # Rollback Transaction 2 assert db.get("a") == "bar" db.commit() # Commit Transaction 1 assert db.get("a") == "bar" assert db.count("foo") == 1 # Only "b" is "foo" now --- ### Challenge 2: Streaming Log Anomaly Detector with Rolling Time-Windows #### 2. Problem Description: In production monitoring, processing massive data streams in real-time is crucial. Implement an anomaly detector that processes a stream of web server log entries. The detector must identify spikes in the HTTP error rate (5xx status codes) within a sliding time window (e.g., 60 seconds). If the ratio of 5xx errors to total requests exceeds a given threshold (e.g., > 10%) *and* a minimum sample size (e.g., at least 50 requests) is met within the active sliding window, the detector must emit an alert. #### 3. Input Specification: * **Stream Source:** A python generator yielding dictionaries representing individual log records. * **Log Record Schema:** python { "timestamp": float, # Unix epoch timestamp in seconds (e.g., 1672531199.123) "status_code": int, # HTTP status code (e.g., 200, 404, 503) "path": str # Requested endpoint path (e.g., "/api/v1/checkout") } * **Detector Configuration:** * `window_size`: An integer representing the window size in seconds (e.g., `60`). * `threshold`: A float between `0.0` and `1.0` representing the error rate limit (e.g., `0.10` for 10%). * `min_requests`: An integer representing the minimum number of logs in the window before alerts can trigger. #### 4. Output Specification: * The detector must be implemented as a generator function that takes the raw stream and configurations as inputs, and yields alert dictionaries only at the precise moment an anomaly is detected. * **Alert Schema:** python { "window_start": float, "window_end": float, "total_requests": int, "error_count": int, "error_rate": float } #### 5. Core Requirements & Constraints: * **Time Complexity:** Processing each log entry from the generator must run in $\mathcal{O}(1)$ amortized time. Re-scanning historical logs within the sliding window for every incoming log is strictly prohibited. * **Memory Complexity:** Bound space complexity to $\mathcal{O}(W)$ where $W$ is the number of logs received within the sliding window. Older logs must be evicted immediately as the window slides forward. * **Stream Properties:** Timestamps are generally monotonically increasing, but the system must handle late-arriving logs (out-of-order logs) up to a maximum delay of `max_out_of_order` seconds (e.g., 2 seconds). Logs arriving older than this delay relative to the highest seen timestamp should be silently dropped. #### 6. Edge Cases to Consider: * **Large Gaps in Activity:** If the server goes completely silent for a period greater than `window_size`, the sliding window must properly expire older logs and recalculate the sliding metrics without needing dummy events. * **Zero-Division Safeguards:** Handle scenarios where the number of requests in a window drops to zero cleanly without raising `ZeroDivisionError`. * **High Burst Density:** Handling hundreds of logs sharing the exact same millisecond timestamp without corrupting window boundaries. #### 7. Recommended Best Practices: * **Data Structures:** Use `collections.deque` for efficient double-ended queue operations to slide windows. * **Clean Generators:** Leverage `yield` and generator pipelines (`log_stream -> anomaly_detector -> alert_handler`) to maintain memory efficiency and lazy execution. * **Validation:** Validate input parameters (e.g., `threshold` must be between 0 and 1, `window_size` must be positive) and raise `ValueError` for invalid settings. #### 8. Example Test Case: python # Setup Configuration window_size = 10 # 10-second sliding window threshold = 0.5 # 50% error rate trigger min_requests = 4 # Mock Generator Input Stream def log_generator(): logs = [ {"timestamp": 100.0, "status_code": 200, "path": "/"}, {"timestamp": 101.0, "status_code": 200, "path": "/"}, {"timestamp": 102.0, "status_code": 500, "path": "/"}, {"timestamp": 103.0, "status_code": 500, "path": "/"}, # At t=103, requests=4, errors=2 (50%). This meets min_requests=4 and threshold=0.5. Trigger! {"timestamp": 115.0, "status_code": 200, "path": "/"}, # At t=115, the window is [105, 115]. The older logs are evicted. # Active window has 1 log (t=115, 200). Error rate=0. No trigger. ] for log in logs: yield log detector = LogAnomalyDetector(window_size, threshold, min_requests) alerts = list(detector.process_stream(log_generator())) assert len(alerts) == 1 assert alerts[0]["error_rate"] == 0.5 assert alerts[0]["total_requests"] == 4 --- ### Challenge 3: Concurrent Task Runner with Dependency Resolution #### 1. Challenge Title: Concurrent Task Runner with Dependency Resolution (DAG Engine) #### 2. Problem Description: Implement an asynchronous execution engine that executes a set of tasks configured as a Directed Acyclic Graph (DAG). Tasks have specified dependencies (other tasks that must complete successfully before they can start). To optimize execution speed, the engine must execute tasks concurrently using python's `asyncio` framework as soon as their dependencies are met. It must support task retries for intermittent failures and handle cascading failures gracefully: if a task fails permanently (exhausting its retries), all downstream tasks depending on it must be marked as `SKIPPED`, while unrelated parallel task branches continue executing unaffected. #### 3. Input Specification: * **Task Definition:** Tasks are defined as class instances or highly structured configurations: * `name`: Unique string identifier matching `^[a-zA-Z0-9_\-]{1,64}$`. * `coroutine`: An awaitable Python callable (e.g., a function returning a coroutine). * `dependencies`: A list of strings containing the names of parent tasks. * `retries`: An integer representing maximum retry attempts (default is `0`). * **Input Schema:** A collection (e.g., list or dict) containing all tasks in the DAG. #### 4. Output Specification: * The runner must return a structured execution report containing: * `status`: A dictionary mapping each task name to its final state (`"SUCCESS"`, `"FAILED"`, `"SKIPPED"`). * `results`: A dictionary mapping each task name to its return value or the raised Exception object. * `execution_order`: A list of task names in the exact sequence they completed. #### 5. Core Requirements & Constraints: * **Cycle Detection:** Before beginning execution, the engine must perform a topological analysis to verify the graph is a valid DAG. If a cycle is detected, it must raise a custom `CyclicDependencyError` and abort execution immediately. * **Concurrency Maximization:** The engine must not execute tasks sequentially if they can run in parallel. Tasks with completed dependencies must be scheduled on the event loop immediately. * **Safe Resource Limits:** The engine must support an optional `max_workers` parameter to limit the number of concurrently executing asynchronous tasks at any single moment. * **No External Libraries:** Must rely entirely on Python's built-in `asyncio` module and standard libraries. #### 6. Edge Cases to Consider: * **Dynamic Task Failure:** An early-stage task fails on all retries. The immediate children and deep descendants must immediately resolve as `SKIPPED` without hanging the event loop, while unrelated sibling nodes run to successful completion. * **Complex Interdependencies:** Diamond dependency graphs (A leads to B and C, which both lead to D). Task D must execute exactly once, only after both B and C complete successfully. * **Zero Dependencies:** Handling multiple root nodes that have no parents (they should start running concurrently immediately). #### 7. Recommended Best Practices: * **Topological Sort:** Use Kahn’s algorithm or DFS (Depth First Search) with recursion-stack state tracking to check for cycles and validate the DAG structure. * **Async Synchronization:** Use primitives such as `asyncio.Event` to coordinate dependencies between tasks, or manage state transitions using an active scheduler loop. * **Exception Preservation:** Ensure that if a task fails, its original traceback/exception object is stored in the results dictionary for debugging. * **Typing & Documentation:** Ensure type hints specify async callables (e.g., `Callable[..., Awaitable[Any]]`). #### 8. Example Test Case: python import asyncio # Define some mock task actions async def step_1(): await asyncio.sleep(0.1) return "Data A" async def step_2(): await asyncio.sleep(0.1) return "Data B" async def step_3(results): # Depends on step_1 and step_2 return f"Combined: {results['step_1']} & {results['step_2']}" # Graph Definition tasks = [ Task(name="step_1", coroutine=step_1, dependencies=[]), Task(name="step_2", coroutine=step_2, dependencies=[]), Task(name="step_3", coroutine=step_3, dependencies=["step_1", "step_2"]) ] runner = AsyncDAGRunner(tasks, max_workers=2) report = asyncio.run(runner.run()) # Validations assert report["status"]["step_1"] == "SUCCESS" assert report["status"]["step_2"] == "SUCCESS" assert report["status"]["step_3"] == "SUCCESS" assert report["results"]["step_3"] == "Combined: Data A & Data B"