Hi everybody,
I'm trying to use Association Proxy in one of my classes relation to
itself. I've got a Task class and Tasks can be dependent to each other, and
I've also wanted to store the dependency type ('start-to-start',
'start-to-end' etc.), so AssociationProxy was the way I've gone. Any way my
problem is with events.
Here is an example illustrating the problem:
import logging
from sqlalchemy import event, Column, Integer, String, ForeignKey
from sqlalchemy.ext.associationproxy import association_proxy
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import relationship
logging.basicConfig()
logger = logging.getLogger(__name__)
logger.setLevel(logging.DEBUG)
Base = declarative_base()
class Task(Base):
__tablename__ = 'Tasks'
id = Column(Integer, primary_key=True)
name = Column(String)
status = Column(String)
depends = association_proxy(
'task_depends_to', # outgoing_edges
'to_node', # to_node
creator=lambda n: TaskDependent(to_node=n)
)
dependent_of = association_proxy(
'task_dependent_of', # incoming_edges
'from_node', # from_node
creator=lambda n: TaskDependent(from_node=n)
)
task_depends_to = relationship(
'TaskDependent',
cascade="all, delete-orphan",
primaryjoin='Tasks.c.id==Task_Dependencies.c.from_node_id'
)
task_dependent_of = relationship(
'TaskDependent',
cascade="all, delete-orphan",
primaryjoin='Tasks.c.id==Task_Dependencies.c.to_node_id'
)
def __init__(self, name='', depends=None, status='WFD'):
self.name = name
self.status = status
if depends is None:
depends = []
self.depends = depends
def __repr__(self):
return "<%s (Task)>" % self.name
def update_status_with_dependent_statuses(self):
if not self.depends:
self.status = 'RTS'
class TaskDependent(Base):
__tablename__ = "Task_Dependencies"
from_node_id = Column(
Integer,
ForeignKey("Tasks.id"),
primary_key=True
)
from_node = relationship(
Task,
back_populates='task_depends_to',
primaryjoin='Task.id==TaskDependent.from_node_id',
)
to_node_id = Column(
Integer,
ForeignKey("Tasks.id"),
primary_key=True
)
to_node = relationship(
Task,
back_populates='task_dependent_of',
primaryjoin="TaskDependent.to_node_id==Task.id",
)
gap = Column(Integer)
def __init__(self, from_node=None, to_node=None, dependency_type=None,
gap=0, gap_model='length'):
self.to_node = to_node
self.from_node = from_node
self.dependency_type = dependency_type
self.gap = gap
@event.listens_for(Task.task_depends_to, 'remove', propagate=True)
def removed_a_dependency(task, task_dependent, initiator):
"""Runs when a task is removed from another tasks dependency list.
"""
# update task status with dependencies
logger.debug('inside removed_a_dependency')
logger.debug('task : %s' % task)
logger.debug('task.depends : %s' % task.depends)
logger.debug('task_dependent : %s' % task_dependent)
logger.debug('task_dependent.to_node : %s' % task_dependent.to_node)
logger.debug('task_dependent.from_node: %s' % task_dependent.from_node)
logger.debug('initiator : %s' % initiator)
task.update_status_with_dependent_statuses()
if __name__ == '__main__':
taskA = Task(name='TaskA')
taskB = Task(name='TaskB')
taskC = Task(name='TaskC', depends=[taskA, taskB])
assert taskC.status == 'WFD'
taskC.depends = []
assert taskC.depends == []
assert taskC.status == 'RTS'
Outputs:
DEBUG:__main__:inside removed_a_dependency
DEBUG:__main__:task : <TaskC (Task)>
DEBUG:__main__:task.depends : [<TaskA (Task)>, <TaskB (Task)>]
DEBUG:__main__:task_dependent : <__main__.TaskDependent object at
0x1fa3c10>
DEBUG:__main__:task_dependent.to_node : <TaskA (Task)>
DEBUG:__main__:task_dependent.from_node: None
DEBUG:__main__:initiator : <sqlalchemy.orm.attributes.Event
object at 0x1fa3f90>
DEBUG:__main__:inside removed_a_dependency
DEBUG:__main__:task : <TaskC (Task)>
DEBUG:__main__:task.depends : [<TaskA (Task)>, <TaskB (Task)>]
DEBUG:__main__:task_dependent : <__main__.TaskDependent object at
0x1fa3cd0>
DEBUG:__main__:task_dependent.to_node : <TaskB (Task)>
DEBUG:__main__:task_dependent.from_node: None
DEBUG:__main__:initiator : <sqlalchemy.orm.attributes.Event
object at 0x1fa3f90>
Traceback (most recent call last):
File "/home/eoyilmaz/association_proxy_remove_event.py", line 121, in
<module>
assert taskC.status == 'RTS'
AssertionError
So, TaskC depends both to TaskA and TaskB. When I set taskC.depends to []
the remove event successfully triggered, and the
Task.update_status_with_dependent_statuses() method is run. And at the end
taskC.depends list is [].
The problem is; even when the second event triggered I still see [TaskA,
TaskB] in task.depends, where it should at least be [TaskB] in second run.
Therefore the update_status_with_dependent_statuses() method always sees
some elements in self.depends and is not able to set the status to RTS.
Is that making sense? Am I missing something in the documentation?
Thanks,
E.Ozgur Yilmaz
eoyilmaz.blogspot.com
--
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 http://groups.google.com/group/sqlalchemy.
For more options, visit https://groups.google.com/groups/opt_out.
