Clean Code and Best Practices Questions

Covers the principles and hands on practices that produce readable, maintainable, and reliable code. Core elements include intent revealing and consistent naming, small focused functions and classes that follow single responsibility, avoiding duplication through refactoring and appropriate abstractions, clear structure and separation of concerns, following language specific idioms and style guides, consistent formatting, concise comments that explain nonobvious intent, defensive programming and robust error handling, edge case handling and input validation, use of linters and static analysis, incremental refactoring techniques, and pragmatic trade offs between ideal design and delivery constraints. Interviewers will also probe involvement in code reviews, version control hygiene, code metrics, and how candidates advocate for and teach coding standards to peers.

EasyTechnical

82 practiced

Logging is preferable to print statements in production. For a batch scoring job (runs nightly), outline a logging strategy: log levels, structured logs vs plain text, correlation IDs, sensitive data handling, and how logs map to alerts and dashboards.

HardTechnical

78 practiced

Implement a concurrency-safe in-memory cache for model inference in Python suitable for a multi-process WSGI server. The cache should support get(key), set(key, value, ttl=None), and handle process-local caches as well as a pluggable external cache backend (e.g., Redis). Emphasize clear abstractions and thread/process safety.

Sample Answer

Approach: build a Cache abstraction with a thread-safe process-local LRU-like dict plus optional external backend implementing the same interface (example: Redis). Local cache uses threading.Lock for concurrency; since WSGI typically forks multiple processes, each process has its own local cache. External backend provides cross-process sharing. TTL is enforced on both layers; local cache has a background cleaner thread.

python

import threading
import time
import pickle
from typing import Any, Optional

class Backend:
    def get(self, key: str) -> Optional[bytes]:
        raise NotImplementedError
    def set(self, key: str, value: bytes, ttl: Optional[int]) -> None:
        raise NotImplementedError

class RedisBackend(Backend):
    def __init__(self, redis_client):
        self.r = redis_client
    def get(self, key):
        return self.r.get(key)
    def set(self, key, value, ttl):
        if ttl:
            self.r.setex(key, ttl, value)
        else:
            self.r.set(key, value)

class ProcessLocalCache:
    def __init__(self, max_items=1000, clean_interval=60):
        self.store = {}  # key -> (expiry_ts or None, bytes)
        self.lock = threading.Lock()
        self.max_items = max_items
        self.cleaner = threading.Thread(target=self._cleaner, daemon=True)
        self.clean_interval = clean_interval
        self.cleaner.start()

    def _cleaner(self):
        while True:
            time.sleep(self.clean_interval)
            now = time.time()
            with self.lock:
                keys = [k for k,(exp,_) in self.store.items() if exp is not None and exp <= now]
                for k in keys: del self.store[k]

    def get(self, key):
        now = time.time()
        with self.lock:
            item = self.store.get(key)
            if not item: return None
            exp, data = item
            if exp is not None and exp <= now:
                del self.store[key]
                return None
            return data

    def set(self, key, value_bytes, ttl=None):
        exp = time.time()+ttl if ttl else None
        with self.lock:
            if len(self.store) >= self.max_items:
                # simple eviction: remove oldest expiry or arbitrary
                self.store.pop(next(iter(self.store)))
            self.store[key] = (exp, value_bytes)

class InferenceCache:
    def __init__(self, backend: Optional[Backend]=None, local_max=1000):
        self.local = ProcessLocalCache(max_items=local_max)
        self.backend = backend

    def get(self, key: str) -> Optional[Any]:
        data = self.local.get(key)
        if data is not None:
            return pickle.loads(data)
        if self.backend:
            b = self.backend.get(key)
            if b:
                obj = pickle.loads(b)
                # populate local for faster subsequent reads
                self.local.set(key, b, ttl=None)  # TTL unknown; could encode expiry in payload
                return obj
        return None

    def set(self, key: str, value: Any, ttl: Optional[int]=None):
        b = pickle.dumps(value)
        self.local.set(key, b, ttl)
        if self.backend:
            self.backend.set(key, b, ttl)

Key points:- Thread-safety via locks for local cache; processes are isolated so no cross-process locks needed.- External backend (Redis) is pluggable for cross-process sharing.- TTL enforced locally; for accurate TTL propagation, encode expiry in payload or rely on backend TTL (Redis handles it).- Eviction here is simple—swap for LRU if needed (collections.OrderedDict or cachetools).Complexity: get/set local O(1); external depends on backend (Redis: network).Edge cases: large pickled objects, serialization errors, clock skew across machines, TTL consistency between local and external. Alternatives: use shared memory (multiprocessing.Manager), or use Redis-only with local negative-cache to avoid stale data.

EasyTechnical

65 practiced

Describe good version-control hygiene for data science projects. Cover: commit granularity and messages, branch naming, handling large data or model artifacts, and ways to prevent secrets from being committed. Include recommended git workflows for small teams (2-10 engineers).

Sample Answer

Commit granularity & messages- Make small, atomic commits: one logical change per commit (e.g., “add data-cleaning step for null handling”, not “many fixes”).- Commit frequently but meaningfully so history shows intent and is reversible.- Messages: use imperative tense, short summary (50 chars), optional blank line, descriptive body explaining why and any trade-offs; reference ticket IDs when relevant.

Branch naming- Use clear, predictable names: feature/<id>-short-description, bugfix/<id>-short, hotfix/<version>, release/<version>.- Keep branches short-lived (merge within days/weeks) to reduce drift.

Handling large data & model artifacts- Never store raw datasets or large model binaries in Git. Use: - git-lfs for moderately large files (and with remote support) - DVC or MLflow to version data and models while keeping pointers in Git - Object storage (S3/GCS) or dedicated artifact registry for heavy assets; commit only metadata/checkpoints and reproducible pipeline configs- Store deterministic pipeline scripts and seed info so artifacts can be reproduced from pointers.

Preventing secrets from being committed- Use environment variables and secret managers (Vault, AWS Secrets Manager). Do not hard-code keys.- Add sensitive files to .gitignore and use templates (.env.example) instead of real secrets.- Enforce pre-commit hooks (pre-commit framework) + git-secrets or truffleHog to scan for secrets before commit.- Have CI checks to detect leaked secrets; rotate credentials if accidentally committed.

Recommended git workflows for small teams (2–10 engineers)- GitHub Flow / Trunk-based variant: - Protect main branch (require PRs, CI passing, code review 1–2 approvals). - Developers create short-lived feature branches, open PRs describing changes and tests, squash/merge on approval. - Use feature flags for risky changes.- For releases needing versioning, add a lightweight release branch flow: create release/<version>, stabilize, then merge back to main and tag.- Automate CI: linting, unit tests, model training smoke tests, DVC data checks, and artifact uploads.

Additional best practices- Keep notebooks minimal in Git; convert heavy analysis to scripts or store outputs externally.- Encourage README and runbooks for reproducing experiments.- Regularly prune stale branches and document branching policy.

HardTechnical

72 practiced

Implement a small library-style API in Python: class StreamingAggregator that supports add(record: dict), merge(other: StreamingAggregator), and summary() -> dict. The aggregator should compute count, sum, and distinct count of a specified key across streaming records. Focus on a clear, testable API and document complexity and merge semantics.

Sample Answer

Approach: implement a mergeable streaming aggregator that tracks count, sum, and distinct count for a specified numeric key (or any hashable value for distinct). Provide a clean API: add(record: dict), merge(other), summary()->dict. I'll implement an exact distinct counter (set-backed) and provide notes about memory/trade-offs and an option to swap in an approximate algorithm (HyperLogLog) for production use.

python

from typing import Any, Dict, Optional

class StreamingAggregator:
    """
    StreamingAggregator(key, approximate=False)

    - key: the record field to aggregate (must exist in incoming records)
    - approximate: reserved flag (not implemented here) to indicate approximate distinct counting

    Methods:
    - add(record: dict): ingest one record
    - merge(other: StreamingAggregator): merge another aggregator into this one (in-place)
    - summary() -> dict: returns {"count": int, "sum": float, "distinct": int}
    """
    def __init__(self, key: str, approximate: bool = False):
        self.key = key
        self.count = 0
        self.sum = 0.0
        self._distinct_set = set()  # exact distinct tracking
        self._approximate = approximate
        if approximate:
            raise NotImplementedError("Approximate mode not implemented; swap in HLL for production.")

    def add(self, record: Dict[str, Any]) -> None:
        if self.key not in record:
            # ignore records without the key (alternative: raise)
            return
        value = record[self.key]
        self.count += 1
        try:
            # allow numeric-like sums (int/float). Non-numeric will raise and be ignored for sum.
            self.sum += float(value)
        except (TypeError, ValueError):
            pass
        # for distinctness use hashable representation
        self._distinct_set.add(value)

    def merge(self, other: "StreamingAggregator") -> None:
        if not isinstance(other, StreamingAggregator):
            raise TypeError("merge expects StreamingAggregator")
        if self.key != other.key:
            raise ValueError("cannot merge aggregators with different keys")
        if self._approximate or other._approximate:
            raise NotImplementedError("Approximate merge not implemented in this example")
        # merge semantics: counts and sums are additive; distincts are set-union
        self.count += other.count
        self.sum += other.sum
        # union distinct sets
        self._distinct_set |= other._distinct_set

    def summary(self) -> Dict[str, Any]:
        return {
            "key": self.key,
            "count": self.count,
            "sum": self.sum,
            "distinct": len(self._distinct_set)
        }

Key points and reasoning:- add: idempotent per record; ignores records missing the key (design choice).- merge: in-place, requires same key; semantics: count and sum are additive; distinct is union (exact).- Exact distinct uses a Python set — simple, deterministic, fully mergeable.- For large-scale streaming use, replace sets with an approximate, mergeable sketch (HyperLogLog) to bound memory; merge semantics remain same (sketch-merge).Complexity:- add: average O(1) amortized for set insert and float addition.- merge: O(min(n1,n2)) or O(n_other) to union sets (cost proportional to smaller set if optimized), here Python does set union O(len(other)).- Space: O(D) where D = number of distinct values (can be large).Edge cases:- Non-numeric values for sum are ignored.- Merging different keys is disallowed to avoid silent corruption.- To support distributed aggregation, use approximate mergeable sketches (HLL) and ensure deterministic hashing.

MediumTechnical

67 practiced

You are reviewing a teammate's module and notice many small functions have duplicated parameter validation code. Propose a refactor strategy to centralize validation (decorators, validators, shared helpers), discuss backward compatibility concerns, and show a short decorator example that validates required fields.

Sample Answer

Approach: centralize validation into a reusable layer (decorators + shared validators) so business logic stays clean, tests are easier, and inconsistent checks are eliminated.

Refactor strategy:- Inventory validations: list duplicated checks (types, ranges, required fields).- Implement small, composable validator functions (e.g., is_required, is_type, in_range) in a validators module.- Create a decorator factory @validate that accepts validators per-arg or a schema dict; apply to functions rather than copy-pasting checks.- Add unit tests for validators and decorated functions.- Migrate incrementally: apply decorator to a few low-risk functions first, run tests, then expand.- Document behavior and deprecation plan for old inline checks.

Backward compatibility:- Preserve original function signatures and exceptions where practical; the decorator should raise the same exception type/message as existing checks or map to a new ValidationError class while providing shims.- Use feature flags or versioned API layers if external callers rely on exact error messages.- Keep old inline checks until decorators are fully covered by tests, then remove.

Decorator example (validates required fields in kwargs):

python

from functools import wraps

class ValidationError(ValueError):
    pass

def require_fields(*fields):
    """Decorator to ensure required kwargs are present and not None."""
    def decorator(fn):
        @wraps(fn)
        def wrapped(*args, **kwargs):
            missing = [f for f in fields if f not in kwargs or kwargs[f] is None]
            if missing:
                raise ValidationError(f"Missing required fields: {', '.join(missing)}")
            return fn(*args, **kwargs)
        return wrapped
    return decorator

# Usage
@require_fields('data', 'model')
def train_model(*, data=None, model=None, epochs=10):
    # business logic assumes data and model are valid
    ...

This approach reduces duplication, clarifies intent, and allows safe, testable incremental migration.

Unlock Full Question Bank

Get access to hundreds of Clean Code and Best Practices interview questions and detailed answers.

Join thousands of developers preparing for their dream job.