Association proxy but for dict keys rather than dict values? #13228
Replies: 1 comment 2 replies
-
the dictionary's "key" and "value" both come from the object and the assoc proxy can proxy any attribute you want, so sure.
to do...what? assoc proxy provides a value. did you want to provide some kind of concatenated value? sure, just add an attribute on your mapped class that does that
no, and there's no reason you'd need to do that, just add a relationship that's a dictionary, you can have as many as you want for the same class.
absolutely
I had Claude take a look and the part where you want to do an "IStr creator" that's unique, without you having to do more verbose intervention each time, requires a pattern like UniqueObject so that you can pull up the appropriate IStr identity given some string value. given prompting based on that I was able to get it to produce a demonstration of all these techniques together, if you want to play with this (it also came up with an interesting way to use """Working example: Gizmo with interned-string-keyed properties.
Uses the UniqueObject recipe
(https://github.com/sqlalchemy/sqlalchemy/wiki/UniqueObject) to
deduplicate IStr instances via a per-session cache in session.info.
The association_proxy creator builds a throwaway IStr(s=k). A
@validates hook on Gizmo.props then calls IStr.as_unique() using
object_session(self) to swap it for the canonical instance, honoring
the unique constraint without any global state or seed_cache step.
"""
from __future__ import annotations
from sqlalchemy import create_engine
from sqlalchemy import ForeignKey
from sqlalchemy import select
from sqlalchemy import String
from sqlalchemy import UniqueConstraint
from sqlalchemy.ext.associationproxy import association_proxy
from sqlalchemy.ext.associationproxy import AssociationProxy
from sqlalchemy.orm import DeclarativeBase
from sqlalchemy.orm import keyfunc_mapping
from sqlalchemy.orm import Mapped
from sqlalchemy.orm import mapped_column
from sqlalchemy.orm import object_session
from sqlalchemy.orm import relationship
from sqlalchemy.orm import Session
from sqlalchemy.orm import validates
class Base(DeclarativeBase):
pass
# ------------------------------------------------------------------
# UniqueObject recipe (from SA wiki, adapted for 2.0-style queries)
# ------------------------------------------------------------------
def _unique(session, cls, hashfunc, queryfunc, constructor, arg, kw):
cache = session.info.setdefault("_unique_cache", {})
key = (cls, hashfunc(*arg, **kw))
if key in cache:
return cache[key]
with session.no_autoflush:
obj = session.scalars(
queryfunc(select(cls), *arg, **kw)
).first()
if obj is None:
obj = constructor(*arg, **kw)
session.add(obj)
cache[key] = obj
return obj
class UniqueMixin:
@classmethod
def unique_hash(cls, *arg, **kw):
raise NotImplementedError()
@classmethod
def unique_filter(cls, query, *arg, **kw):
raise NotImplementedError()
@classmethod
def as_unique(cls, session, *arg, **kw):
return _unique(
session, cls,
cls.unique_hash, cls.unique_filter,
cls, arg, kw,
)
# ------------------------------------------------------------------
# Models
# ------------------------------------------------------------------
class IStr(UniqueMixin, Base):
"""An interned string - unique, deduplicated string value."""
__tablename__ = "interned_string"
id: Mapped[int] = mapped_column(primary_key=True)
s: Mapped[str] = mapped_column(String(30), unique=True)
@classmethod
def unique_hash(cls, s):
return s
@classmethod
def unique_filter(cls, query, s):
return query.where(cls.s == s)
def __repr__(self) -> str:
return f"IStr({self.s!r})"
class Property(Base):
"""A key-value pair associated with a Gizmo."""
__tablename__ = "property"
__table_args__ = (UniqueConstraint("g_id", "k_id"),)
id: Mapped[int] = mapped_column(primary_key=True)
g_id: Mapped[int | None] = mapped_column(
ForeignKey("gizmo.id")
)
k_id: Mapped[int] = mapped_column(
ForeignKey("interned_string.id")
)
v: Mapped[str] = mapped_column(String(30))
gizmo: Mapped[Gizmo | None] = relationship(
back_populates="props",
)
ki: Mapped[IStr] = relationship(lazy="joined")
k: AssociationProxy[str] = association_proxy("ki", "s")
def __repr__(self) -> str:
return f"Property({self.ki!r}={self.v!r})"
class Gizmo(Base):
"""A Gizmo has key-value properties with interned string keys.
Access patterns:
gizmo.props -> dict[str, Property] (keyed by string)
gizmo.p -> dict[str, str] (string -> string)
"""
__tablename__ = "gizmo"
id: Mapped[int] = mapped_column(primary_key=True)
props: Mapped[dict[str, "Property"]] = relationship(
back_populates="gizmo",
collection_class=keyfunc_mapping(
lambda prop: prop.ki.s,
ignore_unpopulated_attribute=True,
),
cascade="all, delete-orphan",
)
p: AssociationProxy[dict[str, str]] = association_proxy(
"props",
"v",
creator=lambda k, v: Property(ki=IStr(s=k), v=v),
)
@validates("props", include_backrefs=False)
def _resolve_istr(self, key, prop):
"""Swap the Property's IStr for the unique instance from the
session cache / database, using object_session(self)."""
session = object_session(self)
if session is not None:
prop.ki = IStr.as_unique(session, s=prop.ki.s)
return prop
def main():
engine = create_engine("sqlite://", echo=True)
Base.metadata.create_all(engine)
with Session(engine) as session:
# Create a Gizmo and set properties via the string proxy.
g = Gizmo()
session.add(g)
g.p["color"] = "red"
g.p["size"] = "large"
g.p["weight"] = "heavy"
session.commit()
print("\n--- After initial commit ---")
print(f"g.p = {dict(g.p)}")
print(f"g.props = {dict(g.props)}")
# Second gizmo reuses existing IStr objects.
g2 = Gizmo()
session.add(g2)
g2.p["color"] = "blue" # reuses IStr("color")
g2.p["shape"] = "round" # new IStr
session.commit()
print("\n--- Second gizmo ---")
print(f"g2.p = {dict(g2.p)}")
# Verify IStr dedup: both gizmos share the same IStr row.
color_1 = g.props["color"].ki
color_2 = g2.props["color"].ki
assert color_1.id == color_2.id, (
"should share the same IStr row"
)
print(
f"\nIStr dedup OK: both use "
f"IStr id={color_1.id} for 'color'"
)
# Update an existing property value (no new IStr created).
g.p["color"] = "green"
session.commit()
print("\n--- After update ---")
print(f"g.p = {dict(g.p)}")
# All interned strings in the DB.
all_istrs = session.scalars(select(IStr)).all()
print("\n--- All interned strings ---")
for i in all_istrs:
print(f" {i}")
# Only 4 unique strings: color, size, weight, shape.
assert len(all_istrs) == 4
if __name__ == "__main__":
main() |
Beta Was this translation helpful? Give feedback.
-
|
Thank you for answering! And thank you for the Follow-up questions below:
What I actually need is something like this: class Gizmo(Base):
"""A Gizmo has key-value properties with interned string keys.
Access patterns:
Unchanged from existing codebase:
gizmo.props -> dict[IStr, Property] (keyed by interned string)
gizmo.pi -> dict[IStr, str] (interned string -> string)
New:
gizmo.p -> dict[str, str] (string -> string)
"""
Can you provide an example? If I understand correctly, an
When proxying a dict-like collection, the keys in the proxy's "view" dict are the same as the keys in the proxied dict—no transformation occurs. Keys are forwarded to the underlying dict unchanged whenever the proxy is accessed. What I want is for the proxy dict to have a different set of keys than the underlying dict. Each key in the proxy's dict is a "view" on a specific key in the underlying dict. There would be a bidirectional transformation to convert each of the proxy's keys to/from its corresponding key in the underlying dict. The proxy dict's values would remain unchanged. For example, I want to be able to do something like this: class Gizmo(Base):
props: Mapped[dict[IStr, Property]] = relationship(...)
...
props_str: AssociationProxy[dict[str, Propery]] = association_proxy(
"props",
# This is currently not supported.
key_attr="s",
key_creator=lambda k, v: v.ki, # TODO: assert v.ki.s == k
)
p: AssociationProxy[dict[str, str]] = association_proxy(
"props_str",
"v"
creator=lambda k, v: Property(ki=IStr(s=k), v=v),
)
This is an extension of my previous question. I want to be able to do something like this: def _prop_key_val_creator(k: str, v: str) -> tuple[IStr, Property]:
ki = IStr(s=k)
return ki, Property(ki=ki, v=v)
class Gizmo(Base):
props: Mapped[dict[IStr, Property]] = relationship(...)
...
p: AssociationProxy[dict[str, str]] = association_proxy(
"props",
# This is currently not supported.
key_attr="s",
value_attr="v",
key_value_creator=_prop_key_val_creator,
)
I want to be able to do something like this: class Gizmo(Base):
...
props_list: Mapped[list[Property]] = relationship(...)
p: AssociationProxy[dict[str, str]] = association_proxy(
"props_list",
# This is currently not supported.
key_func=lambda p: p.ki.s,
value_attr="v",
creator=lambda k, v: Property(ki=IStr(s=k), v=v),
)
But if I add multiple For example, if I do the following, changes to class Gizmo(Base):
props: Mapped[dict[IStr, Property]] = relationship(...)
props_str: Mapped[dict[str, Property]] = relationship(...)Maybe I could use a pair of |
Beta Was this translation helpful? Give feedback.
-
|
OK so the issue is that you dont want two different relationships. Usually, only one of the relationships is considered as "mutable" and the others are marked So OK, given that the association proxy is in fact a proxy, it does seem like a useful feature to provide hooks to have _AssociationDict select a particular attribute on the relationship-configured "key" to use as its own "key". that would be a lot more direct than setting up two relationships and keeping them in sync. as you've noted, the _AssociationDict doesnt have this hook right now. but we have all along had hooks to override what collection class is used overall, so you can make this happen now using a custom _AssociationDict subclass. this hook is called """Working example: Gizmo with IStr-keyed properties + string proxy.
The underlying relationship ``props`` is keyed by IStr objects
(``dict[IStr, Property]``). A custom ``proxy_factory`` subclasses
``_AssociationDict`` to translate string keys to/from IStr keys,
giving ``p`` a ``dict[str, str]`` view.
Uses the UniqueObject recipe
(https://github.com/sqlalchemy/sqlalchemy/wiki/UniqueObject) to
deduplicate IStr instances via a per-session cache in session.info.
"""
from __future__ import annotations
import operator
from typing import Any
from typing import Callable
from sqlalchemy import create_engine
from sqlalchemy import ForeignKey
from sqlalchemy import select
from sqlalchemy import String
from sqlalchemy import UniqueConstraint
from sqlalchemy.ext import associationproxy
from sqlalchemy.ext.associationproxy import association_proxy
from sqlalchemy.ext.associationproxy import AssociationProxy
from sqlalchemy.ext.associationproxy import _AssociationDict
from sqlalchemy.orm import DeclarativeBase
from sqlalchemy.orm import keyfunc_mapping
from sqlalchemy.orm import Mapped
from sqlalchemy.orm import mapped_column
from sqlalchemy.orm import object_session
from sqlalchemy.orm import relationship
from sqlalchemy.orm import Session
from sqlalchemy.orm import validates
class Base(DeclarativeBase):
pass
# ------------------------------------------------------------------
# UniqueObject recipe (from SA wiki, adapted for 2.0-style queries)
# ------------------------------------------------------------------
def _unique(session, cls, hashfunc, queryfunc, constructor, arg, kw):
cache = session.info.setdefault("_unique_cache", {})
key = (cls, hashfunc(*arg, **kw))
if key in cache:
return cache[key]
with session.no_autoflush:
obj = session.scalars(queryfunc(select(cls), *arg, **kw)).first()
if obj is None:
obj = constructor(*arg, **kw)
session.add(obj)
cache[key] = obj
return obj
class UniqueMixin:
@classmethod
def unique_hash(cls, *arg, **kw):
raise NotImplementedError()
@classmethod
def unique_filter(cls, query, *arg, **kw):
raise NotImplementedError()
@classmethod
def as_unique(cls, session, *arg, **kw):
return _unique(
session,
cls,
cls.unique_hash,
cls.unique_filter,
cls,
arg,
kw,
)
# ------------------------------------------------------------------
# Key-translating association proxy dict
# ------------------------------------------------------------------
class _KeyTranslatingAssociationDict(_AssociationDict):
"""_AssociationDict subclass that translates keys via a getter.
The underlying collection uses one key type (e.g. IStr);
this proxy presents a different key type (e.g. str).
"""
_key_getter: Callable[[Any], Any]
def __init__(
self,
lazy_collection,
creator,
getter,
setter,
parent,
key_getter,
):
super().__init__(
lazy_collection,
creator,
getter,
setter,
parent,
)
self._key_getter = key_getter
def _ukey(self, key):
"""Given a proxy key (str), find the underlying key (IStr)."""
for k in self.col:
if self._key_getter(k) == key:
return k
raise KeyError(key)
def __getitem__(self, key):
return self._get(self.col[self._ukey(key)])
def __setitem__(self, key, value):
try:
ukey = self._ukey(key)
except KeyError:
new_obj = self._create(key, value)
self.col.set(new_obj)
else:
self._set(self.col[ukey], key, value)
def __delitem__(self, key):
del self.col[self._ukey(key)]
def __contains__(self, key):
try:
self._ukey(key)
return True
except KeyError:
return False
def __iter__(self):
return (self._key_getter(k) for k in self.col)
def keys(self):
from collections.abc import KeysView
return KeysView(self)
def pop(self, key, *args):
ukey = self._ukey(key)
return self._get(self.col.pop(ukey, *args))
def popitem(self):
k, v = self.col.popitem()
return (self._key_getter(k), self._get(v))
def _key_translating_proxy_factory(
key_getter,
) -> associationproxy._ProxyFactoryProtocol:
"""Return a proxy_factory that builds _KeyTranslatingAssociationDict."""
def factory(lazy_collection, creator, value_attr, parent):
_getter = operator.attrgetter(value_attr)
def getter(obj):
return _getter(obj) if obj is not None else None
def setter(obj, k, v):
setattr(obj, value_attr, v)
return _KeyTranslatingAssociationDict(
lazy_collection,
creator,
getter,
setter,
parent,
key_getter=key_getter,
)
return factory
# ------------------------------------------------------------------
# Models
# ------------------------------------------------------------------
class IStr(UniqueMixin, Base):
"""An interned string - unique, deduplicated string value."""
__tablename__ = "interned_string"
id: Mapped[int] = mapped_column(primary_key=True)
s: Mapped[str] = mapped_column(String(30), unique=True)
@classmethod
def unique_hash(cls, s):
return s
@classmethod
def unique_filter(cls, query, s):
return query.where(cls.s == s)
def __hash__(self) -> int:
return hash(self.s)
def __eq__(self, other: object) -> bool:
if isinstance(other, IStr):
return self.s == other.s
return NotImplemented
def __repr__(self) -> str:
return f"IStr({self.s!r})"
class Property(Base):
"""A key-value pair associated with a Gizmo."""
__tablename__ = "property"
__table_args__ = (UniqueConstraint("g_id", "k_id"),)
id: Mapped[int] = mapped_column(primary_key=True)
g_id: Mapped[int | None] = mapped_column(ForeignKey("gizmo.id"))
k_id: Mapped[int] = mapped_column(ForeignKey("interned_string.id"))
v: Mapped[str] = mapped_column(String(30))
gizmo: Mapped[Gizmo | None] = relationship(
back_populates="props",
)
ki: Mapped[IStr] = relationship(lazy="joined")
k: AssociationProxy[str] = association_proxy("ki", "s")
def __repr__(self) -> str:
return f"Property({self.ki!r}={self.v!r})"
class Gizmo(Base):
"""A Gizmo has key-value properties with interned string keys.
Access patterns:
Unchanged from existing codebase:
gizmo.props -> dict[IStr, Property] (keyed by interned string)
gizmo.pi -> dict[IStr, str] (interned string -> string)
New:
gizmo.p -> dict[str, str] (string -> string)
"""
__tablename__ = "gizmo"
id: Mapped[int] = mapped_column(primary_key=True)
# Keyed by IStr objects.
props: Mapped[dict[IStr, "Property"]] = relationship(
back_populates="gizmo",
collection_class=keyfunc_mapping(
lambda prop: prop.ki,
ignore_unpopulated_attribute=True,
),
cascade="all, delete-orphan",
)
# IStr-keyed value proxy: dict[IStr, str].
pi: AssociationProxy[dict[IStr, str]] = association_proxy(
"props",
"v",
creator=lambda k, v: Property(ki=k, v=v),
)
# String-keyed view via custom proxy: dict[str, str].
p: AssociationProxy[dict[str, str]] = association_proxy(
"props",
"v",
creator=lambda k, v: Property(ki=IStr(s=k), v=v),
proxy_factory=_key_translating_proxy_factory(
key_getter=lambda istr: istr.s,
),
)
@validates("props", include_backrefs=False)
def _resolve_istr(self, key, prop):
"""Swap the Property's IStr for the unique instance from the
session cache / database, using object_session(self)."""
session = object_session(self)
if session is not None:
prop.ki = IStr.as_unique(session, s=prop.ki.s)
return prop
def main():
engine = create_engine("sqlite://", echo=True)
Base.metadata.create_all(engine)
with Session(engine) as session:
g = Gizmo()
session.add(g)
g.p["color"] = "red"
g.p["size"] = "large"
g.p["weight"] = "heavy"
session.commit()
print("\n--- After initial commit ---")
print(f"g.p = {dict(g.p)}")
print(f"g.pi = {dict(g.pi)}")
# Verify props is keyed by IStr objects.
for k, v in g.props.items():
assert isinstance(k, IStr), f"expected IStr, got {type(k)}"
print(f" props[{k!r}] = {v!r}")
# Verify pi is keyed by IStr, values are strings.
for k, v in g.pi.items():
assert isinstance(k, IStr), f"expected IStr, got {type(k)}"
assert isinstance(v, str), f"expected str, got {type(v)}"
print("pi keying OK: IStr keys, str values")
# Second gizmo reuses existing IStr objects.
g2 = Gizmo()
session.add(g2)
g2.p["color"] = "blue" # reuses IStr("color")
g2.p["shape"] = "round" # new IStr
session.commit()
print("\n--- Second gizmo ---")
print(f"g2.p = {dict(g2.p)}")
# Verify IStr dedup: both gizmos share the same IStr row.
color_1 = [p for p in g.props.values() if p.ki.s == "color"][0].ki
color_2 = [p for p in g2.props.values() if p.ki.s == "color"][0].ki
assert color_1.id == color_2.id, "should share the same IStr row"
print(
f"\nIStr dedup OK: both use " f"IStr id={color_1.id} for 'color'"
)
# Update an existing property value (no new IStr created).
g.p["color"] = "green"
session.commit()
print("\n--- After update ---")
print(f"g.p = {dict(g.p)}")
# All interned strings in the DB.
all_istrs = session.scalars(select(IStr)).all()
print("\n--- All interned strings ---")
for i in all_istrs:
print(f" {i}")
# Only 4 unique strings: color, size, weight, shape.
assert len(all_istrs) == 4
if __name__ == "__main__":
main() |
Beta Was this translation helpful? Give feedback.
Uh oh!
There was an error while loading. Please reload this page.
Uh oh!
There was an error while loading. Please reload this page.
-
When used on a dictionary-based relationship, the
association_proxyfunction creates anAssociationProxythat transforms the dictionary's value. My questions:list[Foo]via a dictionary-based association proxy where the dictionary key comes from theFoo.keyattribute that has a unique constraint and the value comes from theFoo.valueattribute)AssociationProxyproxy anotherAssociationProxyattribute?The immediate problem I'm trying to solve is similar to the following:
I could do something like this:
but then
Gizmo.props_str_keyedwould not be synchronized withGizmo.propsorProperty.gizmo. (I need to keepGizmo.propsas-is for now, so replacing it withattribute_keyed_dict("k")is not an option.)Beta Was this translation helpful? Give feedback.
All reactions