47 Apple Python Developer Interview Questions & Answers

def parse_nonnegative_integers(values):
    # Reject a wrong container type with a clear error.
    if not isinstance(values, list):
        raise TypeError("values must be a list")

    # Build a new list so the caller's input is not changed.
    result = []

    # Convert each string and reject negative numbers.
    for value in values:
        number = int(value)
        if number < 0:
            raise ValueError("values must be nonnegative")
        result.append(number)

    return result


# Each tuple contains an input and its expected output.
valid_cases = [
    (["2", "5", "9"], [2, 5, 9]),
    ([], []),
    (["0"], [0]),
    (["4", "4"], [4, 4]),
    (["1000000"], [1000000]),
]

# Run the valid cases and verify that input data is not mutated.
for values, expected in valid_cases:
    original = values.copy()
    actual = parse_nonnegative_integers(values)
    assert actual == expected
    assert values == original


# Verify malformed text and a negative value.
value_error_cases = [
    ["abc"],
    ["7", "bad"],
    ["-1"],
]

for values in value_error_cases:
    try:
        parse_nonnegative_integers(values)
    except ValueError:
        pass
    else:
        raise AssertionError("expected ValueError")


# Verify that a wrong container type raises TypeError.
try:
    parse_nonnegative_integers(None)
except TypeError:
    pass
else:
    raise AssertionError("expected TypeError")

print("All focused tests passed")

from __future__ import annotations

import math
from datetime import datetime, timezone
from decimal import Decimal, InvalidOperation
from typing import Any
from zoneinfo import ZoneInfo, ZoneInfoNotFoundError


def is_missing(value: Any) -> bool:
    # Recognize common missing forms used by Python data inputs.
    if value is None:
        return True

    if isinstance(value, str):
        return value.strip() == ""

    if isinstance(value, float):
        return math.isnan(value)

    if isinstance(value, Decimal):
        return value.is_nan()

    return False


def require_text(value: Any, field_name: str) -> str:
    # Reject missing required text instead of silently inventing a value.
    if is_missing(value):
        raise ValueError(f"{field_name} is required")

    text = str(value).strip()
    if text == "":
        raise ValueError(f"{field_name} is required")

    return text


def localize_naive_datetime(
    value: datetime,
    zone: ZoneInfo,
    field_name: str,
) -> datetime:
    # Test both fold values because a clock hour can repeat.
    valid_utc_values: list[datetime] = []

    for fold in (0, 1):
        candidate = value.replace(tzinfo=zone, fold=fold)
        candidate_utc = candidate.astimezone(timezone.utc)
        round_trip = candidate_utc.astimezone(zone)

        # A valid local time must survive conversion to UTC and back.
        if round_trip.replace(tzinfo=None) == value and round_trip.fold == fold:
            valid_utc_values.append(candidate_utc)

    # Remove duplicate UTC values that represent the same real instant.
    unique_values = sorted(set(valid_utc_values))

    if not unique_values:
        raise ValueError(f"{field_name} is an invalid local time in {zone.key}")

    if len(unique_values) > 1:
        raise ValueError(f"{field_name} is an ambiguous local time in {zone.key}")

    return unique_values[0]


def parse_datetime(
    value: Any,
    source_zone: Any,
    field_name: str,
) -> datetime:
    # Accept a datetime object or an ISO formatted string.
    if isinstance(value, datetime):
        parsed = value
    elif isinstance(value, str) and value.strip():
        text = value.strip()

        # Convert the common Z suffix into the offset form accepted here.
        if text.endswith("Z"):
            text = text[:-1] + "+00:00"

        try:
            parsed = datetime.fromisoformat(text)
        except ValueError as error:
            raise ValueError(f"{field_name} is not a valid ISO datetime") from error
    else:
        raise ValueError(f"{field_name} is required")

    # A naive datetime needs an explicit zone from the source contract.
    if parsed.tzinfo is None or parsed.utcoffset() is None:
        zone_name = require_text(source_zone, "source_zone")

        try:
            zone = ZoneInfo(zone_name)
        except ZoneInfoNotFoundError as error:
            raise ValueError(f"Unknown source_zone: {zone_name}") from error

        return localize_naive_datetime(parsed, zone, field_name)

    # Convert every aware timestamp to the canonical UTC zone.
    return parsed.astimezone(timezone.utc)


def parse_amount(value: Any) -> Decimal:
    # Amount is optional, so missing input receives a documented default.
    if is_missing(value):
        return Decimal("0")

    try:
        amount = Decimal(str(value))
    except InvalidOperation as error:
        raise ValueError("amount must be numeric") from error

    if not amount.is_finite():
        raise ValueError("amount must be finite")

    return amount


def clean_records(
    records: list[dict[str, Any]],
) -> list[dict[str, Any]]:
    # Retain one deterministic winner for each business id.
    selected: dict[str, dict[str, Any]] = {}

    for raw in records:
        record_id = require_text(raw.get("id"), "id")
        source = require_text(raw.get("source"), "source")
        source_zone = raw.get("source_zone")

        cleaned = {
            "id": record_id,
            "source": source,
            "amount": parse_amount(raw.get("amount")),
            "event_time": parse_datetime(
                raw.get("event_time"),
                source_zone,
                "event_time",
            ),
            "updated_at": parse_datetime(
                raw.get("updated_at"),
                source_zone,
                "updated_at",
            ),
        }

        current = selected.get(record_id)

        # Compare every cleaned value except id to create a total tie rule.
        cleaned_rank = (
            cleaned["updated_at"],
            cleaned["source"],
            cleaned["event_time"],
            cleaned["amount"],
        )

        if current is None:
            selected[record_id] = cleaned
        else:
            current_rank = (
                current["updated_at"],
                current["source"],
                current["event_time"],
                current["amount"],
            )

            if cleaned_rank > current_rank:
                selected[record_id] = cleaned

    # Explicit sorting makes output independent of arrival order.
    result = sorted(
        selected.values(),
        key=lambda item: (item["event_time"], item["id"]),
    )

    # Validate the contract before returning data downstream.
    ids = [item["id"] for item in result]
    if len(ids) != len(set(ids)):
        raise ValueError("Duplicate ids remain after cleaning")

    for item in result:
        if not item["amount"].is_finite():
            raise ValueError("amount must be finite")

        for field_name in ("event_time", "updated_at"):
            value = item[field_name]

            if value.tzinfo is None or value.utcoffset() is None:
                raise ValueError(f"{field_name} must be an aware datetime")

            if value.utcoffset() != timezone.utc.utcoffset(value):
                raise ValueError(f"{field_name} must be converted to UTC")

    return result


def main() -> None:
    # Two records share id A1. The newest update must win.
    records = [
        {
            "id": "A1",
            "source": "store",
            "source_zone": "America/Los_Angeles",
            "amount": None,
            "event_time": "2026-07-29T09:00:00",
            "updated_at": "2026-07-29T09:05:00",
        },
        {
            "id": "A1",
            "source": "web",
            "source_zone": "UTC",
            "amount": "25.50",
            "event_time": "2026-07-29T16:00:00+00:00",
            "updated_at": "2026-07-29T16:10:00+00:00",
        },
        {
            "id": "B2",
            "source": "mobile",
            "source_zone": "Asia/Kolkata",
            "amount": "10",
            "event_time": "2026-07-29T20:30:00",
            "updated_at": "2026-07-29T20:35:00",
        },
    ]

    for record in clean_records(records):
        print(
            record["id"],
            record["source"],
            record["amount"],
            record["event_time"].isoformat(),
            record["updated_at"].isoformat(),
        )


if __name__ == "__main__":
    main()

from queue import Queue
from threading import Event, Lock, Thread

# This list keeps the input items in their original order.
items = ["apple", "banana", "apple", "pear", "banana"]

# Worker threads share this dictionary.
counts: dict[str, int] = {}

# The queue safely transfers tasks between threads.
tasks: Queue[str | None] = Queue()

# The lock protects the complete dictionary update.
counts_lock = Lock()

# The event releases all workers after setup is complete.
start_event = Event()


def worker() -> None:
    # Every worker waits for the same start signal.
    start_event.wait()

    while True:
        item = tasks.get()

        try:
            # One stop value ends one worker.
            if item is None:
                return

            # Protect the read and write as one logical operation.
            with counts_lock:
                counts[item] = counts.get(item, 0) + 1
        finally:
            # Mark every received queue item as complete.
            tasks.task_done()


# Create two workers for this example.
workers = [Thread(target=worker) for _ in range(2)]

# Start the threads. They wait at the event.
for thread in workers:
    thread.start()

# Add all real tasks before releasing the workers.
for item in items:
    tasks.put(item)

# Broadcast the start signal to both workers.
start_event.set()

# Wait until all real tasks are processed.
tasks.join()

# Add one stop value for each worker.
for _ in workers:
    tasks.put(None)

# Wait until the stop values are received.
tasks.join()

# Wait for both threads to exit cleanly.
for thread in workers:
    thread.join()

# Sort the output so its displayed order is deterministic.
print(sorted(counts.items()))

import socket
import struct


class ProtocolError(Exception):
    """Raised when received bytes break the message protocol."""


def recv_exact(sock: socket.socket, size: int) -> bytes:
    """Read exactly size bytes or raise if the connection closes early."""
    if size < 0:
        raise ValueError("size must not be negative")

    data = bytearray()

    while len(data) < size:
        try:
            chunk = sock.recv(size - len(data))
        except socket.timeout as exc:
            raise TimeoutError("timed out while reading from the socket") from exc
        except OSError as exc:
            raise ConnectionError("socket read failed") from exc

        if chunk == b"":
            raise ConnectionError("connection closed before the message was complete")

        data.extend(chunk)

    return bytes(data)


def read_length_prefixed_message(
    sock: socket.socket,
    max_message_size: int = 1_048_576,
) -> bytes:
    """Read one message with a four byte unsigned network order length."""
    if max_message_size < 0:
        raise ValueError("max_message_size must not be negative")

    header = recv_exact(sock, 4)
    message_size = struct.unpack("!I", header)[0]

    if message_size > max_message_size:
        raise ProtocolError("declared message size exceeds the allowed limit")

    return recv_exact(sock, message_size)


class NewlineMessageReader:
    """Read newline delimited messages while preserving extra bytes."""

    def __init__(self, max_message_size: int = 1_048_576) -> None:
        if max_message_size < 0:
            raise ValueError("max_message_size must not be negative")

        self.max_message_size = max_message_size
        self.buffer = bytearray()

    def read_message(self, sock: socket.socket) -> bytes:
        """Return one message without the trailing newline."""
        while True:
            newline_index = self.buffer.find(b"\n")

            if newline_index != -1:
                if newline_index > self.max_message_size:
                    raise ProtocolError("line exceeds the allowed limit")

                message = bytes(self.buffer[:newline_index])
                del self.buffer[: newline_index + 1]
                return message

            if len(self.buffer) > self.max_message_size:
                raise ProtocolError("line exceeds the allowed limit")

            try:
                chunk = sock.recv(4096)
            except socket.timeout as exc:
                raise TimeoutError("timed out while reading from the socket") from exc
            except OSError as exc:
                raise ConnectionError("socket read failed") from exc

            if chunk == b"":
                if self.buffer:
                    raise ConnectionError("connection closed before the newline arrived")
                raise EOFError("connection closed")

            self.buffer.extend(chunk)


def main() -> None:
    # Demonstrate a length prefixed message.
    sender, receiver = socket.socketpair()
    try:
        payload = b"hello"
        sender.sendall(struct.pack("!I", len(payload)) + payload)
        print(read_length_prefixed_message(receiver))
    finally:
        sender.close()
        receiver.close()

    # Demonstrate two newline delimited messages received together.
    sender, receiver = socket.socketpair()
    try:
        sender.sendall(b"first\nsecond\n")
        reader = NewlineMessageReader()
        print(reader.read_message(receiver))
        print(reader.read_message(receiver))
    finally:
        sender.close()
        receiver.close()


if __name__ == "__main__":
    main()

import heapq
import math
import os
import tempfile


def _write_sorted_chunk(values, directory, index):
    # Sort only the current bounded chunk in memory.
    values.sort()

    # Store the sorted values in a temporary text file.
    path = os.path.join(directory, f"chunk_{index}.txt")
    with open(path, "w", encoding="utf-8") as output_file:
        for value in values:
            output_file.write(f"{value!r}\n")

    return path


def _read_sorted_chunk(path):
    # Yield one value at a time from one sorted temporary file.
    with open(path, "r", encoding="utf-8") as input_file:
        for line in input_file:
            yield float(line)


def median_from_large_file(path, chunk_size=100000):
    # A positive chunk size is required for bounded processing.
    if chunk_size <= 0:
        raise ValueError("chunk_size must be greater than zero")

    total_count = 0
    chunk_paths = []

    # All temporary chunk files are removed when this block ends.
    with tempfile.TemporaryDirectory() as temporary_directory:
        current_chunk = []
        chunk_index = 0

        # Read the source file without loading the full dataset.
        with open(path, "r", encoding="utf-8") as source_file:
            for line_number, line in enumerate(source_file, start=1):
                text = line.strip()

                # Ignore blank lines.
                if not text:
                    continue

                try:
                    value = float(text)
                except ValueError as error:
                    raise ValueError(f"Invalid number on line {line_number}: {text!r}") from error

                # Reject values that do not have the required finite ordering.
                if not math.isfinite(value):
                    raise ValueError(f"Non finite number on line {line_number}: {text!r}")

                current_chunk.append(value)
                total_count += 1

                # Sort and store the chunk when it reaches the chosen limit.
                if len(current_chunk) == chunk_size:
                    chunk_paths.append(
                        _write_sorted_chunk(
                            current_chunk,
                            temporary_directory,
                            chunk_index,
                        )
                    )
                    current_chunk = []
                    chunk_index += 1

        # Store the last partial chunk.
        if current_chunk:
            chunk_paths.append(
                _write_sorted_chunk(
                    current_chunk,
                    temporary_directory,
                    chunk_index,
                )
            )

        if total_count == 0:
            raise ValueError("Cannot find the median of an empty dataset")

        # Create lazy iterators over the sorted temporary files.
        sorted_iterators = (_read_sorted_chunk(chunk_path) for chunk_path in chunk_paths)

        # Produce values in global sorted order without one full sorted list.
        merged_values = heapq.merge(*sorted_iterators)

        left_index = (total_count - 1) // 2
        right_index = total_count // 2
        left_value = None

        # Stop as soon as the right middle value is reached.
        for index, value in enumerate(merged_values):
            if index == left_index:
                left_value = value

            if index == right_index:
                if left_index == right_index:
                    return value

                # Divide first to reduce overflow risk for large finite floats.
                return left_value / 2.0 + value / 2.0

    raise RuntimeError("Median positions were not reached")


if __name__ == "__main__":
    # This example has six values, so the median is the average of 3 and 5.
    example_path = "numbers.txt"
    with open(example_path, "w", encoding="utf-8") as example_file:
        example_file.write("9\n1\n5\n3\n7\n2\n")

    print(median_from_large_file(example_path, chunk_size=2))
    os.remove(example_path)

import math
from typing import Any


def transform_records(records: list[object]) -> dict[str, list[dict[str, Any]]]:
    # Validate the outer container before processing any records.
    if not isinstance(records, list):
        raise TypeError("records must be a list")

    valid_records: list[dict[str, Any]] = []
    rejected_records: list[dict[str, Any]] = []
    seen_ids: set[int] = set()

    # Enumerate preserves input order and gives each record its original index.
    for index, raw_record in enumerate(records):
        error: str | None = None

        # Every record must be a dictionary.
        if not isinstance(raw_record, dict):
            error = "record must be a dictionary"
        else:
            # Report missing fields before checking their values.
            missing_fields = [field for field in ("id", "name", "price") if field not in raw_record]

            if missing_fields:
                error = "missing required fields: " + ", ".join(missing_fields)
            else:
                record_id = raw_record["id"]
                name = raw_record["name"]
                price = raw_record["price"]

                # Boolean is a subclass of integer, so reject it explicitly.
                if not isinstance(record_id, int) or isinstance(record_id, bool):
                    error = "id must be an integer"
                elif record_id in seen_ids:
                    error = "id must be unique"
                elif not isinstance(name, str) or not name.strip():
                    error = "name must be a nonempty string"
                elif not isinstance(price, (int, float)) or isinstance(price, bool):
                    error = "price must be a number"
                elif not math.isfinite(price):
                    error = "price must be finite"

        # Record every failure instead of silently dropping the input.
        if error is not None:
            rejected_records.append(
                {
                    "index": index,
                    "record": raw_record,
                    "error": error,
                }
            )
            continue

        # The checks above guarantee that these fields have valid values.
        record_id = raw_record["id"]
        name = raw_record["name"]
        price = raw_record["price"]
        seen_ids.add(record_id)

        # Create a normalized copy so the original dictionary is not changed.
        valid_records.append(
            {
                "id": record_id,
                "name": name.strip(),
                "price": float(price),
            }
        )

    return {
        "valid": valid_records,
        "rejected": rejected_records,
    }


if __name__ == "__main__":
    sample_records: list[object] = [
        {"id": 2, "name": " Keyboard ", "price": 99},
        {"id": 1, "name": "Mouse"},
        {"id": 3, "name": "Cable", "price": True},
        {"id": 4, "name": "Adapter", "price": float("nan")},
        {"id": 2, "name": "Monitor", "price": 299.0},
        {"id": 5, "name": "Stand", "price": 49.5, "source": "store"},
        "not a dictionary",
    ]

    result = transform_records(sample_records)

    # Valid records keep their relative input order.
    assert [record["id"] for record in result["valid"]] == [2, 5]

    # Values are normalized into new dictionaries.
    assert result["valid"][0] == {
        "id": 2,
        "name": "Keyboard",
        "price": 99.0,
    }

    # Extra fields are ignored by the explicit output schema.
    assert result["valid"][1] == {
        "id": 5,
        "name": "Stand",
        "price": 49.5,
    }

    # Every bad record is reported with its original position.
    assert [record["index"] for record in result["rejected"]] == [1, 2, 3, 4, 6]
    assert result["rejected"][0]["error"] == "missing required fields: price"
    assert result["rejected"][1]["error"] == "price must be a number"
    assert result["rejected"][2]["error"] == "price must be finite"
    assert result["rejected"][3]["error"] == "id must be unique"
    assert result["rejected"][4]["error"] == "record must be a dictionary"

    # Empty input produces two empty result lists.
    assert transform_records([]) == {"valid": [], "rejected": []}

    # Invalid outer input fails immediately.
    try:
        transform_records({"id": 1})  # type: ignore[arg-type]
    except TypeError as exc:
        assert str(exc) == "records must be a list"
    else:
        raise AssertionError("Expected TypeError")

    print(result)

import socket
import struct
from collections.abc import Iterator


HEADER_SIZE = 4
MAX_FRAME_SIZE = 1024 * 1024
MAX_BUFFER_SIZE = MAX_FRAME_SIZE + 64 * 1024


class FrameError(Exception):
    """Raised when received bytes do not form a valid frame."""


class FrameParser:
    """Incrementally parses four byte length prefixed frames."""

    def __init__(
        self,
        max_frame_size: int = MAX_FRAME_SIZE,
        max_buffer_size: int = MAX_BUFFER_SIZE,
    ) -> None:
        if max_frame_size < 1:
            raise ValueError("max_frame_size must be positive")
        if max_buffer_size < max_frame_size:
            raise ValueError("max_buffer_size must allow one complete payload")

        self.max_frame_size = max_frame_size
        self.max_buffer_size = max_buffer_size
        self.buffer = bytearray()
        self.expected_length: int | None = None

    def feed(self, data: bytes) -> list[bytes]:
        """Adds received bytes and returns all complete payloads."""
        if not data:
            return []

        self.buffer.extend(data)
        messages: list[bytes] = []

        while True:
            # Read the header only when all four header bytes are present.
            if self.expected_length is None:
                if len(self.buffer) < HEADER_SIZE:
                    break

                self.expected_length = struct.unpack("!I", self.buffer[:HEADER_SIZE])[0]
                del self.buffer[:HEADER_SIZE]

                if self.expected_length == 0:
                    raise FrameError("empty frames are not allowed")
                if self.expected_length > self.max_frame_size:
                    raise FrameError("frame is larger than the allowed limit")

            # Keep an incomplete payload for the next socket read.
            if len(self.buffer) < self.expected_length:
                break

            # bytes creates an independent immutable payload for the caller.
            payload = bytes(self.buffer[: self.expected_length])
            del self.buffer[: self.expected_length]
            messages.append(payload)
            self.expected_length = None

        # Check the remaining unread data after complete frames were removed.
        if len(self.buffer) > self.max_buffer_size:
            raise FrameError("receive buffer limit exceeded")

        return messages

    def finish(self) -> None:
        """Checks parser state when the peer closes its sending side."""
        if self.expected_length is not None or self.buffer:
            raise FrameError("connection ended during an incomplete frame")


def encode_frame(
    payload: bytes,
    max_frame_size: int = MAX_FRAME_SIZE,
) -> bytes:
    """Builds one four byte length prefixed frame."""
    if not payload:
        raise ValueError("payload must not be empty")
    if len(payload) > max_frame_size:
        raise ValueError("payload is larger than the allowed limit")

    return struct.pack("!I", len(payload)) + payload


def receive_frames(
    sock: socket.socket,
    idle_timeout: float = 5.0,
    read_size: int = 4096,
) -> Iterator[bytes]:
    """Yields complete frames until the peer closes the connection."""
    if idle_timeout <= 0:
        raise ValueError("idle_timeout must be positive")
    if read_size <= 0:
        raise ValueError("read_size must be positive")

    parser = FrameParser()
    sock.settimeout(idle_timeout)

    while True:
        try:
            chunk = sock.recv(read_size)
        except socket.timeout as exc:
            raise TimeoutError("no bytes arrived before the idle timeout") from exc

        if not chunk:
            parser.finish()
            return

        for message in parser.feed(chunk):
            yield message


def run_tests() -> None:
    # Test every possible split position in one encoded frame.
    frame = encode_frame(b"apple")
    for split_at in range(1, len(frame)):
        parser = FrameParser()
        assert parser.feed(frame[:split_at]) == []
        assert parser.feed(frame[split_at:]) == [b"apple"]
        parser.finish()

    # Test several frames arriving in one read.
    parser = FrameParser()
    combined = encode_frame(b"one") + encode_frame(b"two")
    assert parser.feed(combined) == [b"one", b"two"]
    parser.finish()

    # Test a declared length above the allowed maximum.
    parser = FrameParser(max_frame_size=8, max_buffer_size=16)
    try:
        parser.feed(struct.pack("!I", 9))
        raise AssertionError("oversized frame was accepted")
    except FrameError:
        pass

    # Test a connection ending during an incomplete payload.
    parser = FrameParser()
    assert parser.feed(struct.pack("!I", 5) + b"ab") == []
    try:
        parser.finish()
        raise AssertionError("truncated frame was accepted")
    except FrameError:
        pass

    # Test real socket reads that are smaller than one frame.
    sender, receiver = socket.socketpair()
    try:
        encoded = encode_frame(b"network")
        sender.sendall(encoded)
        sender.shutdown(socket.SHUT_WR)

        received = list(
            receive_frames(
                receiver,
                idle_timeout=1.0,
                read_size=3,
            )
        )
        assert received == [b"network"]
    finally:
        sender.close()
        receiver.close()

    print("All tests passed")


if __name__ == "__main__":
    run_tests()

from __future__ import annotations

import threading
import time
from collections.abc import Callable
from typing import Generic, TypeVar

K = TypeVar("K")
V = TypeVar("V")


class TTLStore(Generic[K, V]):
    def __init__(
        self,
        clock: Callable[[], float] = time.monotonic,
        cleanup_every: int = 100,
        cleanup_limit: int = 20,
    ) -> None:
        # Cleanup settings must allow the scan to make progress.
        if cleanup_every < 1:
            raise ValueError("cleanup_every must be at least 1")
        if cleanup_limit < 1:
            raise ValueError("cleanup_limit must be at least 1")

        # Each dictionary entry contains the value and its expiry time.
        self._data: dict[K, tuple[V, float]] = {}

        # A monotonic clock is used by default.
        # Tests can provide a fake clock instead.
        self._clock = clock

        # The lock protects complete read, check, and delete sequences.
        self._lock = threading.RLock()

        # Cleanup runs after a fixed number of public operations.
        self._cleanup_every = cleanup_every
        self._cleanup_limit = cleanup_limit
        self._operations = 0

        # Cleanup uses a key snapshot and continues from this position.
        self._cleanup_keys: list[K] = []
        self._cleanup_index = 0

    def set(self, key: K, value: V, ttl_seconds: float) -> None:
        # A negative TTL is rejected because it is usually a caller error.
        if ttl_seconds < 0:
            raise ValueError("ttl_seconds cannot be negative")

        with self._lock:
            # Use the same clock for writes, reads, and cleanup.
            expires_at = self._clock() + ttl_seconds
            self._data[key] = (value, expires_at)
            self._record_operation_locked()

    def get(self, key: K, default: V | None = None) -> V | None:
        with self._lock:
            entry = self._data.get(key)

            if entry is None:
                self._record_operation_locked()
                return default

            value, expires_at = entry

            # The entry is expired at the exact expiry boundary.
            if self._clock() >= expires_at:
                del self._data[key]
                self._record_operation_locked()
                return default

            self._record_operation_locked()
            return value

    def delete(self, key: K) -> bool:
        with self._lock:
            existed = key in self._data
            self._data.pop(key, None)
            self._record_operation_locked()
            return existed

    def size(self) -> int:
        with self._lock:
            # This is a bounded cleanup, not a complete expiry scan.
            self._cleanup_locked()
            return len(self._data)

    def _record_operation_locked(self) -> None:
        # The caller already holds the lock.
        self._operations += 1

        if self._operations >= self._cleanup_every:
            self._operations = 0
            self._cleanup_locked()

    def _cleanup_locked(self) -> None:
        # Create a new snapshot after the previous one is fully examined.
        if self._cleanup_index >= len(self._cleanup_keys):
            self._cleanup_keys = list(self._data.keys())
            self._cleanup_index = 0

        now = self._clock()
        checked = 0

        # Examine only a limited number of keys during this pass.
        while self._cleanup_index < len(self._cleanup_keys) and checked < self._cleanup_limit:
            key = self._cleanup_keys[self._cleanup_index]
            self._cleanup_index += 1
            checked += 1

            # The key may have been deleted after the snapshot was created.
            entry = self._data.get(key)

            if entry is not None and now >= entry[1]:
                del self._data[key]


class FakeClock:
    def __init__(self) -> None:
        self.now = 0.0

    def __call__(self) -> float:
        return self.now

    def advance(self, seconds: float) -> None:
        if seconds < 0:
            raise ValueError("seconds cannot be negative")
        self.now += seconds


if __name__ == "__main__":
    # Use a fake clock to demonstrate expiry without real waiting.
    clock = FakeClock()
    store: TTLStore[str, str] = TTLStore(
        clock=clock,
        cleanup_every=3,
        cleanup_limit=2,
    )

    store.set("token", "active", ttl_seconds=5)
    print(store.get("token"))

    # The exact expiry time counts as expired.
    clock.advance(5)
    print(store.get("token", "missing"))
    print(store.size())