fix(mutation): only register sqla events once per mapper #201
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -4,6 +4,7 @@ | |
| from collections.abc import Hashable, Iterator, Sequence | ||
| from dataclasses import dataclass, field | ||
| from typing import TYPE_CHECKING, Any, Generic, Literal, TypeAlias, TypeVar, cast, final | ||
| from weakref import WeakValueDictionary | ||
|
|
||
| from sqlalchemy import event, inspect | ||
| from sqlalchemy.orm import MapperProperty, RelationshipDirection, object_mapper | ||
|
|
@@ -30,8 +31,7 @@ | |
| from strawchemy.typing import MappedGraphQLDTO | ||
| from strawchemy.validation.base import ValidationProtocol | ||
|
|
||
| __all__ = ("EventRegistry", "Input", "InputModel", "LevelInput", "RelationInput", "RelationType") | ||
|
|
||
| T = TypeVar("T", bound=MappedDTO[Any]) | ||
| DeclarativeBaseT = TypeVar("DeclarativeBaseT", bound="DeclarativeBase") | ||
|
|
@@ -103,22 +103,19 @@ def __bool__(self) -> bool: | |
| @dataclass | ||
| class RelationInput(_UnboundRelationInput): | ||
| parent: DeclarativeBase = field(kw_only=True) | ||
| event_registry: EventRegistry = field(kw_only=True) | ||
|
|
||
| def __post_init__(self) -> None: | ||
| super().__post_init__() | ||
| self.event_registry.register(self) | ||
|
|
||
| @classmethod | ||
| def from_unbound(cls, unbound: _UnboundRelationInput, model: DeclarativeBase, registry: EventRegistry) -> Self: | ||
| return cls( | ||
| attribute=unbound.attribute, | ||
| related=unbound.related, | ||
| parent=model, | ||
| event_registry=registry, | ||
| set_=unbound.set, | ||
| add=unbound.add, | ||
| remove=unbound.remove, | ||
|
|
@@ -129,27 +126,85 @@ def from_unbound(cls, unbound: _UnboundRelationInput, model: DeclarativeBase) -> | |
| upsert=unbound.upsert, | ||
| ) | ||
|
|
||
| def handle_set(self, value: DeclarativeBase | None) -> None: | ||
| if value is None: | ||
| return | ||
| if _has_record(value): | ||
| self.set = [value] | ||
| else: | ||
| self.create = [value] | ||
|
|
||
| def handle_append(self, value: DeclarativeBase) -> None: | ||
| if _has_record(value): | ||
| self.add.append(value) | ||
| else: | ||
| self.create.append(value) | ||
|
|
||
| def handle_remove(self, value: DeclarativeBase) -> None: | ||
| if _has_record(value): | ||
| self.add = [model for model in self.add if model is not value] | ||
| else: | ||
| self.create = [model for model in self.create if model is not value] | ||
|
|
||
|
|
||
| def _new_entries() -> WeakValueDictionary[tuple[int, str], RelationInput]: | ||
| return WeakValueDictionary() | ||
|
|
||
|
|
||
| @dataclass | ||
| class EventRegistry: | ||
| """Routes SQLAlchemy relationship events to the RelationInput owning (target, attribute). | ||
|
|
||
| A single set/append/remove listener is registered per relationship attribute for the | ||
| process lifetime. Each event is dispatched to the one RelationInput whose ``parent`` is | ||
| the event's ``target``, looked up in a weakly-held per-parent mapping. | ||
| """ | ||
|
|
||
| _entries: WeakValueDictionary[tuple[int, str], RelationInput] = field(init=False, default_factory=_new_entries) | ||
| """Maps ``(id(parent), attr_key)`` to the owning RelationInput, held weakly. | ||
|
|
||
| The value (``RelationInput``) strongly references ``parent``, so ``id(parent)`` | ||
| cannot be recycled while this entry is live — the key is therefore stable for the | ||
| entry's lifetime. | ||
| """ | ||
| _registry: set[MapperProperty[Any]] = field(init=False, default_factory=set) | ||
| """Relationship attributes whose dispatcher has already been registered.""" | ||
|
|
||
| def register(self, relation: RelationInput) -> None: | ||
| """Wire the relation's attribute once and index the relation by its parent.""" | ||
| self._register(relation.attribute, relation.relation_type) | ||
| # Weak-ownership invariant: the registry holds entries weakly (by value), so a | ||
| # RelationInput that is no longer retained by its owning Input stops routing once | ||
| # collected. The owning Input strongly holds every relation it consumes via | ||
| # self.relations, so live relations are always reachable. | ||
| self._entries[(id(relation.parent), relation.attribute.key)] = relation | ||
|
|
||
| def _register(self, attribute: MapperProperty[Any], relation_type: RelationType) -> None: | ||
| if attribute in self._registry: | ||
| return | ||
| if relation_type is RelationType.TO_ONE: | ||
| event.listens_for(attribute, "set")(self._dispatch_set) | ||
| else: | ||
| event.listens_for(attribute, "append")(self._dispatch_append) | ||
| event.listens_for(attribute, "remove")(self._dispatch_remove) | ||
| self._registry.add(attribute) | ||
|
Comment on lines
+170
to
+190
Contributor
There was a problem hiding this comment. 🩺 Stability & Availability | 🟠 Major | 🏗️ Heavy lift 🧩 Analysis chain🏁 Script executed: #!/bin/bash
set -euo pipefail
# Map the target file and nearby symbols first.
ast-grep outline src/strawchemy/schema/mutation/input.py --view expanded
# Show the relevant line ranges with line numbers.
sed -n '150,340p' src/strawchemy/schema/mutation/input.py | cat -n
# Find EventRegistry, Input, RelationInput, and any registry wiring or locking.
rg -n "class EventRegistry|def register\(|_registry|EventRegistry\(|RelationInput|threading|Lock|RLock|listens_for|event\.listens_for" src/strawchemy/schema/mutation/input.py src/strawchemy/schema -g '!**/__pycache__/**'Repository: gazorby/strawchemy Length of output: 25511 🏁 Script executed: #!/bin/bash
set -euo pipefail
# Inspect how EventRegistry is created and shared by field builders/field definitions.
sed -n '1,240p' src/strawchemy/schema/mutation/field_builder.py | cat -n
printf '\n----\n'
sed -n '1,240p' src/strawchemy/schema/mutation/fields.py | cat -n
# Look for other EventRegistry construction sites.
rg -n "EventRegistry\(" src/strawchemy -g '!**/__pycache__/**'Repository: gazorby/strawchemy Length of output: 18700 🏁 Script executed: #!/bin/bash
set -euo pipefail
# Find every direct Input(...) construction to see whether the registry is always provided.
rg -n "\bInput\s*\(" src/strawchemy -g '!**/__pycache__/**'Repository: gazorby/strawchemy Length of output: 757 🏁 Script executed: #!/bin/bash
set -euo pipefail
# Inspect the registry owner in mapper.py and any other ad-hoc EventRegistry creation paths.
sed -n '90,140p' src/strawchemy/mapper.py | cat -n
printf '\n----\n'
rg -n "EventRegistry\(\)|registry=" src/strawchemy -g '!**/__pycache__/**'Repository: gazorby/strawchemy Length of output: 3997 🏁 Script executed: #!/bin/bash
set -euo pipefail
# Inspect the surrounding mapper code that creates the second EventRegistry.
sed -n '1,180p' src/strawchemy/mapper.py | cat -n
# Check whether Input is part of the public API or used internally elsewhere.
rg -n "^from strawchemy\.schema\.mutation\.input import Input|^from .* import Input|Input\(" src/strawchemy -g '!**/__pycache__/**'Repository: gazorby/strawchemy Length of output: 12262 🏁 Script executed: #!/bin/bash
set -euo pipefail
# Inspect the surrounding mapper code that creates the second EventRegistry.
sed -n '1,180p' src/strawchemy/mapper.py | cat -n
printf '\n----\n'
# Check whether Input is used outside the mutation field code paths.
rg -n "from strawchemy\.schema\.mutation\.input import Input|\\bInput\\(" src/strawchemy -g '!**/__pycache__/**'Repository: gazorby/strawchemy Length of output: 10364 Make listener registration shared and synchronized. The mutation path already reuses one 🤖 Prompt for AI Agents |
||
|
|
||
| def _get_input(self, target: DeclarativeBase, initiator: Any) -> RelationInput | None: | ||
| return self._entries.get((id(target), initiator.key)) | ||
|
|
||
| def _dispatch_set(self, target: DeclarativeBase, value: DeclarativeBase, _oldvalue: Any, initiator: Any) -> None: | ||
| if (relation := self._get_input(target, initiator)) is not None: | ||
| relation.handle_set(value) | ||
|
|
||
| def _dispatch_append(self, target: DeclarativeBase, value: DeclarativeBase, initiator: Any) -> None: | ||
| if (relation := self._get_input(target, initiator)) is not None: | ||
| relation.handle_append(value) | ||
|
|
||
| def _dispatch_remove(self, target: DeclarativeBase, value: DeclarativeBase, initiator: Any) -> None: | ||
|
Comment on lines
+192
to
+203
Contributor
There was a problem hiding this comment. 📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick win 🧩 Analysis chain🏁 Script executed: #!/bin/bash
set -euo pipefail
# Inspect the target file with line numbers and nearby context.
sed -n '1,280p' src/strawchemy/schema/mutation/input.py | cat -n
# Find the event hookup sites and related handler signatures.
rg -n "handle_set|handle_append|handle_remove|_dispatch_set|_dispatch_append|_dispatch_remove|_get_input|AttributeEventToken|initiator|_oldvalue" src/strawchemy -S
# Check project typing/lint configuration for Ruff/mypy rules that might flag Any.
rg -n "ruff|Any|ANN|FA|TC|typing" pyproject.toml setup.cfg tox.ini . -g 'pyproject.toml' -g 'setup.cfg' -g 'tox.ini' -g '*.toml' -S
# Inspect installed SQLAlchemy typing symbols if available in the sandbox runtime.
python3 - <<'PY'
import importlib.util
mods = ["sqlalchemy", "sqlalchemy.orm.attributes"]
for m in mods:
spec = importlib.util.find_spec(m)
print(f"{m}: {'found' if spec else 'missing'}")
if importlib.util.find_spec("sqlalchemy"):
import sqlalchemy
print("sqlalchemy version:", getattr(sqlalchemy, "__version__", "unknown"))
try:
from sqlalchemy.orm.attributes import AttributeEventToken
print("AttributeEventToken:", AttributeEventToken)
except Exception as e:
print("AttributeEventToken import error:", type(e).__name__, e)
PYRepository: gazorby/strawchemy Length of output: 18398 🏁 Script executed: #!/bin/bash
set -euo pipefail
# Show the SQLAlchemy version pinned by this project.
rg -n 'sqlalchemy|SQLAlchemy' pyproject.toml uv.lock poetry.lock requirements*.txt -S
# Inspect the runtime typing surface for the event token and relationship event docs in the installed package, if present.
python3 - <<'PY'
import importlib.util
from pprint import pprint
mods = ["sqlalchemy", "sqlalchemy.orm.attributes", "sqlalchemy.event"]
for m in mods:
spec = importlib.util.find_spec(m)
print(f"{m}: {'found' if spec else 'missing'}")
if importlib.util.find_spec("sqlalchemy"):
import sqlalchemy
print("sqlalchemy.__version__ =", getattr(sqlalchemy, "__version__", "unknown"))
try:
from sqlalchemy.orm.attributes import AttributeEventToken
print("AttributeEventToken =", AttributeEventToken)
except Exception as e:
print("AttributeEventToken import failed:", type(e).__name__, e)
try:
from sqlalchemy.orm import attributes
names = [n for n in dir(attributes) if "Event" in n or "Token" in n or "NO_VALUE" in n]
pprint(names[:80])
except Exception as e:
print("dir(sqlalchemy.orm.attributes) failed:", type(e).__name__, e)
PYRepository: gazorby/strawchemy Length of output: 16741 🏁 Script executed: #!/bin/bash
set -euo pipefail
python3 - <<'PY'
import importlib.util
import inspect
print("sqlalchemy:", bool(importlib.util.find_spec("sqlalchemy")))
if importlib.util.find_spec("sqlalchemy"):
import sqlalchemy
print("sqlalchemy.__version__ =", getattr(sqlalchemy, "__version__", "unknown"))
from sqlalchemy.orm.attributes import AttributeEventToken
print("AttributeEventToken import ok:", AttributeEventToken)
print("AttributeEventToken module:", AttributeEventToken.__module__)
print("AttributeEventToken members:", [n for n in dir(AttributeEventToken) if not n.startswith('_')])
# Show whether the token exposes the key attribute used by the code.
try:
print("has key attribute:", hasattr(AttributeEventToken, "key"))
except Exception as e:
print("hasattr(key) error:", type(e).__name__, e)
# Inspect listener signature docs in installed package if available.
import sqlalchemy.orm.attributes as attrs
for name in ["set_attribute", "append", "remove"]:
obj = getattr(attrs, name, None)
if obj is not None:
try:
print(name, inspect.signature(obj))
except Exception as e:
print(name, "signature error:", type(e).__name__, e)
PYRepository: gazorby/strawchemy Length of output: 174 🌐 Web query:
💡 Result: In SQLAlchemy, the AttributeEventToken (formerly named AttributeEvent in versions prior to 2.0) is a token propagated through a chain of attribute events to track the source and control the propagation of operations [1][2]. When using the AttributeEvents.set listener, the initiator argument receives an instance of this token, which serves as an indicator of the event's origin [1][3][4]. Regarding your specific interests: 1. AttributeEventToken: This class acts as a propagation token [1][2]. Its primary attributes are impl (the AttributeImpl that initiated the event) and op (a symbol indicating the operation type, such as OP_APPEND, OP_REMOVE, OP_REPLACE, or OP_BULK_REPLACE) [1][5]. 2. Role in 'set' events: When an attribute 'set' event occurs, the initiator argument is passed to your listener function [3][4]. While often used internally by SQLAlchemy's backref handlers to manage chains of dependent attribute changes, you can inspect it to understand the source of the event [1][2][4]. 3. oldvalue: In the AttributeEvents.set signature, the oldvalue parameter represents the previous value of the attribute being modified [3][6]. It is passed as an object or potentially None if the attribute had no previous value or if SQLAlchemy was not able to retrieve it [1][6]. 4. Interaction: You can safely use these arguments in your listener signature: def my_set_listener(target, value, oldvalue, initiator): # 'target' is the instance being modified # 'value' is the new value being set # 'oldvalue' is the previous value (or None) # 'initiator' is the AttributeEventToken instance pass The initiator is an object and will not be None under normal operation of the event system, as it is designed to represent the event's path [1][7]. If you are seeing None, it may suggest an unusual invocation path or a custom implementation [7]. Citations:
🌐 Web query:
💡 Result: In SQLAlchemy, AttributeEventToken (formerly named AttributeEvent in versions prior to 2.0) is a specialized object used to track and control the propagation of attribute events across a chain of operations [1][2]. When you define event listeners for ORM attributes—such as those using AttributeEvents.set, AttributeEvents.append, or AttributeEvents.remove—these listeners often accept an initiator argument [1][2]. This initiator is typically an instance of AttributeEventToken [2]. Key characteristics and roles of the AttributeEventToken include: 1. Identification of Source: It acts as an indicator of the source of the event, allowing the system to identify the original operation that triggered a sequence of changes [1][2]. 2. Propagation Control: It is used to manage and control the propagation of events across mutually dependent attributes, most notably in backref event handlers [1][2]. 3. Event Chaining: Because attribute changes (like setting a value or appending to a collection) can trigger cascading changes in related objects or attributes, the token is propagated through this chain to prevent infinite loops or redundant operations [1][2]. 4. Listener Signature: In a standard event listener function (e.g., a validator or a 'set' listener), the initiator is passed as the final argument [3][2]. For example: def my_listener(target, value, oldvalue, initiator): # 'initiator' is the AttributeEventToken instance pass [3] The token provides access to the AttributeImpl (the current initiator) and the operation type (such as OP_APPEND, OP_REMOVE, OP_REPLACE, or OP_BULK_REPLACE), which helps in inspecting the specific nature of the change within the event handler [1][2]. Citations:
Replace the SQLAlchemy event callback annotations. Use 🧰 Tools🪛 Ruff (0.15.20)[warning] 192-192: Dynamically typed expressions (typing.Any) are disallowed in (ANN401) [warning] 195-195: Dynamically typed expressions (typing.Any) are disallowed in (ANN401) [warning] 195-195: Dynamically typed expressions (typing.Any) are disallowed in (ANN401) [warning] 199-199: Dynamically typed expressions (typing.Any) are disallowed in (ANN401) [warning] 203-203: Dynamically typed expressions (typing.Any) are disallowed in (ANN401) 🤖 Prompt for AI AgentsSource: Linters/SAST tools |
||
| if (relation := self._get_input(target, initiator)) is not None: | ||
| relation.handle_remove(value) | ||
|
|
||
|
|
||
| @dataclass | ||
| class _InputVisitor(VisitorProtocol[DeclarativeBaseT], Generic[DeclarativeBaseT, InputModel]): | ||
| input_data: Input[InputModel] | ||
|
|
@@ -228,7 +283,7 @@ def model( | |
| delattr(model, attribute) | ||
|
|
||
| for relation in self.current_relations: | ||
| self.input_data.add_relation(RelationInput.from_unbound(relation, model, self.input_data.registry)) | ||
| self.current_relations.clear() | ||
| # Return dict because .model_validate will be called at root level | ||
| return model if level == 1 or self.input_data.validation is None else params | ||
|
|
@@ -250,8 +305,11 @@ def __init__( | |
| self, | ||
| dtos: MappedGraphQLDTO[InputModel] | Sequence[MappedGraphQLDTO[InputModel]], | ||
| _validation_: ValidationProtocol[InputModel] | None = None, | ||
| *, | ||
| registry: EventRegistry | None = None, | ||
| **override: Any, | ||
| ) -> None: | ||
| self.registry = registry if registry is not None else EventRegistry() | ||
| self.max_level = 0 | ||
| self.relations: list[RelationInput] = [] | ||
| self.instances: list[InputModel] = [] | ||
|
|
@@ -305,6 +363,7 @@ def _add_non_input_relations( | |
| relation = RelationInput( | ||
| attribute=relationship, | ||
| parent=model, | ||
| event_registry=self.registry, | ||
| level=_level, | ||
| input_index=input_index, | ||
| relation_type=relation_type, | ||
|
|
||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Reset TO_ONE relation state on every
setevent.Line 129 currently ignores
None, and Lines 132-135 only update one bucket (setorcreate). That leaves stale state in the other bucket after multiple assignments in one request, which can drive incorrect mutation operations.Proposed fix
🤖 Prompt for AI Agents