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 [email protected]
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
-~----------~----~----~----~------~----~------~--~---
