Association fields

Fields for establishing relationships between domain elements -- connecting entities and aggregates.

See Association Fields reference for usage examples.

HasOne

HasOne(to_cls: str | type, **kwargs: Any)

Bases: Association

Represents an one-to-one association between an aggregate and its entities.

This field is used to define a relationship where an aggregate is associated with at most one instance of a child entity.

Source code in src/protean/fields/association.py

def __init__(self, to_cls: "str | type", **kwargs: Any) -> None:
    super().__init__(**kwargs)

    self._to_cls = to_cls
    self.via: str | None = kwargs.pop("via", None)
    self.change = None

    # Associations are not data fields — they cannot be `required` or `unique`
    self.required = False
    self.unique = False

set

__set__(instance: Any, value: Any) -> None

Setup relationship to be persisted/updated

We track the change in the instance's _temp_cache to determine if the relationship has been added, updated, or deleted. We track two aspects: state and old value.

For HasOne, there are three possible states:

ADDED: The relationship is being added for the first time
UPDATED: The relationship is being updated
DELETED: The relationship is being removed

Of these, the old value is applicable for UPDATED and DELETED states.

Also, we recursively remove child entities if they are associated with the old value.

The temp_cache we set up here is eventually used by the Repository to determine the changes to be persisted.

Source code in src/protean/fields/association.py

def __set__(self, instance: Any, value: Any) -> None:
    """Setup relationship to be persisted/updated

    We track the change in the instance's `_temp_cache` to determine if the relationship
    has been added, updated, or deleted. We track two aspects: state and old value.

    For `HasOne`, there are three possible states:

    - ADDED: The relationship is being added for the first time
    - UPDATED: The relationship is being updated
    - DELETED: The relationship is being removed

    Of these, the old value is applicable for `UPDATED` and `DELETED` states.

    Also, we recursively remove child entities if they are associated with the old value.

    The `temp_cache` we set up here is eventually used by the `Repository` to determine
    the changes to be persisted.
    """
    # Accept dictionary values and convert them to Entity objects
    if isinstance(value, dict):
        value = self._target_cls(**value)

    super().__set__(instance, value)

    if value is not None and not isinstance(value, self._target_cls):
        raise ValidationError(
            {
                "_entity": [
                    f"Value assigned to '{self.field_name}' is not of type '{self._target_cls.__name__}'"
                ]
            }
        )

    # 1. Preserve parent linkage in child entity
    if value is not None:
        # This updates the parent's unique identifier in the child
        #   so that the foreign key relationship is preserved
        instance_id_fld = id_field(instance)
        assert instance_id_fld is not None
        assert instance_id_fld.field_name is not None
        id_value = getattr(instance, instance_id_fld.field_name)
        linked_attribute = self._linked_attribute(instance.__class__)
        if hasattr(value, linked_attribute):
            setattr(
                value, linked_attribute, id_value
            )  # This overwrites any existing linkage, which is correct

        # Add the parent to the child entity cache
        # Temporarily set linkage to parent in child entity
        setattr(value, self._linked_reference(type(instance)), instance)

    # 2. Determine and store the change in the relationship
    assert self.field_name is not None
    current_value = getattr(instance, self.field_name)
    if current_value:
        current_value_id_fld = id_field(current_value)
        assert current_value_id_fld is not None
        assert current_value_id_fld.field_name is not None
        current_value_id = getattr(current_value, current_value_id_fld.field_name)
    else:
        current_value_id = None
    if value:
        value_id_fld = id_field(value)
        assert value_id_fld is not None
        assert value_id_fld.field_name is not None
        value_id = getattr(value, value_id_fld.field_name)
    else:
        value_id = None
    cache = instance._temp_cache.setdefault(self.field_name, HasOneChanges())
    if current_value is None:
        # Entity was not associated earlier
        cache.change = "ADDED"
    elif value is None:
        # Entity was associated earlier, but now being removed
        cache.change = "DELETED"
        cache.old_value = current_value
    elif current_value_id != value_id:
        # A New Entity is being associated replacing the old one
        cache.change = "UPDATED"
        cache.old_value = current_value
    elif current_value_id == value_id and value.state_.is_changed:
        # Entity was associated earlier, but now being updated
        cache.change = "UPDATED"
    else:
        cache.change = None  # The same object has been assigned, No-Op

    self._set_own_value(instance, value)

    # 3. Go Recursive and remove child entities if they are associated with the old value
    if cache.change == "DELETED":
        old_value = cache.old_value
        if has_association_fields(old_value):
            for field_name, field_obj in association_fields(old_value).items():
                if isinstance(field_obj, HasMany):
                    field_obj.remove(old_value, getattr(old_value, field_name))
                elif isinstance(field_obj, HasOne):
                    setattr(old_value, field_name, None)

    if instance._initialized and instance._root is not None:
        instance._root._postcheck()  # Trigger validations from the top

as_dict

as_dict(value: Any) -> Any

Return JSON-compatible value of self

Source code in src/protean/fields/association.py

def as_dict(self, value: Any) -> Any:
    """Return JSON-compatible value of self"""
    if value is not None:
        return value.to_dict()

HasMany

HasMany(to_cls: str | type, **kwargs: Any)

Bases: Association

Represents a one-to-many association between two entities. This field is used to define a relationship where an aggregate has multiple instances of a chil entity.

PARAMETER	DESCRIPTION
`to_cls`	The class of the target entity. TYPE: `type`
`**kwargs`	Additional keyword arguments to be passed to the base field class. TYPE: `Any` DEFAULT: `{}`

Source code in src/protean/fields/association.py

def __init__(self, to_cls: "str | type", **kwargs: Any) -> None:
    super().__init__(**kwargs)

    self._to_cls = to_cls
    self.via: str | None = kwargs.pop("via", None)
    self.change = None

    # Associations are not data fields — they cannot be `required` or `unique`
    self.required = False
    self.unique = False

set

__set__(instance: Any, value: Any) -> None

This supports direct assignment of values to HasMany fields, like: order.items = [item1, item2, item3]

Source code in src/protean/fields/association.py

def __set__(self, instance: Any, value: Any) -> None:
    """This supports direct assignment of values to HasMany fields, like:
    `order.items = [item1, item2, item3]`
    """
    value = value if isinstance(value, list) else [value]

    # Accept dictionary values and convert them to Entity objects
    values = []
    for item in value:
        if isinstance(item, dict):
            values.append(self._target_cls(**item))
        else:
            values.append(item)

    super().__set__(instance, values)

    if value is not None:
        # Ensure cache is initialized before add() tries to read via getattr().
        # Without this, getattr triggers __get__ which falls back to a DB query,
        # failing for event-sourced aggregates that have no database tables.
        try:
            self.get_cached_value(instance)
        except KeyError:
            self.set_cached_value(instance, [])
        self.add(instance, values)

add

add(instance: Any, items: Any) -> None

Available as add_<HasMany Field Name> method on the entity instance.

This method adds one or more linked entities to the source entity. It also diffs the current value with the new value to determine the changes that need to be persisted.

The method also takes care of the attributes of the linked entities, preparing them for persistence.

Each linkage takes care of its own attributes, preparing them for persistence. One exception is deletion of an entity - all child entities have to be marked as removed.

We track the change in the instance's _temp_cache to determine if the relationship has been added, updated, or deleted. We track each item's state and group the changes into three buckets: - ADDED: The relationship is being added for the first time - UPDATED: The relationship is being updated - DELETED: The relationship is being removed

The DELETED objects are detected from the pool of new objects, but it is also possible to remove them directly with the remove method.

The temp_cache we set up here is eventually used by the Repository to determine the changes to be persisted.

PARAMETER	DESCRIPTION
`instance`	The source entity instance. TYPE: `BaseEntity`
`items`	The linked entity or entities to be added. TYPE: `list \| BaseEntity`

Source code in src/protean/fields/association.py

def add(self, instance: Any, items: Any) -> None:
    """
    Available as `add_<HasMany Field Name>` method on the entity instance.

    This method adds one or more linked entities to the source entity. It also diffs the current value with the
    new value to determine the changes that need to be persisted.

    The method also takes care of the attributes of the linked entities, preparing them for persistence.

    Each linkage takes care of its own attributes, preparing them for persistence.
    One exception is deletion of an entity - all child entities have to be marked as removed.

    We track the change in the instance's `_temp_cache` to determine if the relationship
    has been added, updated, or deleted. We track each item's state and group the changes
    into three buckets:
    - ADDED: The relationship is being added for the first time
    - UPDATED: The relationship is being updated
    - DELETED: The relationship is being removed

    The `DELETED` objects are detected from the pool of new objects, but it is also possible to remove them
    directly with the `remove` method.

    The `temp_cache` we set up here is eventually used by the `Repository` to determine
    the changes to be persisted.

    Args:
        instance (BaseEntity): The source entity instance.
        items (list | BaseEntity): The linked entity or entities to be added.
    """
    super().__set__(instance, items)

    # Pre-check invariants before mutation
    if instance._initialized and instance._root is not None:
        instance._root._precheck()

    assert self.field_name is not None
    data = getattr(instance, self.field_name)

    # Convert a single item into a list of items, if necessary
    items = [items] if not isinstance(items, list) else items

    # Validate that all items are of the same type, and the correct type
    for item in items:
        if not isinstance(item, self._target_cls):
            raise ValidationError(
                {
                    "_entity": [
                        f"Value assigned to '{self.field_name}' is not of type '{self._target_cls.__name__}'"
                    ]
                }
            )

    entity_id_fld = id_field(self._target_cls)
    assert entity_id_fld is not None
    assert entity_id_fld.field_name is not None
    instance_id_fld = id_field(instance)
    assert instance_id_fld is not None
    assert instance_id_fld.field_name is not None

    current_value_ids = [getattr(value, entity_id_fld.field_name) for value in data]

    # Remove items when set to empty
    if len(items) == 0 and len(current_value_ids) > 0:
        self.remove(instance, data)

    # Re-read from cache after potential removal so new_data reflects
    # the post-remove state instead of the stale `data` reference.
    try:
        cached = self.get_cached_value(instance)
        new_data = list(cached) if isinstance(cached, list) else [cached]
    except KeyError:
        new_data = list(data)

    cache = instance._temp_cache.setdefault(self.field_name, HasManyChanges())
    for item in items:
        # Items to add
        identity = getattr(item, entity_id_fld.field_name)
        if identity not in current_value_ids:
            # If the same item is added multiple times, the last item added will win
            cache.added[identity] = item

            setattr(
                item,
                self._linked_attribute(type(instance)),
                getattr(instance, instance_id_fld.field_name),
            )

            # Temporarily set linkage to parent in child entity
            setattr(item, self._linked_reference(type(instance)), instance)

            new_data.append(item)
        # Items to update
        elif (
            identity in current_value_ids
            and item.state_.is_persisted
            and item.state_.is_changed
        ):
            setattr(
                item,
                self._linked_attribute(type(instance)),
                getattr(instance, instance_id_fld.field_name),
            )

            # Temporarily set linkage to parent in child entity
            setattr(item, self._linked_reference(type(instance)), instance)

            cache.updated[identity] = item

            # Replace updated item in the working copy
            for i, existing in enumerate(new_data):
                if getattr(existing, entity_id_fld.field_name) == identity:
                    new_data[i] = item
                    break

    # Update cache with complete item list instead of deleting it.
    # This ensures items are immediately visible via __get__ without
    # requiring a DB round-trip, which is essential for event-sourced
    # aggregates that have no database tables.
    self.set_cached_value(instance, new_data)

    if instance._initialized and instance._root is not None:
        instance._root._postcheck()  # Trigger validations from the top

remove

remove(instance: Any, items: Any) -> None

Available as remove_<HasMany Field Name> method on the entity instance.

Remove one or more linked entities from the source entity.

We also recursively remove child entities if they are associated with the removed value.

PARAMETER	DESCRIPTION
`instance`	The source entity instance. TYPE: `BaseEntity`
`items`	The linked entity or entities to be removed. TYPE: `list \| BaseEntity`

Source code in src/protean/fields/association.py

def remove(self, instance: Any, items: Any) -> None:
    """
    Available as `remove_<HasMany Field Name>` method on the entity instance.

    Remove one or more linked entities from the source entity.

    We also recursively remove child entities if they are associated with the removed value.

    Args:
        instance (BaseEntity): The source entity instance.
        items (list | BaseEntity): The linked entity or entities to be removed.
    """
    # Pre-check invariants before mutation
    if instance._initialized and instance._root is not None:
        instance._root._precheck()

    assert self.field_name is not None
    data = getattr(instance, self.field_name)

    # Convert a single item into a list of items, if necessary
    items = [items] if not isinstance(items, list) else items

    # Validate that all items are of the same type, and the correct type
    for item in items:
        if not isinstance(item, self._target_cls):
            raise ValidationError(
                {
                    "_entity": [
                        f"Value assigned to '{self.field_name}' is not of type '{self._target_cls.__name__}'"
                    ]
                }
            )

    entity_id_fld = id_field(self._target_cls)
    assert entity_id_fld is not None
    assert entity_id_fld.field_name is not None

    current_value_ids = [getattr(value, entity_id_fld.field_name) for value in data]

    cache = instance._temp_cache.setdefault(self.field_name, HasManyChanges())
    removed_ids: set[Any] = set()
    for item in items:
        identity = getattr(item, entity_id_fld.field_name)
        if identity in current_value_ids:
            if identity not in cache.removed:
                cache.removed[identity] = item
                removed_ids.add(identity)

        # Remove child entities
        if has_association_fields(item):
            for field_name, field_obj in association_fields(item).items():
                if isinstance(field_obj, HasMany):
                    field_obj.remove(item, getattr(item, field_name))
                elif isinstance(field_obj, HasOne):
                    setattr(item, field_name, None)

    # Update cache with items remaining after removal instead of deleting it.
    if removed_ids:
        new_data = [
            item
            for item in data
            if getattr(item, entity_id_fld.field_name) not in removed_ids
        ]
        self.set_cached_value(instance, new_data)

    if instance._initialized and instance._root is not None:
        instance._root._postcheck()  # Trigger validations from the top

as_dict

as_dict(value: Any) -> list[Any]

Return JSON-compatible value of self.

PARAMETER	DESCRIPTION
`value`	The value to be converted to a JSON-compatible format. TYPE: `list`

RETURNS	DESCRIPTION
`list`	A list of dictionaries representing the linked entities. TYPE: `list[Any]`

Source code in src/protean/fields/association.py

def as_dict(self, value: Any) -> list[Any]:
    """
    Return JSON-compatible value of self.

    Args:
        value (list): The value to be converted to a JSON-compatible format.

    Returns:
        list: A list of dictionaries representing the linked entities.
    """
    return [item.to_dict() for item in value]

get

get(instance: Any, **kwargs: Any) -> Any

Fetch a single linked entity based on the provided criteria.

Available as get_one_from_<HasMany Field Name> method on the entity instance.

PARAMETER	DESCRIPTION
`**kwargs`	The filtering criteria. TYPE: `Any` DEFAULT: `{}`

Source code in src/protean/fields/association.py

def get(self, instance: Any, **kwargs: Any) -> Any:
    """Fetch a single linked entity based on the provided criteria.

    Available as `get_one_from_<HasMany Field Name>` method on the entity instance.

    Args:
        **kwargs (Any): The filtering criteria.
    """
    data = self.filter(instance, **kwargs)

    if len(data) == 0:
        raise exceptions.ObjectNotFoundError(
            "No linked entities matching criteria found"
        )

    if len(data) > 1:
        raise exceptions.TooManyObjectsError(
            "Multiple linked entities matching criteria found"
        )

    return data[0]

filter

filter(instance: Any, **kwargs: Any) -> list[Any]

Filter the linked entities based on the provided criteria.

Available as filter_<HasMany Field Name> method on the entity instance.

PARAMETER	DESCRIPTION
`**kwargs`	The filtering criteria. TYPE: `Any` DEFAULT: `{}`

Source code in src/protean/fields/association.py

def filter(self, instance: Any, **kwargs: Any) -> list[Any]:
    """Filter the linked entities based on the provided criteria.

    Available as `filter_<HasMany Field Name>` method on the entity instance.

    Args:
        **kwargs (Any): The filtering criteria.
    """
    assert self.field_name is not None
    data = getattr(instance, self.field_name)
    return [
        item
        for item in data
        if all(getattr(item, key) == value for key, value in kwargs.items())
    ]

Reference

Reference(to_cls: str | type, **kwargs: Any)

Bases: FieldCacheMixin, Field

A field representing a reference to another entity. This field is used to establish the reverse relationship to the remote entity.

PARAMETER	DESCRIPTION
`to_cls`	The target entity class or its name. TYPE: `str \| type`
`**kwargs`	Additional keyword arguments to be passed to the base `Field` class. TYPE: `Any` DEFAULT: `{}`

Source code in src/protean/fields/association.py

def __init__(self, to_cls: "str | type", **kwargs: Any) -> None:
    super().__init__(**kwargs)
    self._to_cls = to_cls

    self.relation = _ReferenceField(self)

linked_attribute `property`

linked_attribute: str

Return linkage attribute to the target class

This method is initially called from __set_name__() -> get_attribute_name() at which point, the to_cls has not been initialized properly. We simply default the linked attribute to 'id' in that case.

Eventually, when setting value the first time, the to_cls entity is initialized and the attribute name is reset correctly.

get_attribute_name

get_attribute_name() -> str

Return formatted attribute name for the shadow field

Source code in src/protean/fields/association.py

def get_attribute_name(self) -> str:
    """Return formatted attribute name for the shadow field"""
    return self.referenced_as or "{}_{}".format(
        self.field_name, self.linked_attribute
    )

get_shadow_field

get_shadow_field() -> tuple[str | None, _ReferenceField]

Return shadow field Primarily used during Entity initialization to register shadow field

Source code in src/protean/fields/association.py

def get_shadow_field(self) -> "tuple[str | None, _ReferenceField]":
    """Return shadow field
    Primarily used during Entity initialization to register shadow field"""
    return (self.attribute_name, self.relation)

get

__get__(instance: Any, owner: Any) -> Any

Retrieve associated objects

Source code in src/protean/fields/association.py

def __get__(self, instance: Any, owner: Any) -> Any:
    """Retrieve associated objects"""
    reference_obj = None
    if hasattr(instance, "state_"):
        try:
            reference_obj = self.get_cached_value(instance)
        except KeyError:
            # Fetch target object by own Identifier
            id_field = self.get_attribute_name()

            id_value = None
            if hasattr(instance, id_field):
                id_value = getattr(instance, id_field)

            if id_value:
                reference_obj = self._fetch_objects(self.linked_attribute, id_value)
                if reference_obj:
                    self._set_own_value(instance, reference_obj)
                else:
                    # No Objects were found in the remote entity with this Entity's ID
                    pass

    return reference_obj

set

__set__(instance: Any, value: Any) -> None

Override __set__ to coordinate between relation field and its shadow attribute

Source code in src/protean/fields/association.py

def __set__(self, instance: Any, value: Any) -> None:
    """Override `__set__` to coordinate between relation field and its shadow attribute"""
    value = self._load(value)

    if value:
        id_fld = id_field(value)
        assert id_fld is not None
        assert id_fld.field_name is not None
        if getattr(value, id_fld.field_name) is None:
            raise ValueError(
                "Target Object must be saved before being referenced",
                self.field_name,
            )
        else:
            self._set_own_value(instance, value)
            self._set_relation_value(
                instance, getattr(value, self.linked_attribute)
            )
    else:
        self._reset_values(instance)

as_dict

as_dict(value: Any) -> Any

Return JSON-compatible value of self

Source code in src/protean/fields/association.py

def as_dict(self, value: Any) -> Any:
    """Return JSON-compatible value of self"""
    raise NotImplementedError

Association fields

HasOne

__set__

as_dict

HasMany

__set__

add

remove

as_dict

get

filter

Reference

linked_attribute property

get_attribute_name

get_shadow_field

__get__

__set__

as_dict

set

set

linked_attribute `property`

get

set