Source code
from __future__ import annotations
import abc
import contextlib
import enum
import itertools
import linecache
import sys
import types
import unicodedata
import weakref
from collections.abc import Callable, Mapping
from functools import cached_property
from typing import Any, NamedTuple, TypeVar
from . import _compat, _config, setters
from ._compat import (
PY_3_11_PLUS,
PY_3_13_PLUS,
_AnnotationExtractor,
_get_annotations,
_lazy_is_generator,
get_generic_base,
)
from .exceptions import (
DefaultAlreadySetError,
FrozenInstanceError,
NotAnAttrsClassError,
UnannotatedAttributeError,
)
class _ClassBuilder:
"""
Iteratively build *one* class.
"""
__slots__ = (
"_add_method_dunders",
"_attr_names",
"_attrs",
"_base_attr_map",
"_base_names",
"_cache_hash",
"_cls",
"_cls_dict",
"_delete_attribs",
"_frozen",
"_has_custom_setattr",
"_has_post_init",
"_has_pre_init",
"_is_exc",
"_on_setattr",
"_pre_init_has_args",
"_repr_added",
"_script_snippets",
"_slots",
"_weakref_slot",
"_wrote_own_setattr",
)
def __init__(
self,
cls: type,
these,
auto_attribs: bool,
props: ClassProps,
has_custom_setattr: bool,
):
attrs, base_attrs, base_map = _transform_attrs(
cls,
these,
auto_attribs,
props.kw_only,
props.collected_fields_by_mro,
props.field_transformer,
)
self._cls = cls
self._cls_dict = dict(cls.__dict__) if props.is_slotted else {}
self._attrs = attrs
self._base_names = {a.name for a in base_attrs}
self._base_attr_map = base_map
self._attr_names = tuple(a.name for a in attrs)
self._slots = props.is_slotted
self._frozen = props.is_frozen
self._weakref_slot = props.has_weakref_slot
self._cache_hash = (
props.hashability is ClassProps.Hashability.HASHABLE_CACHED
)
self._has_pre_init = bool(getattr(cls, "__attrs_pre_init__", False))
self._pre_init_has_args = False
if self._has_pre_init:
# Check if the pre init method has more arguments than just `self`
# We want to pass arguments if pre init expects arguments
import inspect
pre_init_func = cls.__attrs_pre_init__
pre_init_signature = inspect.signature(pre_init_func)
self._pre_init_has_args = len(pre_init_signature.parameters) > 1
self._has_post_init = bool(getattr(cls, "__attrs_post_init__", False))
self._delete_attribs = not bool(these)
self._is_exc = props.is_exception
self._on_setattr = props.on_setattr_hook
self._has_custom_setattr = has_custom_setattr
self._wrote_own_setattr = False
self._cls_dict["__attrs_attrs__"] = self._attrs
self._cls_dict["__attrs_props__"] = props
if props.is_frozen:
self._cls_dict["__setattr__"] = _frozen_setattrs
self._cls_dict["__delattr__"] = _frozen_delattrs
self._wrote_own_setattr = True
elif self._on_setattr in (
_DEFAULT_ON_SETATTR,
setters.validate,
setters.convert,
):
has_validator = has_converter = False
loopIterator = __import__("builtins").iter(attrs)
while True:
try:
a = __import__("builtins").next(loopIterator)
except __import__("builtins").StopIteration:
break
if a.validator is not None:
has_validator = True
if a.converter is not None:
has_converter = True
if has_validator and has_converter:
break
if (
(
self._on_setattr == _DEFAULT_ON_SETATTR
and not (has_validator or has_converter)
)
or (self._on_setattr == setters.validate and not has_validator)
or (self._on_setattr == setters.convert and not has_converter)
):
# If class-level on_setattr is set to convert + validate, but
# there's no field to convert or validate, pretend like there's
# no on_setattr.
self._on_setattr = None
if props.added_pickling:
(
self._cls_dict["__getstate__"],
self._cls_dict["__setstate__"],
) = self._make_getstate_setstate()
# tuples of script, globs, hook
self._script_snippets: list[
tuple[str, dict, Callable[[dict, dict], Any]]
] = []
self._repr_added = False
# We want to only do this check once; in 99.9% of cases these
# exist.
if not hasattr(self._cls, "__module__") or not hasattr(
self._cls, "__qualname__"
):
self._add_method_dunders = self._add_method_dunders_safe
else:
self._add_method_dunders = self._add_method_dunders_unsafe
def __repr__(self):
return f"<_ClassBuilder(cls={self._cls.__name__})>"
def _eval_snippets(self) -> None:
"""
Evaluate any registered snippets in one go.
"""
script = "\n".join([snippet[0] for snippet in self._script_snippets])
globs = {}
loopIteratorA = __import__("builtins").iter(self._script_snippets)
while True:
try:
_, snippet_globs, _ = __import__("builtins").next(loopIteratorA)
except __import__("builtins").StopIteration:
break
globs.update(snippet_globs)
locs = _linecache_and_compile(
script,
_generate_unique_filename(self._cls, "methods"),
globs,
)
loopIteratorB = __import__("builtins").iter(self._script_snippets)
while True:
try:
_, _, hook = __import__("builtins").next(loopIteratorB)
except __import__("builtins").StopIteration:
break
hook(self._cls_dict, locs)
def build_class(self):
"""
Finalize class based on the accumulated configuration.
Builder cannot be used after calling this method.
"""
self._eval_snippets()
if self._slots is True:
cls = self._create_slots_class()
self._cls.__attrs_base_of_slotted__ = weakref.ref(cls)
else:
cls = abc.update_abstractmethods(self._patch_original_class())
# The method gets only called if it's not inherited from a base class.
# _has_own_attribute does NOT work properly for classmethods.
if (
getattr(cls, "__attrs_init_subclass__", None)
and "__attrs_init_subclass__" not in cls.__dict__
):
cls.__attrs_init_subclass__()
return cls
def _patch_original_class(self):
"""
Apply accumulated methods and return the class.
"""
cls = self._cls
base_names = self._base_names
# Clean class of attribute definitions (`attr.ib()`s).
if self._delete_attribs:
for name in self._attr_names:
if (
name not in base_names
and getattr(cls, name, _SENTINEL) is not _SENTINEL
):
# An AttributeError can happen if a base class defines a
# class variable and we want to set an attribute with the
# same name by using only a type annotation.
with contextlib.suppress(AttributeError):
delattr(cls, name)
# Attach our dunder methods.
loopIteratorC = __import__("builtins").iter(self._cls_dict.items())
while True:
try:
name, value = __import__("builtins").next(loopIteratorC)
except __import__("builtins").StopIteration:
break
setattr(cls, name, value)
# If we've inherited an attrs __setattr__ and don't write our own,
# reset it to object's.
if not self._wrote_own_setattr and getattr(
cls, "__attrs_own_setattr__", False
):
cls.__attrs_own_setattr__ = False
if not self._has_custom_setattr:
cls.__setattr__ = _OBJ_SETATTR
return cls
def _create_slots_class(self):
"""
Build and return a new class with a `__slots__` attribute.
"""
cd = {
k: v
for k, v in self._cls_dict.items()
if k not in (*tuple(self._attr_names), "__dict__", "__weakref__")
}
# 3.14.0rc2+
if hasattr(sys, "_clear_type_descriptors"):
sys._clear_type_descriptors(self._cls)
# If our class doesn't have its own implementation of __setattr__
# (either from the user or by us), check the bases, if one of them has
# an attrs-made __setattr__, that needs to be reset. We don't walk the
# MRO because we only care about our immediate base classes.
# XXX: This can be confused by subclassing a slotted attrs class with
# XXX: a non-attrs class and subclass the resulting class with an attrs
# XXX: class. See `test_slotted_confused` for details. For now that's
# XXX: OK with us.
if not self._wrote_own_setattr:
cd["__attrs_own_setattr__"] = False
if not self._has_custom_setattr:
loopIteratorD = __import__("builtins").iter(self._cls.__bases__)
while True:
try:
base_cls = __import__("builtins").next(loopIteratorD)
except __import__("builtins").StopIteration:
break
if base_cls.__dict__.get("__attrs_own_setattr__", False):
cd["__setattr__"] = _OBJ_SETATTR
break
# Traverse the MRO to collect existing slots
# and check for an existing __weakref__.
existing_slots = {}
weakref_inherited = False
loopIteratorE = __import__("builtins").iter(self._cls.__mro__[1:-1])
while True:
try:
base_cls = __import__("builtins").next(loopIteratorE)
except __import__("builtins").StopIteration:
break
if base_cls.__dict__.get("__weakref__", None) is not None:
weakref_inherited = True
existing_slots.update(
{
name: getattr(base_cls, name)
for name in getattr(base_cls, "__slots__", [])
}
)
base_names = set(self._base_names)
names = self._attr_names
if (
self._weakref_slot
and "__weakref__" not in getattr(self._cls, "__slots__", ())
and "__weakref__" not in names
and not weakref_inherited
):
names += ("__weakref__",)
cached_properties = {
name: cached_prop.func
for name, cached_prop in cd.items()
if isinstance(cached_prop, cached_property)
}
# Collect methods with a `__class__` reference that are shadowed in the new class.
# To know to update them.
additional_closure_functions_to_update = []
if cached_properties:
import inspect
class_annotations = _get_annotations(self._cls)
for name, func in cached_properties.items():
# Add cached properties to names for slotting.
names += (name,)
# Clear out function from class to avoid clashing.
del cd[name]
additional_closure_functions_to_update.append(func)
annotation = inspect.signature(func).return_annotation
if annotation is not inspect.Parameter.empty:
class_annotations[name] = annotation
original_getattr = cd.get("__getattr__")
if original_getattr is not None:
additional_closure_functions_to_update.append(original_getattr)
cd["__getattr__"] = _make_cached_property_getattr(
cached_properties, original_getattr, self._cls
)
# We only add the names of attributes that aren't inherited.
# Setting __slots__ to inherited attributes wastes memory.
slot_names = [name for name in names if name not in base_names]
# There are slots for attributes from current class
# that are defined in parent classes.
# As their descriptors may be overridden by a child class,
# we collect them here and update the class dict
reused_slots = {
slot: slot_descriptor
for slot, slot_descriptor in existing_slots.items()
if slot in slot_names
}
slot_names = [name for name in slot_names if name not in reused_slots]
cd.update(reused_slots)
if self._cache_hash:
slot_names.append(_HASH_CACHE_FIELD)
cd["__slots__"] = tuple(slot_names)
cd["__qualname__"] = self._cls.__qualname__
# Create new class based on old class and our methods.
cls = type(self._cls)(self._cls.__name__, self._cls.__bases__, cd)
# The following is a fix for
# <https://github.com/python-attrs/attrs/issues/102>.
# If a method mentions `__class__` or uses the no-arg super(), the
# compiler will bake a reference to the class in the method itself
# as `method.__closure__`. Since we replace the class with a
# clone, we rewrite these references so it keeps working.
for item in itertools.chain(
cls.__dict__.values(), additional_closure_functions_to_update
):
if isinstance(item, (classmethod, staticmethod)):
# Class- and staticmethods hide their functions inside.
# These might need to be rewritten as well.
closure_cells = getattr(item.__func__, "__closure__", None)
elif isinstance(item, property):
# Workaround for property `super()` shortcut (PY3-only).
# There is no universal way for other descriptors.
closure_cells = getattr(item.fget, "__closure__", None)
else:
closure_cells = getattr(item, "__closure__", None)
if not closure_cells: # Catch None or the empty list.
continue
for cell in closure_cells:
try:
match = cell.cell_contents is self._cls
except ValueError: # noqa: PERF203
# ValueError: Cell is empty
pass
else:
if match:
cell.cell_contents = cls
return cls
def add_repr(self, ns):
script, globs = _make_repr_script(self._attrs, ns)
def _attach_repr(cls_dict, globs):
cls_dict["__repr__"] = self._add_method_dunders(globs["__repr__"])
self._script_snippets.append((script, globs, _attach_repr))
self._repr_added = True
return self
def add_str(self):
if not self._repr_added:
msg = "__str__ can only be generated if a __repr__ exists."
raise ValueError(msg)
def __str__(self):
return self.__repr__()
self._cls_dict["__str__"] = self._add_method_dunders(__str__)
return self
def _make_getstate_setstate(self):
"""
Create custom __setstate__ and __getstate__ methods.
"""
# __weakref__ is not writable.
state_attr_names = tuple(
an for an in self._attr_names if an != "__weakref__"
)
def slots_getstate(self):
"""
Automatically created by attrs.
"""
return {name: getattr(self, name) for name in state_attr_names}
hash_caching_enabled = self._cache_hash
def slots_setstate(self, state):
"""
Automatically created by attrs.
"""
__bound_setattr = _OBJ_SETATTR.__get__(self)
if isinstance(state, tuple):
# Backward compatibility with attrs instances pickled with
# attrs versions before v22.2.0 which stored tuples.
loopIteratorF = __import__("builtins").iter(zip(state_attr_names, state, strict=False))
while True:
try:
name, value = __import__("builtins").next(loopIteratorF)
except __import__("builtins").StopIteration:
break
__bound_setattr(name, value)
else:
loopIteratorG = __import__("builtins").iter(state_attr_names)
while True:
try:
name = __import__("builtins").next(loopIteratorG)
except __import__("builtins").StopIteration:
break
if name in state:
__bound_setattr(name, state[name])
# The hash code cache is not included when the object is
# serialized, but it still needs to be initialized to None to
# indicate that the first call to __hash__ should be a cache
# miss.
if hash_caching_enabled:
__bound_setattr(_HASH_CACHE_FIELD, None)
return slots_getstate, slots_setstate
def make_unhashable(self):
self._cls_dict["__hash__"] = None
return self
def add_hash(self):
script, globs = _make_hash_script(
self._cls,
self._attrs,
frozen=self._frozen,
cache_hash=self._cache_hash,
)
def attach_hash(cls_dict: dict, locs: dict) -> None:
cls_dict["__hash__"] = self._add_method_dunders(locs["__hash__"])
self._script_snippets.append((script, globs, attach_hash))
return self
def add_init(self):
script, globs, annotations = _make_init_script(
self._cls,
self._attrs,
self._has_pre_init,
self._pre_init_has_args,
self._has_post_init,
self._frozen,
self._slots,
self._cache_hash,
self._base_attr_map,
self._is_exc,
self._on_setattr,
attrs_init=False,
)
def _attach_init(cls_dict, globs):
init = globs["__init__"]
init.__annotations__ = annotations
cls_dict["__init__"] = self._add_method_dunders(init)
self._script_snippets.append((script, globs, _attach_init))
return self
def add_replace(self):
# We create a local `evolve` proxy because it gets modified in place.
def __replace__(*args, **changes):
return evolve(*args, **changes)
self._cls_dict["__replace__"] = self._add_method_dunders(__replace__)
return self
def add_match_args(self):
self._cls_dict["__match_args__"] = tuple(
field.name
for field in self._attrs
if field.init and not field.kw_only
)
def add_attrs_init(self):
script, globs, annotations = _make_init_script(
self._cls,
self._attrs,
self._has_pre_init,
self._pre_init_has_args,
self._has_post_init,
self._frozen,
self._slots,
self._cache_hash,
self._base_attr_map,
self._is_exc,
self._on_setattr,
attrs_init=True,
)
def _attach_attrs_init(cls_dict, globs):
init = globs["__attrs_init__"]
init.__annotations__ = annotations
cls_dict["__attrs_init__"] = self._add_method_dunders(init)
self._script_snippets.append((script, globs, _attach_attrs_init))
return self
def add_eq(self):
cd = self._cls_dict
script, globs = _make_eq_script(self._attrs)
def _attach_eq(cls_dict, globs):
cls_dict["__eq__"] = self._add_method_dunders(globs["__eq__"])
self._script_snippets.append((script, globs, _attach_eq))
cd["__ne__"] = __ne__
return self
def add_order(self):
cd = self._cls_dict
cd["__lt__"], cd["__le__"], cd["__gt__"], cd["__ge__"] = (
self._add_method_dunders(meth)
for meth in _make_order(self._cls, self._attrs)
)
return self
def add_setattr(self):
sa_attrs = {}
loopIteratorH = __import__("builtins").iter(self._attrs)
while True:
try:
a = __import__("builtins").next(loopIteratorH)
except __import__("builtins").StopIteration:
break
on_setattr = a.on_setattr or self._on_setattr
if on_setattr and on_setattr is not setters.NO_OP:
sa_attrs[a.name] = (
a,
on_setattr,
_lazy_is_generator(on_setattr),
)
if not sa_attrs:
return self
if self._has_custom_setattr:
# We need to write a __setattr__ but there already is one!
msg = "Can't combine custom __setattr__ with on_setattr hooks."
raise ValueError(msg)
# docstring comes from _add_method_dunders
def __setattr__(self, name, val):
try:
a, hook, is_gen = sa_attrs[name]
except KeyError:
_OBJ_SETATTR(self, name, val)
return
if is_gen():
gen = hook(self, a, val)
nval = next(gen)
_OBJ_SETATTR(self, name, nval)
try:
next(gen)
except StopIteration:
return
gen.close()
msg = "Generator on_setattr hook yielded more than once."
raise RuntimeError(msg)
nval = hook(self, a, val)
_OBJ_SETATTR(self, name, nval)
return
self._cls_dict["__attrs_own_setattr__"] = True
self._cls_dict["__setattr__"] = self._add_method_dunders(__setattr__)
self._wrote_own_setattr = True
return self
def _add_method_dunders_unsafe(self, method: Callable) -> Callable:
"""
Add __module__ and __qualname__ to a *method*.
"""
method.__module__ = self._cls.__module__
method.__qualname__ = f"{self._cls.__qualname__}.{method.__name__}"
method.__doc__ = (
f"Method generated by attrs for class {self._cls.__qualname__}."
)
return method
def _add_method_dunders_safe(self, method: Callable) -> Callable:
"""
Add __module__ and __qualname__ to a *method* if possible.
"""
with contextlib.suppress(AttributeError):
method.__module__ = self._cls.__module__
with contextlib.suppress(AttributeError):
method.__qualname__ = f"{self._cls.__qualname__}.{method.__name__}"
with contextlib.suppress(AttributeError):
method.__doc__ = f"Method generated by attrs for class {self._cls.__qualname__}."
return method