On Saturday 05 January 2008 03:08:54 Michael Bayer wrote:
> On Jan 4, 2008, at 7:44 AM, Dave Harrison wrote:
> > On Friday 04 January 2008 23:32:21 Alexandre da Silva wrote:
> >>> Is there an easy way of flushing all objects that are categorised
> >>> as new in the session ??
> >>
> >> I think you can use session.clear(), it will remove objects from
> >> session, and persistent objects will stay on database.
> >>
> >> Att
> >>
> >> Alexandre
> >
> > I had been using session.clear up until now, but as of 0.4.2 I've
> > found that I get this error for alot of my tests,
> >
> >    object has no attribute '_sa_session_id'
>
> thats a bug.  can you please provide a reproducing test case ?   
> this is actually pretty serious and a new release will be available
> the moment we fix this.

Hey Mike,

Below is a minimal test case that always produces the below failure
for me under 0.4.2 but not under 0.4.1,

Cheers
Dave

======================================================================
ERROR: test_repr (__main__.uScheduledEvent)
----------------------------------------------------------------------
Traceback (most recent call last):
  File "./test_db_minimal.py", line 19, in tearDown
    db.session.clear()
  
File 
"/usr/local/lib/python2.5/site-packages/SQLAlchemy-0.4.2-py2.5.egg/sqlalchemy/orm/scoping.py",
 
line 74, in do
    return getattr(self.registry(), name)(*args, **kwargs)
  
File 
"/usr/local/lib/python2.5/site-packages/SQLAlchemy-0.4.2-py2.5.egg/sqlalchemy/orm/session.py",
 
line 577, in clear
    self._unattach(instance)
  
File 
"/usr/local/lib/python2.5/site-packages/SQLAlchemy-0.4.2-py2.5.egg/sqlalchemy/orm/session.py",
 
line 1039, in _unattach
    if instance._sa_session_id == self.hash_key:
AttributeError: 'MonthlyEvent' object has no attribute '_sa_session_id'

----------------------------------------------------------------------

----------------------------
---- dbapi_minimal.py ------
----------------------------

import datetime, uuid
from sqlalchemy import *
from sqlalchemy.orm import *

session = scoped_session(sessionmaker(autoflush=False, 
transactional=True))
mapper = session.mapper
metadata = MetaData()

## Static variables and mappings
EventType_Daily = 0
EventType_Weekly = 1
EventType_Monthly = 2
EventType_Yearly = 3

eventTable = Table(
    'event',
    metadata,
    Column('id', Integer, primary_key=True),
    Column('name', String(100), nullable=False, unique=True),
    Column('type', Integer, nullable=False),
    Column('action', BLOB),         # Pickled arguments for running an 
action
    Column('step', Integer, nullable=False, default=1),
    Column('fromDate', DateTime),           # None indicates now
    Column('untilDate', DateTime),          # None indicates forever
    Column('monthly_dayofweek', Boolean),
)

scheduledEventTable = Table(
    'scheduledevent',
    metadata,
    Column('id', Integer, primary_key=True),
    Column('event_id', Integer, ForeignKey('event.id')),
    Column('runtime', DateTime),
)

class Event(object):
    def __init__(self, name, action, step=1, fromDate=None, 
untilDate=None):
        """
            action    : Pickled ActionArgument object for running the 
event
            fromDate  : First time to run this event
            untilDate : Time at which this event is no longer valid
        """
        self.name = name
        self.action = action
        self.step = step
        self.fromDate = fromDate
        self.untilDate = untilDate

    def next(self, last):
        """
            last    : most recent timestamp when this event was executed
        """
        raise NotImplementedError("Events must provide a next()")

    def __repr__(self):
        return "Event type:%s id:%s" % (self.typeName, self.id)


class MonthlyEvent(Event):
    """
        MonthlyEvent has an additional init arg of dayofweek.
        E.G if (dayofweek == True) then do every first Monday
        E.G if (dayofweek == False) then do every nth day of the month
        [beware choosing a day > 28 lest some months skip it]
    """
    def __init__(self, name, action, step=1, fromDate=None, 
untilDate=None, dayofweek=True):
        Event.__init__(self, name, action, step, fromDate, untilDate)
        self.dayofweek = dayofweek

    def next(self, last):
        monthQuot,monthRem = divmod((last.month + self.step), 12)
        # Div cycles to 0, so we need to handle this differently
        # 0 indicates December in the current year
        if monthRem == 0:
            monthRem = 12
            monthQuot = 0

        incMonth = monthRem
        incYear = last.year + monthQuot
        newTS = datetime.datetime(
            year = incYear,
            month = incMonth,
            day = last.day,
            hour = self.fromDate.hour,
            minute = self.fromDate.minute,
        )
        if newTS > self.untilDate:
            return None
        return newTS


class ScheduledEvent(object):
    """
        This is an event that is waiting for its runtime to arrive
    """
    def __init__(self, event, runtime):
        self.event = event
        self.runtime = runtime

    def __repr__(self):
        return "ScheduledEvent(%s) @ %s" % (self.id, str(self.runtime))


eventMapper = mapper(
    Event,
    eventTable,
    polymorphic_on=eventTable.c.type,
)
mapper(
    MonthlyEvent,
    inherits=eventMapper,
    polymorphic_identity = EventType_Monthly,
)

mapper(
    ScheduledEvent,
    scheduledEventTable,
    properties = {
        "event": relation(Event, lazy=False),
    },
)

def connect(uri):
    engine = create_engine(uri, strategy="threadlocal")
    metadata.bind = engine
    return engine

def create():
    metadata.create_all()

def drop():
    metadata.drop_all()

----------------------------
---- test_db_minimal.py ----
----------------------------

import unittest
import cStringIO, pickle, datetime
import dbapi_minimal as db
import unittest

class MemTester(unittest.TestCase):
    FIRST_TEST=True
    
    def setUpAll(self):
        db.connect('sqlite://memory')

    def setUp(self):
        if self.FIRST_TEST:
            self.setUpAll()
            self.FIRST_TEST=False
        db.create()

    def tearDown(self):
        db.session.clear()
        db.drop()

class uScheduledEvent(MemTester):
    data = {
        "name" : "Some event",
        "action" : pickle.dump({"foo": 1}, cStringIO.StringIO()),
        "step" : 4,
        "fromDate" : datetime.datetime(2007,11,30,9,30),
        "untilDate" : datetime.datetime(2010,12,20,9,30),
    }

    def test_add(self):
        e = db.MonthlyEvent(dayofweek = True, **self.data)
        se = db.ScheduledEvent(e, datetime.datetime.now())

    def test_repr(self):
        e = db.MonthlyEvent(dayofweek = True, **self.data)
        se = db.ScheduledEvent(e, datetime.datetime.now())

if __name__ == '__main__':
    unittest.main()

--~--~---------~--~----~------------~-------~--~----~
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
-~----------~----~----~----~------~----~------~--~---

Reply via email to