Python Programming Questions

Requirements

• Watch a shared folder, validate each report, aggregate metrics, and write one daily summary.
• Keep running if one file is bad.

Design
I would prefer a scheduled scanner over a pure file watcher, because shared folders can miss events. The CLI would:

1. list files in incoming/
2. move each file atomically to processing/
3. validate and parse it
4. write metrics to a temp summary file
5. rename the temp file to the final output when complete

Idempotent design
Idempotent means running it twice produces the same result, with no duplicate output. I would track a checksum or a (date, filename) key in a small state file or SQLite table. If the same file appears twice, skip it. If the run dies midway, rerun only unprocessed files.

Debuggability

• structured logs with filename, checksum, line number, and error
• quarantine bad files in failed/ with a reason file
• counters for processed, skipped, and failed
• a --file replay mode for one broken input

Worked example: if ops_2025-07-03.csv has a bad line 18, the run should still finish the other files, mark that one as failed, and include the failure in the final summary. That makes the job observable instead of mysterious.

Requirements: per-user 7-day rolling sum by event_time, scalable.

Idiomatic, vectorized Pandas (vectorized meaning the loop over rows runs as fast, compiled bulk code instead of a slow Python-level for loop, one row at a time):

import pandas as pd

df = pd.DataFrame({
    "user_id": ["u1", "u1", "u1", "u1", "u2"],
    "event_time": pd.to_datetime([
        "2024-01-01", "2024-01-03", "2024-01-05", "2024-01-09", "2024-01-02"
    ]),
    "value": [10, 20, 30, 40, 5],
})

# ensure datetime and sort
df['event_time'] = pd.to_datetime(df['event_time'])
df = df.sort_values(['user_id', 'event_time'])

# set index for time-based rolling and compute 7-day sum per user
result = (
    df.set_index('event_time')
      .groupby('user_id')['value']
      .rolling('7D')
      .sum()
      .reset_index(name='rolling_7d_sum')
)

Worked example, verified with pandas on CPython 3.12:

df = pd.DataFrame({
    "user_id": ["u1", "u1", "u1", "u1", "u2"],
    "event_time": pd.to_datetime([
        "2024-01-01", "2024-01-03", "2024-01-05", "2024-01-09", "2024-01-02"
    ]),
    "value": [10, 20, 30, 40, 5],
})
print(result)  # (built by running df through the code above)

Output:

  user_id event_time  rolling_7d_sum
0      u1 2024-01-01            10.0
1      u1 2024-01-03            30.0
2      u1 2024-01-05            60.0
3      u1 2024-01-09            90.0
4      u2 2024-01-02             5.0

Tracing u1's rows: Jan 1 has no prior rows in its trailing 7-day window, so the sum is just its own value, 10. Jan 3's window (Jan 3 back to Dec 27) includes Jan 1 and Jan 3, 10 + 20 = 30. Jan 5's window (back to Dec 29) includes Jan 1, 3, and 5, 10 + 20 + 30 = 60. Jan 9's window (back to Jan 2) EXCLUDES Jan 1, which is 8 days earlier, outside the 7-day cutoff, so it sums only Jan 3, 5, and 9: 20 + 30 + 40 = 90. This is exactly what makes it a time-based, not row-count-based, window: the number of prior rows included varies depending on how many actually fall within the trailing 7 real days, not a fixed count of the last N rows.

Notes: Uses time-based rolling with groupby which is vectorized and memory-efficient for large data. For millions of rows, ensure event_time is datetime64, operate on chunked parquet files (parquet: a compressed, columnar file format for tabular data, well suited to reading only the columns and row-groups a job actually needs) if memory constrained, and consider Dask or PySpark for distributed scaling once the data genuinely no longer fits on one machine: Dask mirrors the Pandas API but splits data into partitions and runs the same groupby/rolling-style operations across them in parallel; PySpark is a different, JVM-backed distributed engine with its own (similar but not identical) DataFrame API. Plain, chunked Pandas is enough for anything that still fits on a single machine's memory; reach for Dask/PySpark only once it genuinely does not.

Approach: implement LRU with a dict for storage and a doubly-linked list for order; use threading.RLock (a lock that the SAME thread can safely acquire again without deadlocking itself, unlike a plain threading.Lock, in case the cache logic ever needs to re-enter the lock while already holding it) for concurrency. Decorator returns wrapper that locks around lookups and updates, minimizing lock hold time.

Implementation:

import threading
from functools import wraps

def lru_cache(maxsize=128):
    def deco(func):
        cache = {}
        head = tail = None
        lock = threading.RLock()
        class Node:
            __slots__=('key','val','prev','next')
            def __init__(self,k,v):
                self.key=k;self.val=v;self.prev=self.next=None
        def _move_to_front(node):
            nonlocal head, tail
            if node is head:
                return  # already the most-recently-used entry, nothing to do
            # unlink node from wherever it currently sits
            if node.prev:
                node.prev.next = node.next
            if node.next:
                node.next.prev = node.prev
            if node is tail:
                tail = node.prev
            # relink it at the head (the most-recently-used end)
            node.prev = None
            node.next = head
            if head:
                head.prev = node
            head = node
            if tail is None:
                tail = node
        @wraps(func)
        def wrapper(*args, **kwargs):
            nonlocal head, tail   # wrapper reassigns both below on eviction; without
                                   # this declaration LEGB makes them locals instead
            key=(args,tuple(sorted(kwargs.items())))
            with lock:
                node=cache.get(key)
                if node:
                    _move_to_front(node); return node.val
            val=func(*args, **kwargs)
            with lock:
                if key in cache: return cache[key].val
                node=Node(key,val); cache[key]=node; _move_to_front(node)
                if len(cache)>maxsize:
                    # cache is over capacity: evict the true tail, the
                    # least-recently-used entry, from both the linked list
                    # and the dict
                    lru_node = tail
                    tail = lru_node.prev
                    if tail:
                        tail.next = None
                    else:
                        head = None
                    del cache[lru_node.key]
            return val
        return wrapper
    return deco

Worked example: verified on CPython 3.12, calling the decorated function through a full eviction cycle so the policy can actually be watched working, not just taken on faith.

calls = []

@lru_cache(maxsize=2)
def square(n):
    calls.append(n)
    return n * n

print(square(1))   # 1  (miss: cache empty, computed and cached; order (MRU->LRU): [1])
print(square(2))   # 4  (miss: computed and cached; order: [2, 1])
print(square(1))   # 1  (hit: served from cache, 1 moves back to the front; order: [1, 2])
print(square(3))   # 9  (miss: cache was full at {1, 2}; since 1 was just reused, 2 is now
                    # the least-recently-used entry and gets evicted to make room for 3; order: [3, 1])
print(square(2))   # 4  (miss again: 2 was evicted in the previous step, so this recomputes
                    # instead of hitting the cache; order: [2, 3])
print(calls)        # [1, 2, 3, 2]  -- 2 appears twice: proof it was actually evicted and
                     # had to be recomputed, not just a claim

Complexity: O(1) average get/set.

Contention: a single global lock, one shared threading.RLock protecting the whole cache, serializes cache access: only one thread at a time can even check whether something is cached, regardless of which key it wants, which becomes a bottleneck under heavy concurrent traffic. Two ways to reduce that: a read-mostly optimistic check (read the dict for a hit without taking the lock first, since a plain dict read is safe to race on for a snapshot lookup, and only take the lock to confirm the hit and update the linked-list ordering, so the common cache-hit path spends less time holding the lock), or shard locks (split one cache into several smaller caches, each with its own separate lock, and route each key to one shard by hashing it, e.g. shard = hash(key) % num_shards; two threads reading keys that land in different shards no longer contend for the same lock at all, at the cost of maxsize now being enforced per shard rather than globally).

The Global Interpreter Lock (GIL) is a mutex in CPython that ensures only one native thread executes Python bytecode at a time. It simplifies memory management but serializes CPU-bound Python code.

Workloads where threads still help despite the GIL:

• I/O-bound workloads: threads waiting on network, disk, or database I/O release the GIL during blocking calls, improving throughput (e.g., web crawlers, concurrent HTTP clients). Concretely, verified on CPython 3.12: two threads that each print, sleep (standing in for a blocking network call), then print again:

import threading, time

order = []

def worker(name, delay):
    order.append(f"{name} start")
    time.sleep(delay)      # GIL released for the duration of the sleep
    order.append(f"{name} done")

t1 = threading.Thread(target=worker, args=("A", 0.1))
t2 = threading.Thread(target=worker, args=("B", 0.1))
t1.start(); t2.start()
t1.join(); t2.join()
print(order)

A representative run printed ['A start', 'B start', 'B done', 'A done']: both start entries appear before either done, which is only possible because thread A released the GIL during its time.sleep(0.1), letting thread B start and run its own code during that wait, rather than A blocking B out entirely the way a genuine CPU-bound loop with no sleep/I-O would (there, the two threads would still only ever have one of them executing Python bytecode at a time, and the interleaving above would not occur). The exact order of the two done entries is not guaranteed (it depends on OS scheduling), but the invariant that matters, both starts landing before either finish, reliably demonstrates the GIL being released during the blocking wait, not a specific timing.

• Native-code parallelism: when threads call C extensions that release the GIL (numpy heavy computations, image processing libraries, cryptography), CPU work runs in parallel across cores.

For CPU-bound pure-Python tasks use multiprocessing or native extensions to achieve parallelism.

Approach

I would read the file line by line with bufio.Scanner. Opening errors are fatal, because there is no file to process. Line parse errors are recoverable, meaning a bad line does not stop the whole file, so I return them in a slice and keep the good records. If scanning itself fails, I return the partial data plus the scan error so the caller can decide whether to fail the job.

package main

import (
    "bufio"
    "fmt"
    "os"
    "strconv"
    "strings"
)

type Record struct {
    Name string
    Age  int
}

type ParseIssue struct {
    Line int
    Raw  string
    Err  error
}

func ParseFile(path string) ([]Record, []ParseIssue, error) {
    f, err := os.Open(path)
    if err != nil {
        return nil, nil, err
    }
    defer f.Close()

    scanner := bufio.NewScanner(f)
    scanner.Buffer(make([]byte, 1024), 1024*1024)

    var records []Record
    var issues []ParseIssue
    lineNum := 0

    for scanner.Scan() {
        lineNum++
        line := strings.TrimSpace(scanner.Text())
        if line == "" {
            continue
        }

        parts := strings.Split(line, ",")
        if len(parts) != 2 {
            issues = append(issues, ParseIssue{Line: lineNum, Raw: line, Err: fmt.Errorf("expected name,age")})
            continue
        }

        age, err := strconv.Atoi(strings.TrimSpace(parts[1]))
        if err != nil {
            issues = append(issues, ParseIssue{Line: lineNum, Raw: line, Err: err})
            continue
        }

        records = append(records, Record{
            Name: strings.TrimSpace(parts[0]),
            Age:  age,
        })
    }

    if err := scanner.Err(); err != nil {
        return records, issues, err
    }

    return records, issues, nil
}

Key points

• open failure: return immediately
• parsing failure: collect issue and continue
• scan failure: return partial results and an error

Example: with Ann,32, bad-line, and Bob,41, you get 2 records and 1 issue.

Complexity: O(n) time, O(k) memory for stored records and issues.