My method is as follows; for an SQLAlchemy ORM object x, call: y =
deepcopy_sqla_object(x, session), with functions defined as below. It walks
relationships (as defined by each ORM class; if there is no "relationship"
then it won't proceed, which can be helpful), copying objects, and then
rebuilds the relationships within the copies. However, it requires that all
ORM classes so traversed accept an __init__() call with no parameters [so
in practice: def __init__(self, *kwargs): ... self.something =
kwargs.pop('something', somedefault)].
from sqlalchemy.inspection import inspect
from sqlalchemy.orm import class_mapper
def copy_sqla_object(obj, omit_fk=True):
"""
Given an SQLAlchemy object, creates a new object (FOR WHICH THE OBJECT
MUST SUPPORT CREATION USING __init__() WITH NO PARAMETERS), and copies
across all attributes, omitting PKs, FKs (by default), and relationship
attributes.
"""
cls = type(obj)
mapper = class_mapper(cls)
newobj = cls() # not: cls.__new__(cls)
pk_keys = set([c.key for c in mapper.primary_key])
rel_keys = set([c.key for c in mapper.relationships])
prohibited = pk_keys | rel_keys
if omit_fk:
fk_keys = set([c.key for c in mapper.columns if c.foreign_keys])
prohibited = prohibited | fk_keys
for k in [p.key for p in mapper.iterate_properties
if p.key not in prohibited]:
try:
value = getattr(obj, k)
setattr(newobj, k, value)
except AttributeError:
pass
return newobj
def deepcopy_sqla_object(startobj, session, flush=True):
"""
For this to succeed, the object must take a __init__ call with no
arguments. (We can't specify the required args/kwargs, since we are
copying
a tree of arbitrary objects.)
"""
objmap = {} # keys = old objects, values = new objects
# Pass 1: iterate through all objects. (Can't guarantee to get
# relationships correct until we've done this, since we don't know
whether
# or where the "root" of the PK tree is.)
stack = [startobj]
while stack:
oldobj = stack.pop(0)
if oldobj in objmap: # already seen
continue
newobj = copy_sqla_object(oldobj)
session.add(newobj)
objmap[oldobj] = newobj
insp = inspect(oldobj)
for relationship in insp.mapper.relationships:
related = getattr(oldobj, relationship.key)
if relationship.uselist:
stack.extend(related)
elif related is not None:
stack.append(related)
# Pass 2: set all relationship properties.
for oldobj, newobj in objmap.items():
insp = inspect(oldobj)
# insp.mapper.relationships is of type
# sqlalchemy.utils._collections.ImmutableProperties, which is
basically
# a sort of AttrDict.
for relationship in insp.mapper.relationships:
# The relationship is an abstract object (so getting the
# relationship from the old object and from the new, with e.g.
# newrel = newinsp.mapper.relationships[oldrel.key],
# yield the same object. All we need from it is the key name.
related_old = getattr(oldobj, relationship.key)
if relationship.uselist:
related_new = [objmap[r] for r in related_old]
elif related_old is not None:
related_new = objmap[related_old]
else:
related_new = None
setattr(newobj, relationship.key, related_new)
# Done
if flush:
session.flush()
return objmap[startobj] # returns the new object matching startobj
--
You received this message because you are subscribed to the Google Groups
"sqlalchemy" group.
To unsubscribe from this group and stop receiving emails from it, send an email
to [email protected].
To post to this group, send email to [email protected].
Visit this group at https://groups.google.com/group/sqlalchemy.
For more options, visit https://groups.google.com/d/optout.
