Python Architect: 50+ Junior to Mid Interview Deep-Dives (2026)

# Advanced memory management example
import sys
import gc

class ComplexObject:
    def __init__(self, data):
        self.data = data
        self.circular_ref = None
    
    def __del__(self):
        print(f"Deleting {self.data}")

# Question: Explain reference cycles and how Python handles them
def create_reference_cycle():
    obj1 = ComplexObject("A")
    obj2 = ComplexObject("B")
    obj1.circular_ref = obj2
    obj2.circular_ref = obj1  # Reference cycle created
    
    # Reference counting won't free these
    return obj1, obj2

# Interview Question: What happens here?
def memory_management_question():
    obj1, obj2 = create_reference_cycle()
    print(f"Ref count obj1: {sys.getrefcount(obj1) - 1}")
    print(f"Ref count obj2: {sys.getrefcount(obj2) - 1}")
    
    # Explicit garbage collection
    collected = gc.collect()
    print(f"Garbage collected: {collected} objects")
    
    # Check thresholds
    print(f"GC Thresholds: {gc.get_threshold()}")

from functools import wraps
from typing import Any, Callable
import time

# Question: Design a decorator factory with parameters
def retry(max_attempts: int = 3, delay: float = 1.0):
    """
    Interview Focus: Explain closure scope, *args/**kwargs, decorator chaining
    """
    def decorator(func: Callable) -> Callable:
        @wraps(func)
        def wrapper(*args, **kwargs) -> Any:
            last_exception = None
            for attempt in range(max_attempts):
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    last_exception = e
                    if attempt < max_attempts - 1:
                        time.sleep(delay * (2 ** attempt))  # Exponential backoff
                    print(f"Attempt {attempt + 1} failed: {e}")
            raise last_exception
        return wrapper
    return decorator

# Question: Create a class decorator for singleton pattern
def singleton(cls):
    instances = {}
    @wraps(cls)
    def wrapper(*args, **kwargs):
        if cls not in instances:
            instances[cls] = cls(*args, **kwargs)
        return instances[cls]
    return wrapper

# Question: Property descriptor with validation
class ValidatedAttribute:
    """Descriptor protocol deep dive"""
    def __init__(self, validator: Callable):
        self.validator = validator
        self.storage_name = None
    
    def __set_name__(self, owner, name):
        self.storage_name = name
    
    def __get__(self, obj, objtype=None):
        if obj is None:
            return self
        return getattr(obj, f"_validated_{self.storage_name}")
    
    def __set__(self, obj, value):
        if self.validator(value):
            setattr(obj, f"_validated_{self.storage_name}", value)
        else:
            raise ValueError(f"Invalid value: {value}")

class UserProfile:
    age = ValidatedAttribute(lambda x: 0 <= x <= 120)
    email = ValidatedAttribute(lambda x: "@" in str(x))

import asyncio
import concurrent.futures
from multiprocessing import Pool
import threading
import time

class ConcurrencyComparison:
    """
    Question: When would you use each approach?
    1. Threading: I/O-bound operations with GIL limitation
    2. Multiprocessing: CPU-bound operations
    3. Asyncio: High-concurrency I/O with single-threaded efficiency
    """
    
    # I/O-bound example
    def threaded_io_operations(self, urls):
        results = []
        lock = threading.Lock()
        
        def fetch_url(url):
            # Simulate network I/O
            time.sleep(0.1)
            with lock:
                results.append(f"Fetched {url}")
        
        threads = []
        for url in urls:
            thread = threading.Thread(target=fetch_url, args=(url,))
            thread.start()
            threads.append(thread)
        
        for thread in threads:
            thread.join()
        return results
    
    # CPU-bound example
    def process_cpu_operations(self, numbers):
        with Pool() as pool:
            return pool.map(self._heavy_computation, numbers)
    
    def _heavy_computation(self, n):
        return sum(i * i for i in range(n))
    
    # Async example
    async def async_io_operations(self, urls):
        async def fetch(url):
            await asyncio.sleep(0.1)
            return f"Async fetched {url}"
        
        tasks = [fetch(url) for url in urls]
        return await asyncio.gather(*tasks)

# Question: Explain GIL implications
def gil_impact_demonstration():
    """
    GIL prevents true parallel execution of Python bytecode in threads
    """
    import dis
    
    def cpu_intensive():
        return sum(x * x for x in range(10**6))
    
    # Show bytecode
    print(dis.dis(cpu_intensive))
    
    # Compare execution times
    import timeit
    
    # Single-threaded
    single_time = timeit.timeit(cpu_intensive, number=10)
    
    # Multi-threaded (GIL bottleneck)
    import threading
    
    def threaded_version():
        threads = []
        for _ in range(4):
            t = threading.Thread(target=cpu_intensive)
            t.start()
            threads.append(t)
        for t in threads:
            t.join()
    
    threaded_time = timeit.timeit(threaded_version, number=1)
    
    return {"single": single_time, "threaded": threaded_time}

# System design patterns implementation
from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import Optional
import hashlib
import base64

@dataclass
class URLData:
    original_url: str
    short_code: str
    created_at: float
    access_count: int = 0

# Strategy Pattern for different encoding
class EncodingStrategy(ABC):
    @abstractmethod
    def encode(self, url: str) -> str:
        pass

class HashEncoding(EncodingStrategy):
    def encode(self, url: str) -> str:
        hash_obj = hashlib.md5(url.encode())
        return base64.urlsafe_b64encode(hash_obj.digest()[:6]).decode()

class CounterEncoding(EncodingStrategy):
    def __init__(self):
        self.counter = 0
    
    def encode(self, url: str) -> str:
        self.counter += 1
        return base64.urlsafe_b64encode(str(self.counter).encode()).decode()

# Repository Pattern for data access
class URLRepository:
    def __init__(self):
        self.url_by_code = {}
        self.url_by_hash = {}
    
    def save(self, url_data: URLData) -> None:
        self.url_by_code[url_data.short_code] = url_data
        self.url_by_hash[hash(url_data.original_url)] = url_data
    
    def find_by_code(self, code: str) -> Optional[URLData]:
        return self.url_by_code.get(code)
    
    def find_by_url(self, url: str) -> Optional[URLData]:
        return self.url_by_hash.get(hash(url))

# Cache Layer with LRU strategy
from collections import OrderedDict

class LRUCache:
    def __init__(self, capacity: int = 1000):
        self.cache = OrderedDict()
        self.capacity = capacity
    
    def get(self, key: str) -> Optional[URLData]:
        if key not in self.cache:
            return None
        self.cache.move_to_end(key)
        return self.cache[key]
    
    def put(self, key: str, value: URLData) -> None:
        if key in self.cache:
            self.cache.move_to_end(key)
        self.cache[key] = value
        if len(self.cache) > self.capacity:
            self.cache.popitem(last=False)

# Service Layer
class URLShorteningService:
    def __init__(self, encoding_strategy: EncodingStrategy):
        self.repository = URLRepository()
        self.cache = LRUCache()
        self.encoding_strategy = encoding_strategy
    
    def shorten(self, original_url: str) -> str:
        # Check cache first
        cached = self.cache.get(hash(original_url))
        if cached:
            return cached.short_code
        
        # Check existing
        existing = self.repository.find_by_url(original_url)
        if existing:
            return existing.short_code
        
        # Create new
        short_code = self.encoding_strategy.encode(original_url)
        url_data = URLData(
            original_url=original_url,
            short_code=short_code,
            created_at=time.time()
        )
        
        self.repository.save(url_data)
        self.cache.put(hash(original_url), url_data)
        
        return short_code
    
    def resolve(self, short_code: str) -> Optional[str]:
        # Cache lookup
        cached = self.cache.get(short_code)
        if cached:
            cached.access_count += 1
            return cached.original_url
        
        # Repository lookup
        url_data = self.repository.find_by_code(short_code)
        if url_data:
            url_data.access_count += 1
            self.cache.put(short_code, url_data)
            return url_data.original_url
        
        return None

# SQLAlchemy optimization patterns
from sqlalchemy import create_engine, Column, Integer, String, ForeignKey
from sqlalchemy.orm import relationship, sessionmaker, joinedload, selectinload
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

class User(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True)
    name = Column(String(50))
    # Lazy loading by default
    posts = relationship("Post", back_populates="author")

class Post(Base):
    __tablename__ = 'posts'
    id = Column(Integer, primary_key=True)
    title = Column(String(100))
    user_id = Column(Integer, ForeignKey('users.id'))
    author = relationship("User", back_populates="posts")
    comments = relationship("Comment", back_populates="post")

class Comment(Base):
    __tablename__ = 'comments'
    id = Column(Integer, primary_key=True)
    content = Column(String(500))
    post_id = Column(Integer, ForeignKey('posts.id'))
    post = relationship("Post", back_populates="comments")

# N+1 Problem Example
def n_plus_one_problem(session):
    """Inefficient: Makes 1 query for users + N queries for posts"""
    users = session.query(User).all()
    for user in users:
        print(f"User: {user.name}, Posts: {len(user.posts)}")  # New query each time!

# Solution 1: Joined Load
def optimized_joined_load(session):
    """Single query with JOINs"""
    users = session.query(User).options(
        joinedload(User.posts).joinedload(Post.comments)
    ).all()
    return users

# Solution 2: Select IN Load
def optimized_selectin_load(session):
    """Two optimized queries"""
    users = session.query(User).options(
        selectinload(User.posts).selectinload(Post.comments)
    ).all()
    return users

# Solution 3: Bulk queries
def optimized_bulk_queries(session):
    """Manual optimization"""
    users = session.query(User).all()
    user_ids = [user.id for user in users]
    
    # Single query for all posts
    from sqlalchemy.orm import aliased
    posts = session.query(Post).filter(Post.user_id.in_(user_ids)).all()
    
    # Group posts by user
    posts_by_user = {}
    for post in posts:
        posts_by_user.setdefault(post.user_id, []).append(post)
    
    for user in users:
        user._posts = posts_by_user.get(user.id, [])
    
    return users

# Database indexing strategies
class IndexingStrategies:
    """
    Interview Questions:
    1. When to use composite indexes?
    2. Covering indexes vs regular indexes
    3. Index selectivity and cardinality
    """
    
    def create_optimal_indexes(self):
        # Composite index for WHERE + ORDER BY
        index_sql = """
        CREATE INDEX idx_user_status_created ON users 
        (status, created_at DESC) 
        WHERE status = 'active';
        """
        
        # Partial index for specific queries
        partial_index = """
        CREATE INDEX idx_active_users ON users (email) 
        WHERE is_active = TRUE;
        """
        
        return {
            "composite": index_sql,
            "partial": partial_index
        }

from contextlib import contextmanager
import psycopg2
from typing import Iterator

# Question: Implement database connection pool with context managers
class ConnectionPool:
    def __init__(self, max_connections=10):
        self.max_connections = max_connections
        self._connections = []
        self._semaphore = threading.Semaphore(max_connections)
    
    @contextmanager
    def get_connection(self) -> Iterator[psycopg2.extensions.connection]:
        """Interview focus: Exception safety in context managers"""
        self._semaphore.acquire()
        conn = None
        try:
            if self._connections:
                conn = self._connections.pop()
            else:
                conn = psycopg2.connect("dbname=test")
            yield conn
            # Successful execution
            self._connections.append(conn)
        except Exception as e:
            if conn:
                conn.close()
            raise e
        finally:
            self._semaphore.release()
    
    def __enter__(self):
        return self
    
    def __exit__(self, exc_type, exc_val, exc_tb):
        for conn in self._connections:
            conn.close()
        self._connections.clear()

# Async context manager example
import aiohttp
from contextlib import asynccontextmanager

@asynccontextmanager
async def async_session_manager():
    """Modern Python 3.10+ async patterns"""
    session = aiohttp.ClientSession()
    try:
        yield session
    finally:
        await session.close()

import numpy as np
from functools import lru_cache
from numba import jit
import cython  # For type hints in Cython

# Question: Optimize numerical computation
class NumericalOptimization:
    
    # Pure Python (slow)
    def slow_matrix_multiply(self, a, b):
        size = len(a)
        result = [[0] * size for _ in range(size)]
        for i in range(size):
            for j in range(size):
                for k in range(size):
                    result[i][j] += a[i][k] * b[k][j]
        return result
    
    # NumPy optimized (vectorized)
    def numpy_matrix_multiply(self, a, b):
        return np.dot(a, b)
    
    # Numba JIT compilation
    @jit(nopython=True, parallel=True)
    def numba_matrix_multiply(self, a, b):
        size = a.shape[0]
        result = np.zeros((size, size))
        for i in range(size):
            for j in range(size):
                for k in range(size):
                    result[i, j] += a[i, k] * b[k, j]
        return result
    
    # Memory view optimization (Cython-like)
    def memory_efficient_operation(self):
        import array
        arr = array.array('d', [1.0, 2.0, 3.0, 4.0])
        # Memory view for zero-copy operations
        mem_view = memoryview(arr)
        return sum(mem_view)

# Caching strategies comparison
class CachingStrategies:
    
    @lru_cache(maxsize=128)
    def fibonacci_lru(self, n):
        if n < 2:
            return n
        return self.fibonacci_lru(n-1) + self.fibonacci_lru(n-2)
    
    def custom_cache_decorator(self):
        """Implement custom caching with TTL"""
        import time
        
        def decorator(func):
            cache = {}
            
            def wrapper(*args, **kwargs):
                key = str(args) + str(kwargs)
                if key in cache:
                    value, timestamp = cache[key]
                    if time.time() - timestamp < 3600:  # 1 hour TTL
                        return value
                result = func(*args, **kwargs)
                cache[key] = (result, time.time())
                return result
            
            return wrapper
        return decorator

import pytest
from unittest.mock import Mock, patch, MagicMock
from hypothesis import given, strategies as st

# Property-based testing
class TestBankAccount:
    
    @given(st.integers(min_value=0, max_value=10000))
    def test_deposit_never_negative(self, amount):
        account = BankAccount()
        account.deposit(amount)
        assert account.balance >= 0
    
    @given(
        st.integers(min_value=0, max_value=1000),
        st.integers(min_value=0, max_value=1000)
    )
    def test_commutative_deposits(self, a, b):
        account1 = BankAccount()
        account2 = BankAccount()
        
        account1.deposit(a)
        account1.deposit(b)
        
        account2.deposit(b)
        account2.deposit(a)
        
        assert account1.balance == account2.balance

# Mock patterns
class TestExternalService:
    
    @patch('requests.get')
    def test_api_call_with_retry(self, mock_get):
        # Setup mock responses
        mock_get.side_effect = [
            ConnectionError(),
            MagicMock(status_code=200, json=lambda: {'data': 'test'})
        ]
        
        service = ExternalService()
        result = service.fetch_with_retry()
        
        assert result == {'data': 'test'}
        assert mock_get.call_count == 2
    
    def test_async_mocking(self):
        """Mock async context managers"""
        mock_session = AsyncMock()
        mock_session.get.return_value.__aenter__.return_value.json = AsyncMock(
            return_value={'status': 'ok'}
        )
        
        with patch('aiohttp.ClientSession', return_value=mock_session):
            service = AsyncService()
            result = asyncio.run(service.fetch_data())
            
            assert result['status'] == 'ok'

# Docker optimization for Python
DOCKERFILE_TEMPLATE = """
# Multi-stage build for production
FROM python:3.10-slim as builder

WORKDIR /app
COPY requirements.txt .
RUN pip install --user --no-warn-script-location -r requirements.txt

FROM python:3.10-slim as runtime
WORKDIR /app

# Copy only necessary files
COPY --from=builder /root/.local /root/.local
COPY app.py .

# Non-root user for security
RUN useradd -m -u 1000 appuser && chown -R appuser:appuser /app
USER appuser

# Environment optimization
ENV PYTHONUNBUFFERED=1 \
    PYTHONDONTWRITEBYTECODE=1 \
    PATH="/root/.local/bin:${PATH}"

CMD ["python", "app.py"]
"""

# Kubernetes deployment patterns
K8S_DEPLOYMENT = """
apiVersion: apps/v1
kind: Deployment
metadata:
  name: python-api
spec:
  replicas: 3
  strategy:
    rollingUpdate:
      maxSurge: 1
      maxUnavailable: 0
  selector:
    matchLabels:
      app: python-api
  template:
    metadata:
      labels:
        app: python-api
    spec:
      containers:
      - name: api
        image: python-api:latest
        ports:
        - containerPort: 8000
        resources:
          requests:
            memory: "256Mi"
            cpu: "100m"
          limits:
            memory: "512Mi"
            cpu: "500m"
        livenessProbe:
          httpGet:
            path: /health
            port: 8000
          initialDelaySeconds: 30
          periodSeconds: 10
        readinessProbe:
          httpGet:
            path: /ready
            port: 8000
          initialDelaySeconds: 5
          periodSeconds: 5
"""

# Structural Pattern Matching (Python 3.10+)
class HTTPResponseHandler:
    
    def handle_response(self, response):
        match response:
            case {'status': 200, 'data': data}:
                return self.process_data(data)
            
            case {'status': 404}:
                raise NotFoundError("Resource not found")
            
            case {'status': 500, 'error': error}:
                raise ServerError(f"Server error: {error}")
            
            case {'status': status} if 400 <= status < 500:
                raise ClientError(f"Client error: {status}")
            
            case _:
                raise UnknownError("Unexpected response")

# Type hint improvements (Python 3.10+)
from typing import TypeAlias, ParamSpec, Concatenate
from collections.abc import Callable

ResponseType: TypeAlias = dict[str, str | int | list]

P = ParamSpec("P")

def logged_api_call(func: Callable[Concatenate[str, P], ResponseType]) -> Callable[P, ResponseType]:
    def wrapper(*args: P.args, **kwargs: P.kwargs) -> ResponseType:
        print(f"Calling {func.__name__}")
        return func("api-key", *args, **kwargs)
    return wrapper

# Parenthesized context managers (Python 3.10+)
def multiple_context_managers():
    with (
        open('file1.txt') as f1,
        open('file2.txt') as f2,
        suppress(FileNotFoundError)
    ):
        data1 = f1.read()
        data2 = f2.read()
        return data1 + data2

class SecureCoding:
    
    def prevent_sql_injection(self):
        """Interview question: How to prevent SQL injection?"""
        
        # UNSAFE - String concatenation
        unsafe_query = f"SELECT * FROM users WHERE name = '{user_input}'"
        
        # SAFE - Parameterized queries (SQLAlchemy)
        from sqlalchemy import text
        safe_query = text("SELECT * FROM users WHERE name = :name")
        result = session.execute(safe_query, {"name": user_input})
        
        # SAFE - ORM query
        safe_orm = session.query(User).filter(User.name == user_input)
        
        return {
            "unsafe": unsafe_query,
            "safe_parameterized": safe_query,
            "safe_orm": safe_orm
        }
    
    def secure_password_handling(self):
        """Password hashing best practices"""
        import bcrypt
        import secrets
        
        # Generate secure password
        password = secrets.token_urlsafe(32)
        
        # Hash with salt
        salt = bcrypt.gensalt(rounds=12)
        hashed = bcrypt.hashpw(password.encode(), salt)
        
        # Verify
        is_valid = bcrypt.checkpw(password.encode(), hashed)
        
        return {
            "password": password,
            "hashed": hashed,
            "is_valid": is_valid
        }
    
    def prevent_deserialization_attacks(self):
        """Safe deserialization patterns"""
        import pickle
        
        # UNSAFE
        unsafe_data = pickle.loads(user_input)
        
        # SAFE - Use JSON or safer alternatives
        import json
        safe_data = json.loads(user_input)
        
        # SAFE - Restricted unpickling
        class RestrictedUnpickler(pickle.Unpickler):
            def find_class(self, module, name):
                # Only allow safe classes
                allowed_modules = {'__main__', 'datetime'}
                if module not in allowed_modules:
                    raise pickle.UnpicklingError(f"Unsafe module: {module}")
                return super().find_class(module, name)
        
        safe_unpickled = RestrictedUnpickler(io.BytesIO(user_input)).load()
        
        return {
            "unsafe": unsafe_data,
            "json_safe": safe_data,
            "restricted_pickle": safe_unpickled
        }