Backend Live Coding

# ======================================================================
# Q: Merge overlapping intervals (e.g. promotion periods).
#
# O(n log n) because of the sorting.
# ======================================================================

def merge(intervals: list[tuple[int, int]]) -> list[tuple[int, int]]:
    result = []
    for start, end in sorted(intervals):
        if result and start <= result[-1][1]:
            result[-1] = (result[-1][0], max(result[-1][1], end))
        else:
            result.append((start, end))
    return result

merge([(1, 3), (2, 6), (8, 10), (9, 12)])   # [(1, 6), (8, 12)]

# ======================================================================
# Q: Find two products whose prices add up exactly to a budget (two-sum).
#
# O(n) time and O(n) memory, instead of O(n²) with two loops. For money I work
# with integer cents.
# ======================================================================

def two_sum(prices: list[int], budget: int) -> tuple[int, int] | None:
    seen = {}                                # price -> index
    for i, p in enumerate(prices):
        if budget - p in seen:
            return seen[budget - p], i
        seen[p] = i
    return None

# ======================================================================
# Q: Implement an LRU cache with capacity N (get and put in O(1)).
#
# O(1) for both. If asked without OrderedDict: a dict plus a doubly linked list.
# In practice, functools.lru_cache is already available.
# ======================================================================

from collections import OrderedDict

class LRUCache:
    def __init__(self, capacity: int):
        self.capacity = capacity
        self.data = OrderedDict()

    def get(self, key):
        if key not in self.data:
            return None
        self.data.move_to_end(key)          # most recently used
        return self.data[key]

    def put(self, key, value):
        self.data[key] = value
        self.data.move_to_end(key)
        if len(self.data) > self.capacity:
            self.data.popitem(last=False)   # evicts the least recently used

# ======================================================================
# Q: Design a URL shortener.
#
# Requirements: far more reads than writes (e.g. 100:1), low redirect latency,
# unique short codes. Code: base62 of the auto-increment id (short, collision-
# free, but predictable) or a random 7-character code with a unique constraint
# and retry. Storage: PostgreSQL with a unique index on the code. Read path:
# Redis cache (code -> URL), 301 or 302 (302 allows analytics). Analytics:
# clicks are not written synchronously to the database. They go to a queue and
# are batch inserted. Other concerns: link expiry, rate limiting on creation,
# checking for malicious URLs.
# ======================================================================

ALPHABET = string.digits + string.ascii_letters      # 62 characters

def base62(n: int) -> str:
    s = []
    while True:
        n, rem = divmod(n, 62)
        s.append(ALPHABET[rem])
        if n == 0:
            return ''.join(reversed(s))

# ======================================================================
# Q: Design a rate limiter: 100 requests per minute per API key.
#
# Options: fixed window (INCR with EXPIRE, simple, but allows double the
# requests at the window boundary), sliding window with a sorted set (accurate,
# uses more memory), token bucket (allows bursts, the most common in APIs).
# Note: here we also count the requests that were rejected. For absolute
# accuracy, write it as a Lua script. Return 429 with Retry-After and
# X-RateLimit-Remaining headers. Redis is shared by all workers.
# ======================================================================

def allow(api_key, limit=100, window=60):
    now = time.time()
    key = f'rl:{api_key}'
    pipe = r.pipeline()
    pipe.zremrangebyscore(key, 0, now - window)     # drop the old entries
    pipe.zadd(key, {f'{now}:{uuid.uuid4().hex}': now})
    pipe.zcard(key)
    pipe.expire(key, window)
    _, _, count, _ = pipe.execute()
    return count <= limit