fix(mutation): only register sqla events once per mapper - #201
Conversation
|
Warning Review limit reached
Next review available in: 12 minutes Enable usage-based reviews in Billing to review now. Otherwise, wait until the next included review is available. How can I continue?After more reviews become available, a review can be triggered using the To avoid repeated limits, reduce automatic review volume by pausing incremental auto-reviews earlier, using label-based review opt-in, excluding WIP or generated PR titles, or requesting reviews manually when the PR is ready. If your team needs uninterrupted high-volume reviews, an organization admin can enable usage-based reviews. How do review limits work?CodeRabbit enforces per-developer PR review limits for each organization. Most developers receive the normal plan review availability. For paid Pro and Pro+ PR reviews, CodeRabbit uses adaptive limits for sustained high-volume activity. When a developer's recent PR review activity reaches the 95th percentile or higher among CodeRabbit users, additional reviews become available more gradually as earlier reviews age out of the rolling window. Please refer docs for additional details. Review details⚙️ Run configurationConfiguration used: Path: .coderabbit.yaml Review profile: ASSERTIVE Plan: Pro Run ID: 📒 Files selected for processing (6)
📝 WalkthroughWalkthroughIntroduces ChangesEventRegistry: relationship event routing
pyproject.toml formatting
Estimated code review effort: 3 (Moderate) | ~25 minutes 🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✨ Finishing Touches🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 2
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
tests/unit/test_event_registry.py (1)
32-99: 🧹 Nitpick | 🔵 Trivial | ⚡ Quick winAdd TO_ONE set/clear regression coverage.
These tests only validate TO_MANY append/remove wiring paths. Please add one TO_ONE case that exercises reassignment and clear (
value -> other_value -> None) so stale state regressions inhandle_setare caught.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@tests/unit/test_event_registry.py` around lines 32 - 99, The existing test cases only validate TO_MANY append/remove wiring paths. Add a new test function that covers TO_ONE relationship set/clear operations to catch stale state regressions in the handle_set method. Create a test that exercises TO_ONE reassignment (changing from one value to another) and clearing (setting to None), similar to how test_event_routes_only_to_owning_relation validates TO_MANY operations. The test should create a Color instance with a TO_ONE relationship, reassign that relationship to a different value, then clear it to None, and verify that the EventRegistry properly tracks each state change in the corresponding RelationInput without stale data.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@pyproject.toml`:
- Line 316: In the ruff exclude configuration, the alembic pattern entry
contains two trailing spaces that need to be removed. Locate the line with
"alembic/* " in the exclude list and remove the two trailing spaces after the
asterisk, leaving just "alembic/*" to match the clean formatting of all other
entries in the exclude list.
In `@src/strawchemy/schema/mutation/input.py`:
- Around line 129-136: The handle_set method needs to reset both the TO_ONE
relation state buckets on every set event to prevent stale state from persisting
across multiple assignments in one request. Currently, when value is None, the
method returns early without clearing the other bucket, and when a value is
provided, only one bucket (self.set or self.create) is updated. Fix this by
resetting both self.set and self.create to empty lists at the start of the
handle_set method before the None check, ensuring both buckets are always
cleared when a new set event occurs, then selectively populate the appropriate
bucket based on the provided value.
---
Outside diff comments:
In `@tests/unit/test_event_registry.py`:
- Around line 32-99: The existing test cases only validate TO_MANY append/remove
wiring paths. Add a new test function that covers TO_ONE relationship set/clear
operations to catch stale state regressions in the handle_set method. Create a
test that exercises TO_ONE reassignment (changing from one value to another) and
clearing (setting to None), similar to how
test_event_routes_only_to_owning_relation validates TO_MANY operations. The test
should create a Color instance with a TO_ONE relationship, reassign that
relationship to a different value, then clear it to None, and verify that the
EventRegistry properly tracks each state change in the corresponding
RelationInput without stale data.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: ASSERTIVE
Plan: Pro
Run ID: a100093f-d476-408b-996c-5156ecfe7ab1
📒 Files selected for processing (6)
pyproject.tomlsrc/strawchemy/mapper.pysrc/strawchemy/schema/mutation/field_builder.pysrc/strawchemy/schema/mutation/fields.pysrc/strawchemy/schema/mutation/input.pytests/unit/test_event_registry.py
| "node_modules", | ||
| "venv", | ||
| "tests/codegen/snapshots", | ||
| "alembic/* ", |
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
# Verify whether "alembic/* " with trailing spaces is the intended pattern.
# Check if this pattern appears elsewhere in the repo or if it's unique to this PR.
rg -n 'alembic/\*' pyproject.toml --no-heading
git log --oneline -p --follow -S 'alembic/*' -- pyproject.toml | head -50 2>/dev/null || echo "(git history unavailable)"Repository: gazorby/strawchemy
Length of output: 1852
🏁 Script executed:
sed -n '290,330p' pyproject.toml | cat -ARepository: gazorby/strawchemy
Length of output: 1104
Remove trailing whitespace from ruff exclude glob pattern.
Line 316 contains "alembic/* " with two trailing spaces inside the string literal. All other entries in the same exclude list are clean (e.g., .bzr, .direnv, tests/codegen/snapshots), making this trailing whitespace clearly unintentional. The glob pattern will not match the intended paths with this extra whitespace.
Change "alembic/* " to "alembic/*".
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@pyproject.toml` at line 316, In the ruff exclude configuration, the alembic
pattern entry contains two trailing spaces that need to be removed. Locate the
line with "alembic/* " in the exclude list and remove the two trailing spaces
after the asterisk, leaving just "alembic/*" to match the clean formatting of
all other entries in the exclude list.
| def handle_set(self, value: DeclarativeBase | None) -> None: | ||
| if value is None: | ||
| return | ||
| if _has_record(value): | ||
| self.set = [value] | ||
| else: | ||
| self.create = [value] | ||
|
|
There was a problem hiding this comment.
Reset TO_ONE relation state on every set event.
Line 129 currently ignores None, and Lines 132-135 only update one bucket (set or create). That leaves stale state in the other bucket after multiple assignments in one request, which can drive incorrect mutation operations.
Proposed fix
def handle_set(self, value: DeclarativeBase | None) -> None:
- if value is None:
- return
- if _has_record(value):
- self.set = [value]
- else:
- self.create = [value]
+ self.create = []
+ if value is None:
+ self.set = None
+ return
+ if _has_record(value):
+ self.set = [value]
+ else:
+ self.set = []
+ self.create = [value]🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/strawchemy/schema/mutation/input.py` around lines 129 - 136, The
handle_set method needs to reset both the TO_ONE relation state buckets on every
set event to prevent stale state from persisting across multiple assignments in
one request. Currently, when value is None, the method returns early without
clearing the other bucket, and when a value is provided, only one bucket
(self.set or self.create) is updated. Fix this by resetting both self.set and
self.create to empty lists at the start of the handle_set method before the None
check, ensuring both buckets are always cleared when a new set event occurs,
then selectively populate the appropriate bucket based on the provided value.
Codecov Report✅ All modified and coverable lines are covered by tests. Additional details and impacted files@@ Coverage Diff @@
## main #201 +/- ##
==========================================
- Coverage 93.26% 93.22% -0.04%
==========================================
Files 72 72
Lines 6486 6524 +38
Branches 853 858 +5
==========================================
+ Hits 6049 6082 +33
- Misses 296 297 +1
- Partials 141 145 +4 ☔ View full report in Codecov by Harness. |
There was a problem hiding this comment.
Actionable comments posted: 2
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
tests/unit/test_event_registry.py (1)
97-150: 🩺 Stability & Availability | 🔵 Trivial | ⚡ Quick winAdd a regression for unshared/default registries.
These tests only prove dedupe within a single shared registry. Add a case that builds two independent
EventRegistry/defaultInputpaths and asserts the mapper attribute is still wired once, matching the PR objective.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@tests/unit/test_event_registry.py` around lines 97 - 150, Add a regression test covering the unshared/default registry path, not just the shared EventRegistry case. Extend the event registry tests to build two independent registries or two default Input constructions and verify the mapper attribute listener on _fruits_prop() is still registered only once. Reuse the existing counting_listens_for pattern and the EventRegistry, Input, and Strawchemy entry points so the new test exercises the same wiring behavior without relying on a shared registry instance.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@src/strawchemy/schema/mutation/input.py`:
- Around line 192-203: The SQLAlchemy event callback signatures in RelationInput
use the wrong annotation types, so update the methods in RelationInput to match
the actual event payloads. In _dispatch_set, _dispatch_append, and
_dispatch_remove, annotate initiator as AttributeEventToken, use object for
_oldvalue, and allow DeclarativeBase | None for value where applicable, since
clearing a to-one relation passes None and handle_set already supports it. Make
the signature changes consistently across these event handler methods so the
callback types align with SQLAlchemy.
- Around line 170-190: The listener wiring in EventRegistry is still reachable
through Input(..., registry=None), which can create a separate registry and
allow duplicate SQLAlchemy listeners on the same attribute. Make the registry
path shared and mandatory (or otherwise ensure all Inputs reuse the same
registry), and add synchronization around EventRegistry._register so concurrent
first-time registration of the same MapperProperty cannot attach listeners
twice.
---
Outside diff comments:
In `@tests/unit/test_event_registry.py`:
- Around line 97-150: Add a regression test covering the unshared/default
registry path, not just the shared EventRegistry case. Extend the event registry
tests to build two independent registries or two default Input constructions and
verify the mapper attribute listener on _fruits_prop() is still registered only
once. Reuse the existing counting_listens_for pattern and the EventRegistry,
Input, and Strawchemy entry points so the new test exercises the same wiring
behavior without relying on a shared registry instance.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: ASSERTIVE
Plan: Pro
Run ID: e59e3ac8-483a-486a-a490-2804c9e0de49
📒 Files selected for processing (2)
src/strawchemy/schema/mutation/input.pytests/unit/test_event_registry.py
| _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) |
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 EventRegistry per Strawchemy instance, but Input(..., registry=None) still creates a fresh registry. That leaves a public path that can attach duplicate SQLAlchemy listeners to the same mapper attribute, and _register() can race if the same registry is first used concurrently. Make the registry mandatory or guard listener wiring with a lock.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/strawchemy/schema/mutation/input.py` around lines 170 - 190, The listener
wiring in EventRegistry is still reachable through Input(..., registry=None),
which can create a separate registry and allow duplicate SQLAlchemy listeners on
the same attribute. Make the registry path shared and mandatory (or otherwise
ensure all Inputs reuse the same registry), and add synchronization around
EventRegistry._register so concurrent first-time registration of the same
MapperProperty cannot attach listeners twice.
| 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: |
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:
SQLAlchemy attribute event set listener initiator type AttributeEventToken oldvalue object None documentation
💡 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:
- 1: https://github.com/sqlalchemy/sqlalchemy/blob/main/lib/sqlalchemy/orm/attributes.py
- 2: https://docs.sqlalchemy.org/en/20/orm/internals.html
- 3: http://docs.sqlalchemy.org/en/latest/orm/events.html
- 4: https://docs.sqlalchemy.org/en/13/orm/events.html
- 5: https://github.com/zzzeek/sqlalchemy/blob/main/lib/sqlalchemy/orm/attributes.py
- 6: https://docs.sqlalchemy.org/en/20/_modules/examples/custom_attributes/listen_for_events.html
- 7: http://docs.sqlalchemy.org/en/latest/orm/session_api.html
🌐 Web query:
site:docs.sqlalchemy.org AttributeEventToken set listener initiator oldvalue object relationship event
💡 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:
- 1: https://docs.sqlalchemy.org/en/13/orm/internals.html?highlight=instancestate
- 2: https://docs.sqlalchemy.org/en/20/orm/internals.html
- 3: https://docs.sqlalchemy.org/en/21/core/event.html
Replace the SQLAlchemy event callback annotations. Use AttributeEventToken for initiator, object for _oldvalue, and DeclarativeBase | None for value; clearing a to-one relation dispatches None, and handle_set already accepts it.
🧰 Tools
🪛 Ruff (0.15.20)
[warning] 192-192: Dynamically typed expressions (typing.Any) are disallowed in initiator
(ANN401)
[warning] 195-195: Dynamically typed expressions (typing.Any) are disallowed in _oldvalue
(ANN401)
[warning] 195-195: Dynamically typed expressions (typing.Any) are disallowed in initiator
(ANN401)
[warning] 199-199: Dynamically typed expressions (typing.Any) are disallowed in initiator
(ANN401)
[warning] 203-203: Dynamically typed expressions (typing.Any) are disallowed in initiator
(ANN401)
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/strawchemy/schema/mutation/input.py` around lines 192 - 203, The
SQLAlchemy event callback signatures in RelationInput use the wrong annotation
types, so update the methods in RelationInput to match the actual event
payloads. In _dispatch_set, _dispatch_append, and _dispatch_remove, annotate
initiator as AttributeEventToken, use object for _oldvalue, and allow
DeclarativeBase | None for value where applicable, since clearing a to-one
relation passes None and handle_set already supports it. Make the signature
changes consistently across these event handler methods so the callback types
align with SQLAlchemy.
Source: Linters/SAST tools
76d6247 to
f2a6c04
Compare
1 participant
Summary by CodeRabbit
Inputnow support using a shared registry instance.