PythonStudio
Interview Problems

Image Processor

medium · job queue / producer–consumer

Build an image processor with a single worker and a FIFO queue. The constructor takes a `process_fn` — the function applied to each job. `submit(job)` appends the job to the queue and returns a job id (integers starting at 1, in submission order). `status(job_id)` returns 'queued', 'running', 'done', 'cancelled', or 'unknown' (for an id that was never submitted). `result(job_id)` returns the processed value for a done job, and None otherwise. `cancel(job_id)` cancels a queued job and returns True; it returns False for a running, done, cancelled, or unknown job — a running job always finishes. `step()` advances the simulation by one unit: if a job is running, finish it (run `process_fn` on it, store the result, mark it done); otherwise start the next queued job (FIFO) and mark it running. If nothing is queued and nothing is running, `step()` does nothing. This is a simulation of the single-worker design — one step() call advances the simulation by one unit, so the behaviour is testable without threads. The scaling discussion below is the real-world version. Browser test contract: the harness constructs ImageProcessor(lambda job: job.upper()) for every case. Jobs are strings. Keep the process_fn constructor and apply that supplied callable; do not hard-code uppercase in the class. The legacy capacity field in the JSON test envelope is unused for this exercise. step() returns None in every state. Example 1: submit('a') → 1; status(1) → 'queued'; step() → None; status(1) → 'running'; step() → None; result(1) → 'A'. Example 2: submit('b') → 1; cancel(1) → True; status(1) → 'cancelled'; result(1) → None. Return the exact five status strings from the statement. No real threads or sockets are used.

Starter code

class ImageProcessor:
    def __init__(self, process_fn):
        pass

    def submit(self, job):
        pass

    def status(self, job_id):
        pass

    def result(self, job_id):
        pass

    def cancel(self, job_id):
        pass

    def step(self):
        pass
Loading editor…

Each test has a five-second execution limit. Your practice drafts and solved progress stay in this browser.

Return to the Python compiler