Slowly I am becoming bald...
Here is a stand-alone script that reproduces the behaviour. Code is a bit
verbose, had copied it from my actual project, and it uses PostgreSQL
database, and you have to issue an "CREATE SCHEMA sample" beforehand. Sorry
about that.
The script's output is below. When I inspect the table with psql, the
"deleted" record is still present. And SQLAlchemy has logged the statements
without errors.
BEGIN TEST
=== COUNTING BEFORE ==============================
2011-08-17 01:33:41,779 INFO sqlalchemy.engine.base.Engine BEGIN (implicit)
2011-08-17 01:33:41,779 INFO sqlalchemy.engine.base.Engine SELECT
count(sample.rolemember.id) AS count_1
FROM sample.rolemember
2011-08-17 01:33:41,779 INFO sqlalchemy.engine.base.Engine {}
=== FETCHING ID TO DELETE ==============================
2011-08-17 01:33:41,780 INFO sqlalchemy.engine.base.Engine SELECT
sample.rolemember.id AS sample_rolemember_id, sample.rolemember.ctime AS
sample_rolemember_ctime, sample.rolemember.mtime AS sample_rolemember_mtime,
sample.rolemember.principal_id AS sample_rolemember_principal_id,
sample.rolemember.role_id AS sample_rolemember_role_id,
sample.rolemember.owner AS sample_rolemember_owner, sample.rolemember.editor
AS sample_rolemember_editor
FROM sample.rolemember ORDER BY sample.rolemember.id
LIMIT %(param_1)s OFFSET %(param_2)s
2011-08-17 01:33:41,780 INFO sqlalchemy.engine.base.Engine {'param_1': 1,
'param_2': 0}
=== DELETING ==============================
2011-08-17 01:33:41,781 INFO sqlalchemy.engine.base.Engine DELETE FROM
sample.rolemember WHERE sample.rolemember.id = %(id_1)s
2011-08-17 01:33:41,781 INFO sqlalchemy.engine.base.Engine {'id_1': 7L}
=== COUNTING AFTER ==============================
2011-08-17 01:33:41,782 INFO sqlalchemy.engine.base.Engine SELECT
count(sample.rolemember.id) AS count_1
FROM sample.rolemember
2011-08-17 01:33:41,782 INFO sqlalchemy.engine.base.Engine {}
=== RESULT ==============================
Total before: 7
Deleted id: 7
Total after: 6
--
You received this message because you are subscribed to the Google Groups
"sqlalchemy" group.
To view this discussion on the web visit
https://groups.google.com/d/msg/sqlalchemy/-/yIobsELwojwJ.
To post to this group, send email to sqlalchemy@googlegroups.com.
To unsubscribe from this group, send email to
sqlalchemy+unsubscr...@googlegroups.com.
For more options, visit this group at
http://groups.google.com/group/sqlalchemy?hl=en.
# -*- coding: utf-8 -*-
import transaction
from sqlalchemy import *
from sqlalchemy.orm import scoped_session
from zope.sqlalchemy import ZopeTransactionExtension
from sqlalchemy.orm import sessionmaker
from sqlalchemy.orm import relationship, backref
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.ext.declarative import declared_attr
from sqlalchemy.exc import IntegrityError
DbSession = scoped_session(sessionmaker(extension=ZopeTransactionExtension()))
DbBase = declarative_base()
DbEngine = create_engine('postgresql://pyramid_pym:pyramid_pym@localhost:5434/pyramid_pym', echo=True)
DbSession.configure(bind=DbEngine)
DbBase.metadata.bind = DbEngine
SYSTEM_UID = 1
ROOT_UID = 2
SAMPLE_DATA_UID = 97
UNIT_TESTER_UID = 98
NOBODY_UID = 99
SYSTEM_GID = 1
WHEEL_GID = 2
USERS_GID = 3
EVERYONE_GID = 99
class PymMixin(object):
id = Column(BigInteger, primary_key=True, nullable=False)
ctime = Column(DateTime, server_default=func.current_timestamp(), nullable=False)
@declared_attr
def owner(cls):
return Column(BigInteger,
ForeignKey("sample.principal.id", onupdate="CASCADE", ondelete="RESTRICT"),
nullable=False)
mtime = Column(DateTime, nullable=True)
@declared_attr
def editor(cls):
return Column(BigInteger,
ForeignKey("sample.principal.id", onupdate="CASCADE", ondelete="RESTRICT"),
nullable=True)
class RoleMember(DbBase, PymMixin):
__tablename__ = "rolemember"
__table_args__ = (
UniqueConstraint('role_id', 'principal_id'),
{'schema': 'sample'}
)
principal_id = Column(BigInteger,
ForeignKey("sample.principal.id", onupdate="CASCADE", ondelete="CASCADE"),
nullable=False)
role_id = Column(BigInteger,
ForeignKey("sample.role.id", onupdate="CASCADE", ondelete="CASCADE"),
nullable=False)
role = relationship('Role', backref='principal_assocs')
principal = relationship('Principal',
primaryjoin='Principal.id==RoleMember.principal_id',
backref='role_assocs' )
def __str__(self):
return "<RoleMember(id={0}, role_id='{1}', principal_id='{2}'>".format(
self.id, self.role_id, self.principal_id)
class Principal(DbBase, PymMixin):
__tablename__ = "principal"
__table_args__ = {'schema': 'sample'}
principal = Column(Unicode(255), nullable=False, index=True, unique=True)
email = Column(Unicode(128), nullable=False)
def __str__(self):
return "<Principal(id={0}, principal='{1}', email='{2}'>".format(
self.id, self.principal, self.email)
class Role(DbBase, PymMixin):
__tablename__ = "role"
__table_args__ = {'schema': 'sample'}
name = Column(Unicode(64), nullable=False, index=True, unique=True)
notes = Column(Unicode(255))
def __str__(self):
return "<Role(id={0}, name='{1}'>".format(
self.id, self.name)
def create_all():
DbBase.metadata.create_all(DbEngine)
try:
populate()
except IntegrityError as e:
transaction.abort()
print "*** ERROR: ", e
def populate():
session = DbSession()
# Principal 'system'
# Need to create this principal first to have an owner for following entities
p = Principal(principal=u'system',email=u'system@localhost')
p.id = SYSTEM_UID
p.owner = SYSTEM_UID
session.add(p)
session.flush()
# Create roles
r = Role(name=u'system')
r.id = SYSTEM_GID
r.owner = SYSTEM_UID
session.add(r)
r_wheel = Role(name=u'wheel')
r_wheel.id = WHEEL_GID
r_wheel.owner = SYSTEM_UID
session.add(r_wheel)
r_users = Role(name=u'users')
r_users.id = USERS_GID
r_users.owner = SYSTEM_UID
session.add(r_users)
# This role should not have members.
# Not-authenticated users are automatically member of 'everyone'
r = Role(name=u'everyone')
r.id = EVERYONE_GID
r.notes = u'Everyone (including not authenticated users)'
r.owner = SYSTEM_UID
session.add(r)
session.flush() # Needed to set ids of above objects
# Put 'system' into its roles
e = RoleMember(role_id=r_users.id, principal_id=p.id)
e.owner = SYSTEM_UID
session.add(e)
e = RoleMember(role_id=r_wheel.id, principal_id=p.id)
e.owner = SYSTEM_UID
session.add(e)
session.flush()
# Principal 'root'
p = Principal(principal=u'root',email=u'root@localhost')
p.id = ROOT_UID
p.owner = SYSTEM_UID
session.add(p)
session.flush()
e = RoleMember(role_id=r_users.id, principal_id=p.id)
e.owner = SYSTEM_UID
session.add(e)
e = RoleMember(role_id=r_wheel.id, principal_id=p.id)
e.owner = SYSTEM_UID
session.add(e)
session.flush()
# Principal 'sample_data'
# This principal is not member of any role
p = Principal(principal=u'sample_data',email=u'sample_data@localhost')
p.id = SAMPLE_DATA_UID
p.owner = SYSTEM_UID
session.add(p)
session.flush()
# Principal 'unit_tester'
p = Principal(principal=u'unit_tester',email=u'unit_tester@localhost')
p.id = UNIT_TESTER_UID
p.owner = SYSTEM_UID
session.add(p)
session.flush()
e = RoleMember(role_id=r_users.id, principal_id=p.id)
e.owner = SYSTEM_UID
session.add(e)
e = RoleMember(role_id=r_wheel.id, principal_id=p.id)
e.owner = SYSTEM_UID
session.add(e)
session.flush()
# Principal 'nobody'
# This principal is not member of any roles
# Not-authenticated users are automatically 'nobody'
p = Principal(principal=u'nobody',email=u'nobody@localhost')
p.id = NOBODY_UID
p.owner = SYSTEM_UID
session.add(p)
session.flush()
# Regular users have ID > 100
session.execute('ALTER SEQUENCE sample.principal_id_seq RESTART WITH 101')
# Regular roles have ID > 100
session.execute('ALTER SEQUENCE sample.role_id_seq RESTART WITH 101')
session.flush()
# Principal 'Foo' for testing
p = Principal(principal=u'foo',email=u'foo@localhost')
p.owner = SYSTEM_UID
e = RoleMember(role_id=r_users.id)
e.owner = SYSTEM_UID
p.role_assocs.append(e)
session.add(p)
session.flush()
transaction.commit()
if __name__ == '__main__':
import pprint
pp = pprint.PrettyPrinter(indent=4)
sess = DbSession()
create_all()
print "\n\n", "="*78, "\n\n"
print "BEGIN TEST"
print "\n=== COUNTING BEFORE", "="*30, "\n"
total_before = sess.query(func.count(RoleMember.id)).scalar()
print "\n=== FETCHING ID TO DELETE", "="*30, "\n"
e = sess.query(RoleMember).order_by(RoleMember.id).first()
deleted_id = e.id
print "\n=== DELETING", "="*30, "\n"
sess.query(RoleMember).filter(RoleMember.id==deleted_id).delete(synchronize_session=False)
print "\n=== COUNTING AFTER", "="*30, "\n"
total_after = sess.query(func.count(RoleMember.id)).scalar()
print "\n=== RESULT", "="*30, "\n"
print "Total before:", total_before
print "Deleted id:", deleted_id
print "Total after:", total_after
print "\nUse e.g. psql to inspect db table\n"
