Python Descriptors — Missing Instance Dict Shared State Bug

# Demonstrate the data vs non-data descriptor priority difference

class DataDescriptor:
    """Defines both __get__ and __set__ — always wins over instance dict."""

    def __set_name__(self, owner_class, attribute_name):
        # Called automatically when the class body is processed (Python 3.6+)
        # Gives us the attribute name without needing to pass it manually
        self.storage_key = f'_dd_{attribute_name}'

    def __get__(self, instance, owner_class):
        if instance is None:
            # Accessed on the class itself, not an instance — return the descriptor
            return self
        return instance.__dict__.get(self.storage_key, 'NOT SET')

    def __set__(self, instance, value):
        print(f'  [DataDescriptor.__set__] storing {value!r}')
        instance.__dict__[self.storage_key] = value


class NonDataDescriptor:
    """Only defines __get__ — instance dict takes priority over this."""

    def __get__(self, instance, owner_class):
        if instance is None:
            return self
        return 'value from NonDataDescriptor'


class Experiment:
    data_attr    = DataDescriptor()     # data descriptor
    nondata_attr = NonDataDescriptor()  # non-data descriptor


experiment = Experiment()

# --- Data descriptor priority demo ---
print('=== Data Descriptor ===')
experiment.data_attr = 'hello'          # triggers __set__
# Now manually jam a value into __dict__ under the descriptor's storage key
# The descriptor stores under '_dd_data_attr', but let's try the PUBLIC name
experiment.__dict__['data_attr'] = 'sneaky direct write'
print('instance.__dict__["data_attr"] =', experiment.__dict__.get('data_attr'))
print('experiment.data_attr =', experiment.data_attr)  # descriptor still wins

# --- Non-data descriptor priority demo ---
print('\n=== Non-Data Descriptor ===')
print('Before shadowing:', experiment.nondata_attr)     # from descriptor
experiment.__dict__['nondata_attr'] = 'instance dict wins'
print('After shadowing: ', experiment.nondata_attr)     # instance dict wins now

# --- Accessing descriptor on the class (instance=None path) ---
print('\n=== Class-level access ===')
print('Experiment.data_attr:', Experiment.data_attr)    # returns descriptor itself

from typing import Any, Type, Optional


class TypeValidated:
    """
    A reusable data descriptor that enforces type and optional numeric bounds.
    Store one instance as a class attribute; Python handles the per-instance
    data automatically via instance.__dict__.
    """

    def __init__(
        self,
        expected_type: Type,
        min_value: Optional[float] = None,
        max_value: Optional[float] = None,
    ):
        self.expected_type = expected_type
        self.min_value     = min_value
        self.max_value     = max_value
        self.storage_key   = None  # filled in by __set_name__

    def __set_name__(self, owner_class: Type, attribute_name: str):
        # Python calls this automatically during class body execution.
        # We prefix with '_tv_' to avoid colliding with the public name.
        self.public_name = attribute_name
        self.storage_key = f'_tv_{attribute_name}'

    def __get__(self, instance: Any, owner_class: Type):
        if instance is None:
            return self  # class-level access returns the descriptor
        # Retrieve from instance.__dict__; raise AttributeError if not yet set
        try:
            return instance.__dict__[self.storage_key]
        except KeyError:
            raise AttributeError(
                f'{owner_class.__name__}.{self.public_name} has not been set'
            )

    def __set__(self, instance: Any, value: Any):
        # --- Type check ---
        if not isinstance(value, self.expected_type):
            raise TypeError(
                f'{self.public_name} expects {self.expected_type.__name__}, '
                f'got {type(value).__name__} instead'
            )
        # --- Range check (only for numeric types) ---
        if self.min_value is not None and value < self.min_value:
            raise ValueError(
                f'{self.public_name} must be >= {self.min_value}, got {value}'
            )
        if self.max_value is not None and value > self.max_value:
            raise ValueError(
                f'{self.public_name} must be <= {self.max_value}, got {value}'
            )
        instance.__dict__[self.storage_key] = value

    def __delete__(self, instance: Any):
        # Removing the key from instance dict effectively 'unsets' the attribute
        instance.__dict__.pop(self.storage_key, None)
        print(f'  [TypeValidated] {self.public_name} deleted from instance')


class WeatherReading:
    """
    A single descriptor class powers three completely different validated
    attributes. No code duplication, no @property boilerplate per attribute.
    """
    temperature_celsius = TypeValidated(float, min_value=-89.2, max_value=56.7)
    humidity_percent    = TypeValidated(float, min_value=0.0,   max_value=100.0)
    station_id          = TypeValidated(str)

    def __init__(self, station_id: str, temperature: float, humidity: float):
        self.station_id          = station_id
        self.temperature_celsius = temperature
        self.humidity_percent    = humidity

    def __repr__(self):
        return (
            f'WeatherReading(station={self.station_id!r}, '
            f'temp={self.temperature_celsius}°C, '
            f'humidity={self.humidity_percent}%)'
        )


# --- Happy path ---
reading = WeatherReading(station_id='LOND-001', temperature=21.5, humidity=65.0)
print('Created:', reading)

# --- Mutation works and is validated ---
reading.temperature_celsius = -10.0
print('Updated temp:', reading)

# --- Wrong type ---
try:
    reading.temperature_celsius = '22 degrees'  # string, not float
except TypeError as e:
    print('TypeError caught:', e)

# --- Out of range ---
try:
    reading.humidity_percent = 150.0
except ValueError as e:
    print('ValueError caught:', e)

# --- Delete the attribute ---
del reading.temperature_celsius
try:
    print(reading.temperature_celsius)
except AttributeError as e:
    print('AttributeError caught:', e)

# --- Check instance __dict__ — notice the _tv_ prefixed keys ---
print('\nInstance __dict__:', reading.__dict__)

# Re-implement Python's built-in property as a pure descriptor.
# This is essentially what CPython's property does in C — same logic.

class managed_property:
    """
    A pure-Python reimplementation of the built-in property descriptor.
    Supports getter, setter, and deleter chaining just like @property.
    """

    def __init__(self, fget=None, fset=None, fdel=None, doc=None):
        self.fget = fget
        self.fset = fset
        self.fdel = fdel
        # Mimic property's behaviour: use getter's docstring if none given
        self.__doc__ = doc or (fget.__doc__ if fget else None)

    def __get__(self, instance, owner_class):
        if instance is None:
            return self  # class-level access
        if self.fget is None:
            raise AttributeError('unreadable attribute')
        return self.fget(instance)  # call the getter with the instance

    def __set__(self, instance, value):
        if self.fset is None:
            raise AttributeError("can't set attribute — no setter defined")
        self.fset(instance, value)  # call the setter

    def __delete__(self, instance):
        if self.fdel is None:
            raise AttributeError("can't delete attribute — no deleter defined")
        self.fdel(instance)

    # These mirror @property.setter / @property.deleter chaining
    def setter(self, fset):
        # Return a NEW managed_property with fset filled in; fget is preserved
        return type(self)(self.fget, fset, self.fdel, self.__doc__)

    def deleter(self, fdel):
        return type(self)(self.fget, self.fset, fdel, self.__doc__)


# ----- Reimplemented classmethod for completeness -----

class managed_classmethod:
    """Non-data descriptor that binds the class, not the instance."""

    def __init__(self, func):
        self.func = func

    def __get__(self, instance, owner_class):
        if owner_class is None:
            owner_class = type(instance)
        # Return a callable with owner_class pre-bound as first argument
        def bound_class_method(*args, **kwargs):
            return self.func(owner_class, *args, **kwargs)
        return bound_class_method


# ----- Demo: use managed_property like @property -----

class BankAccount:
    def __init__(self, owner: str, initial_balance: float):
        self.owner   = owner
        self._balance = initial_balance  # raw storage

    @managed_property
    def balance(self) -> float:
        """Current account balance in GBP."""
        print('  [getter called]')
        return self._balance

    @balance.setter
    def balance(self, amount: float):
        print(f'  [setter called with {amount}]')
        if amount < 0:
            raise ValueError('Balance cannot go negative')
        self._balance = amount

    @balance.deleter
    def balance(self):
        print('  [deleter called — closing account]')
        self._balance = 0.0

    @managed_classmethod
    def open_zero_balance(cls, owner: str) -> 'BankAccount':
        return cls(owner, 0.0)


account = BankAccount('Alice', 1000.0)
print('Balance:', account.balance)

account.balance = 1500.0
print('New balance:', account.balance)

try:
    account.balance = -50
except ValueError as e:
    print('Caught:', e)

del account.balance
print('After deletion:', account.balance)

# classmethod via descriptor
empty_account = BankAccount.open_zero_balance('Bob')
print('\nBob account balance:', empty_account.balance)

import time
import threading
from functools import cached_property


# ============================================================
# 1. PERFORMANCE BENCHMARK — raw dict vs property vs descriptor
# ============================================================

class DirectDict:
    def __init__(self):
        self.radius = 5.0  # plain attribute — stored in __dict__


class WithProperty:
    def __init__(self):
        self._radius = 5.0

    @property
    def radius(self):
        return self._radius


class RadiusDescriptor:
    def __get__(self, instance, owner):
        if instance is None:
            return self
        return instance.__dict__.get('_radius', 0.0)

    def __set__(self, instance, value):
        instance.__dict__['_radius'] = value


class WithDescriptor:
    radius = RadiusDescriptor()

    def __init__(self):
        self.radius = 5.0


READS = 2_000_000

for label, obj in [('Direct dict', DirectDict()), ('@property', WithProperty()), ('Custom descriptor', WithDescriptor())]:
    start = time.perf_counter()
    for _ in range(READS):
        _ = obj.radius
    elapsed = time.perf_counter() - start
    print(f'{label:<22} {READS:,} reads in {elapsed:.3f}s  ({elapsed/READS*1e9:.1f} ns/read)')


# ============================================================
# 2. THREAD-SAFE LAZY DESCRIPTOR
#    Use case: expensive resource initialised once per instance
# ============================================================

class ThreadSafeLazy:
    """
    A thread-safe lazy descriptor using a per-instance lock.
    Suitable for expensive __init__ work (DB connections, parsing, etc.)
    cached_property is NOT thread-safe; this one is.
    """

    def __set_name__(self, owner, name):
        self.public_name  = name
        self.cache_key    = f'_lazy_{name}'
        self.lock_key     = f'_lazy_lock_{name}'

    def __init__(self, factory):
        self.factory = factory  # callable that produces the expensive value

    def __get__(self, instance, owner):
        if instance is None:
            return self
        # Ensure a per-instance, per-attribute lock exists
        if self.lock_key not in instance.__dict__:
            instance.__dict__[self.lock_key] = threading.Lock()
        lock = instance.__dict__[self.lock_key]

        # Double-checked locking pattern
        if self.cache_key not in instance.__dict__:
            with lock:
                if self.cache_key not in instance.__dict__:  # second check inside lock
                    print(f'  [ThreadSafeLazy] computing {self.public_name}...')
                    instance.__dict__[self.cache_key] = self.factory(instance)

        return instance.__dict__[self.cache_key]

    # No __set__ — this is intentionally a non-data descriptor so instance dict wins
    # after first computation. If you need to invalidate, add __set__/__delete__.


class ReportGenerator:
    def __init__(self, data_source: str):
        self.data_source = data_source

    @ThreadSafeLazy
    def processed_data(self):
        """Simulate an expensive data processing step."""
        time.sleep(0.1)  # pretend this is a slow DB query
        return [f'row_{i}' for i in range(1000)]

    @ThreadSafeLazy
    def summary_stats(self):
        """Computed from processed_data — also lazy."""
        return {'count': len(self.processed_data), 'source': self.data_source}


report = ReportGenerator('warehouse_db')
print('\nFirst access (computes):')
print('Row count:', len(report.processed_data))

print('\nSecond access (cached — no recompute):')
print('Row count:', len(report.processed_data))

print('\nSummary:', report.summary_stats)

# Thread safety test — 10 threads race to initialise the same attribute
results = []

def read_data():
    results.append(id(report.processed_data))  # should all be the same object

threads = [threading.Thread(target=read_data) for _ in range(10)]
for t in threads: t.start()
for t in threads: t.join()

print('\nAll threads got same object?', len(set(results)) == 1)

import threading

# Trap 1: Multiple inheritance MRO confusion

class DescriptorA:
    def __get__(self, instance, owner):
        return 'A'

class DescriptorB:
    def __get__(self, instance, owner):
        return 'B'

class BaseA:
    attr = DescriptorA()

class BaseB:
    attr = DescriptorB()

class Child(BaseA, BaseB):
    pass

# Which one wins? Depends on MRO: Child -> BaseA -> BaseB
print('Child.attr:', Child.attr)  # Returns A
print('MRO:', [c.__name__ for c in Child.__mro__])

# Fix: explicitly declare on child
class FixedChild(BaseA, BaseB):
    attr = DescriptorA()  # explicit choice

# Trap 2: Circular lazy loading

_recursion_depth = threading.local()

class SafeLazy:
    def __set_name__(self, owner, name):
        self.name = name
        self.cache_key = '_' + name

    def __init__(self, factory):
        self.factory = factory

    def __get__(self, instance, owner):
        if instance is None:
            return self
        if not hasattr(_recursion_depth, 'depth'):
            _recursion_depth.depth = 0
        if _recursion_depth.depth > 2:
            raise RecursionError(f"Circular dependency detected for {self.name}")
        if self.cache_key not in instance.__dict__:
            _recursion_depth.depth += 1
            try:
                instance.__dict__[self.cache_key] = self.factory(instance)
            finally:
                _recursion_depth.depth -= 1
        return instance.__dict__[self.cache_key]


class CircularDemo:
    @SafeLazy
    def first(self):
        return self.second + '-first'

    @SafeLazy
    def second(self):
        return self.first + '-second'

obj = CircularDemo()
try:
    print(obj.first)
except RecursionError as e:
    print('Caught recursion:', e)

# Trap 3: Descriptor without __repr__ makes debugging hard

class GoodDescriptor:
    def __set_name__(self, owner, name):
        self.name = name
    def __get__(self, instance, owner):
        return 42
    def __repr__(self):
        return f'<GoodDescriptor for {self.name}>'

class WithGood:
    price = GoodDescriptor()

print('repr:', WithGood.price)  # Shows name

// io.thecodeforge
import time
from django.db import models

class User(models.Model):
    name = models.CharField(max_length=100)

# Simulating 10k accesses — each triggers descriptor __get__
users = [User(name=f"user_{i}") for i in range(10000)]

start = time.perf_counter()
for u in users:
    _ = u.name  # Descriptor __get__: checks cache, returns str
end = time.perf_counter()

print(f"10k descriptor accesses: {end - start:.4f}s")

// io.thecodeforge
import copy

class LockedAttr:
    def __set_name__(self, owner, name):
        self.storage = f"_{name}_locked"
    
    def __get__(self, obj, objtype=None):
        if obj is None:
            return self
        return getattr(obj, self.storage, None)
    
    def __set__(self, obj, value):
        if hasattr(obj, self.storage):
            raise AttributeError(f"{self.storage[1:]} is read-only")
        object.__setattr__(obj, self.storage, value)

class Config:
    api_key = LockedAttr()
    
    def __init__(self, key):
        self.api_key = key  # Works once

cfg = Config("sk-abc")
try:
    cfg.api_key = "hacked"
except AttributeError as e:
    print(e)  # "api_key is read-only"

# Copy still works
cfg2 = copy.deepcopy(cfg)
print(cfg2.api_key)  # 'sk-abc'