Python Programming Questions

Approach

I’d normalize by lowercasing, removing punctuation, and collapsing repeated spaces. Then I’d use a set of normalized keys to keep only the first occurrence. A set is a data structure that gives fast membership checks, so this stays simple and efficient for a few thousand strings.

import re
from typing import List


def normalize_title(title: str) -> str:
    cleaned = re.sub(r'[\W_]+', ' ', title.lower())
    return ' '.join(cleaned.split())


def dedupe_titles(titles: List[str]) -> List[str]:
    seen = set()
    result = []

    for title in titles:
        norm = normalize_title(title)
        if norm in seen:
            continue
        seen.add(norm)
        result.append(norm)

    return result


titles = ['  Weekly Update!', 'weekly update', 'Budget Review', 'Budget review  ']
print(dedupe_titles(titles))

The regex pattern r'[\W_]+' is what actually strips punctuation: \W (capital W) matches any character that is NOT a letter, digit, or underscore, so it catches punctuation, symbols, and stray whitespace runs in one go; adding _ to the character class folds underscores into that same "replace with a space" rule too, since \W alone would NOT match an underscore (underscores count as word characters). The + means one-or-more, so any run of these unwanted characters collapses to a single space rather than leaving multiple spaces behind, and ' '.join(cleaned.split()) then trims leading/trailing spaces and collapses any remaining internal runs down to exactly one space each.

Example

Input: [' Weekly Update!', 'weekly update', 'Budget Review', 'Budget review ']

Output:

['weekly update', 'budget review']

That is the literal stdout of print(dedupe_titles(titles)) from the code block above.

Why this works: the first weekly update is kept, and the later duplicate is skipped because it normalizes to the same key.

Complexity: O(n * m) time, where m is string length, and O(n) extra space for the set.

If the team wants to keep original casing, I would store the first original string in result instead of the normalized version.

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.

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.

Experiment design
Compare three functions in a tight loop: (A) raise/catch exception on error, (B) return error code and check, (C) pre-validated path (no error). Use timeit and large N.

Code:

import timeit

def raise_path(n):
    for i in range(n):
        try:
            if i%100==0: raise ValueError
        except ValueError:
            pass

def errorcode_path(n):
    for i in range(n):
        ok = True            # the normal, no-error case
        if i%100==0: ok = False   # the simulated error case
        if not ok: pass

def prevalidated_path(n):
    for i in range(n):
        pass                 # no error branch at all: the baseline

n=1000000
print(timeit.timeit(lambda: raise_path(n), number=3))
print(timeit.timeit(lambda: errorcode_path(n), number=3))
print(timeit.timeit(lambda: prevalidated_path(n), number=3))

A correction to the code above (a genuine bug, not just a style choice): as originally sketched, errorcode_path set ok = False unconditionally at the top of every iteration and never set it back to True on the normal path, so if not ok: was True on every single iteration, not just the 1-in-100 simulated errors. That does not model "check an error code" at all, it just runs the pass branch every time. The fix is the one shown above: ok = True by default (the common, no-error case), flipped to False only on the simulated error, so errorcode_path and raise_path are actually testing the same 1% error frequency against each other. The third function, prevalidated_path, is also added here: it has no error branch whatsoever, and exists specifically as the baseline "cost of the loop itself, with no error handling of any kind" that the question's three-way comparison (raise/catch, return-code, pre-validated) asks for; the original sketch defined only the first two.

Interpretation

• Run this at varying simulated error frequencies (change i%100==0 to i%1==0 for 100% errors, or remove the if entirely for 0%) rather than trusting a single frequency; the shape of the gap between raise_path and errorcode_path as a function of error frequency is the actual finding, a single absolute number from one machine is not portable or reproducible elsewhere and should not be reported as "the" result. prevalidated_path's time is the floor: the cost of iterating n times with no error-handling machinery of any kind, useful as the baseline both other paths are measured against.
• The mechanism, not a specific number, is what to lead with: constructing and raising a real exception involves allocating an exception object, populating a traceback, and unwinding the stack to find a matching except, real, nonzero work that a plain boolean check never does; at 0% error frequency, the try block itself still costs something (setting up the exception-handling frame) even though nothing is ever raised, so raise_path at 0% errors is a fair comparison of that baseline try-frame overhead against errorcode_path's baseline if overhead. As error frequency rises toward 100%, the actual raise/unwind cost starts to dominate raise_path's total time in a way it never does for errorcode_path, which does the same constant amount of work (ok = True, one comparison) whether or not that iteration's error flag ends up True.
• For production ETL, prefer error-code or pre-validation for expected, frequent errors, reserve exceptions for truly exceptional control flow, situations that are rare enough that even a real per-raise cost barely matters in aggregate, and where the cleaner control-flow and forced handling (you cannot silently ignore a raised exception the way an unchecked error code can be ignored) outweigh that cost.
• Profile memory and CPU (not just the timeit total) to ensure GC or traceback construction isn't dominating the comparison for reasons unrelated to the actual branch being tested, for example, a test harness that also does unrelated allocation inside the timed loop.

Approach: schedule tasks, wait with timeout, cancel pending tasks after timeout, re-raise exceptions from completed tasks. (A coroutine is what you get back from calling an async def function; calling it does not run it yet, asyncio.create_task(c) is what actually schedules a coroutine to start running concurrently with everything else.)

Implementation:

import asyncio

async def gather_with_timeout(coros, timeout):
    tasks = [asyncio.create_task(c) for c in coros]
    done, pending = await asyncio.wait(tasks, timeout=timeout)
    # cancel pending
    for t in pending:
        t.cancel()
    # await cancellations to finish
    if pending:
        await asyncio.gather(*pending, return_exceptions=True)
    # propagate exceptions from done
    results = []
    for t in done:
        exc = t.exception()
        if exc:
            # re-raise first exception
            raise exc
        results.append(t.result())
    return tuple(results), tuple(pending)

# Usage example
async def main():
    async def work(n):
        await asyncio.sleep(n); return n
    done, pending = await gather_with_timeout([work(0.1), work(2)], timeout=1)
    print(done, pending)

# asyncio.run(main())

Notes: ensures pending tasks are cancelled and awaited to avoid warnings; exceptions from completed tasks propagate.

Running it: uncommenting asyncio.run(main()) and adding print(len(pending), pending[0].cancelled()) right after print(done, pending) shows what actually happens: work(0.1) finishes well inside the 1-second timeout and work(2) does not, so done prints as (0.1,) (the tuple of results from the finished work(0.1)) and the follow-up print shows 1 True, one pending task, and it reports cancelled() == True, confirming the still-running work(2) really was cancelled rather than left dangling.