Here is a test case for the bug. The bug only manifests itself if the
transaction that slices the list also modifies all the remaining objects in the
list.
Barry
----- Original Message ----
From: Michael Bayer <[EMAIL PROTECTED]>
To: sqlalchemy@googlegroups.com
Sent: Wednesday, October 24, 2007 5:01:43 PM
Subject: [sqlalchemy] Re: Problem when slicing a relation list
hey Barry -
again, can you please attach a working test case for this one ? attached is
mine, which tests this exact operation for four diferent kinds of relation()s -
one-to-many and many to many, with and without delete cascade on the relation.
passes for 0.3 (including 0.3.10) and 0.4.
-----Inline Attachment Follows-----
from sqlalchemy import *
from sqlalchemy.orm import *
def test(m2m=False, cascade=False, useclear=False):
engine = create_engine('sqlite://', echo=True)
meta = MetaData(engine)
a = Table('a', meta, Column('id', Integer, primary_key=True), Column('foo',
String(30)))
if m2m:
b = Table('b', meta, Column('id', Integer, primary_key=True),
Column('foo', String(30)))
else:
b = Table('b', meta, Column('id', Integer, primary_key=True),
Column('foo', String(30)), Column('a_id', Integer, ForeignKey('a.id')))
if m2m:
atob = Table('atob', meta, Column('a_id', Integer, ForeignKey('a.id')),
Column('b_id', Integer, ForeignKey('b.id')), )
else:
atob = None
class A(object):
def __init__(self, foo):
self.foo = foo
class B(object):
def __init__(self, foo):
self.foo = foo
if cascade:
use_cascade = "all, delete-orphan"
else:
use_cascade = "save-update"
mapper(A, a, properties={
'bs':relation(B, secondary=atob, cascade=use_cascade)
})
mapper(B, b)
meta.create_all()
a1 = A('a1')
a1.bs.append(B('b1'))
a1.bs.append(B('b2'))
a1.bs.append(B('b3'))
sess = create_session()
sess.save(a1)
sess.flush()
if m2m:
assert atob.count().scalar() == 3
else:
assert b.count(b.c.a_id == None).scalar() == 0
assert b.count().scalar() == 3
if useclear:
sess.clear()
a1 = sess.query(A).get(a1.id)
assert len(a1.bs) == 3
a1.bs = a1.bs[1:]
sess.flush()
if m2m:
assert atob.count().scalar() == 2
else:
assert b.count(b.c.a_id != None).scalar() == 2
if cascade:
assert b.count().scalar() == 2
else:
assert b.count().scalar() == 3
if useclear:
sess.clear()
a1 = sess.query(A).get(a1.id)
assert len(a1.bs) == 2
for m2m in (True, False):
for cascade in (True, False):
for useclear in (True, False):
test(m2m, cascade, useclear)
On Oct 24, 2007, at 4:18 PM, Barry Hart wrote:
I found a problem in SqlAlchemy 0.3.10 this week: slicing a list property
caused the whole association list to be deleted. For example, suppose I want to
remove the first 3 items from a list of related items:
my_obj.related = my_obj.related[3:]
When these changes were saved , my_obj.related was empty the next time it
loaded. This code, however, worked correctly:
for i in range(0, 3):
del my_obj.related[0]
Barry
__________________________________________________
Do You Yahoo!?
Tired of spam? Yahoo! Mail has the best spam protection around
http://mail.yahoo.com
__________________________________________________
Do You Yahoo!?
Tired of spam? Yahoo! Mail has the best spam protection around
http://mail.yahoo.com
--~--~---------~--~----~------------~-------~--~----~
You received this message because you are subscribed to the Google Groups
"sqlalchemy" group.
To post to this group, send email to sqlalchemy@googlegroups.com
To unsubscribe from this group, send email to [EMAIL PROTECTED]
For more options, visit this group at
http://groups.google.com/group/sqlalchemy?hl=en
-~----------~----~----~----~------~----~------~--~---
from sqlalchemy import *
from sqlalchemy.ext.assignmapper import assign_mapper
import sqlalchemy
from sqlalchemy.ext import activemapper, sessioncontext
import datetime
engine = None
def create_engine():
global engine
engine = sqlalchemy.create_engine('sqlite://')
#engine = sqlalchemy.create_engine('postgres://postgres:[EMAIL
PROTECTED]:5432/testdb')
metadata.connect(engine)
def create_session():
return sqlalchemy.create_session(bind_to=engine)
def fuzzy_search(column, value):
"""Case insensitive search allowing partial string matches."""
return func.lower(column).like('%%%s%%' % value.lower())
metadata = activemapper.metadata
create_engine()
session = activemapper.Objectstore(create_session)
activemapper.objectstore = session
g_count = 10
##########################################################################
# Classes
##########################################################################
class ObjX(object):
def __init__(self):
self.y = ObjY()
def analyze(self):
#del self.y.zs[0] # Using this line works.
self.y.zs = self.y.zs[1:] # Using this line fails.
for i in range(0, len(self.y.zs)):
s = self.y.zs[i]
# Modify the other objects in the list. Without this, the test
still works.
s.units = 10
#
class ObjY(object): pass
class ObjZ(object): pass
##########################################################################
# Tables
##########################################################################
obj_x_table = Table(
'obj_x', metadata,
Column('id', Integer, primary_key=True),
Column('desc', Unicode(255), default="")
)
obj_y_table = Table(
'obj_y', metadata,
Column('id', Integer, primary_key=True),
Column('obj_x_id', Integer, ForeignKey('obj_x.id'))
)
obj_z_table = Table(
'predict_time_period', metadata,
Column('id', Integer, primary_key=True),
Column('obj_y_id', Integer, ForeignKey('obj_y.id')),
Column('start_date', DateTime),
Column('units', Float)
)
##########################################################################
# Mappings
##########################################################################
assign_mapper(session.context, ObjX, obj_x_table, properties=dict(
y=relation(ObjY, uselist=False)
))
assign_mapper(session.context, ObjY, obj_y_table, properties=dict(
zs=relation(ObjZ, order_by=obj_z_table.c.start_date,
cascade="all,delete-orphan")
))
assign_mapper(session.context, ObjZ, obj_z_table)
##########################################################################
# Functions
##########################################################################
def createX():
x = ObjX()
for i in range(0, g_count):
z = ObjZ()
z.units = i
x.y.zs.append(z)
return x
##########################################################################
# Main program
##########################################################################
obj_x_table.create(checkfirst=True)
obj_y_table.create(checkfirst=True)
obj_z_table.create(checkfirst=True)
x = createX()
assert len(x.y.zs) == g_count
session.flush()
session.clear()
x = ObjX.select()[0]
assert len(x.y.zs) == g_count # Starts with g_count objects
x.analyze() # Slices one object from the list
assert len(x.y.zs) == g_count - 1 # Check the count again
session.flush()
session.clear()
x = ObjX.select()[0]
assert len(x.y.zs) == g_count - 1, 'Expected %d, got %d' % (g_count - 1,
len(x.y.zs))
