hello gaetan,
thanks for your reply.
attached are a small test program to demonstrate the problem, along
with an additional extension in the same spirit with versioned, which
i'd definitely like you to have and eventually publish.
i found the same problem with elixir 0.5.2 and 0.6.1, and sa 0.4.6 and 0.5.0b3.
best regards,
alex
On Sun, Aug 31, 2008 at 11:34 AM, Gaetan de Menten <[EMAIL PROTECTED]> wrote:
>
> On Sun, Aug 31, 2008 at 8:09 AM, alex bodnaru <[EMAIL PROTECTED]> wrote:
>>
>> hello friends,
>>
>> i notice that the before_insert method in an orm MapperExtension
>> derived class
>
> Yes indeed.
>
>> is not being called at all, thus invalidating these extensions.
>>
>> any idea of how could i proceed?
>
> What do you mean? In which case is it not called?
> --
> Gaëtan de Menten
> http://openhex.org
>
> >
>
--~--~---------~--~----~------------~-------~--~----~
You received this message because you are subscribed to the Google Groups
"SQLElixir" 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/sqlelixir?hl=en
-~----------~----~----~----~------~----~------~--~---
"""
A plugin for Elixir, which adds and maintains columns for the user that creates,
and respectivelly updates, a record.
Those columns are: record_inserter, record_updater.
The user may be supplied directly, or a function getting an argument to fetch
the user may be supplied. the argument is optional, defaulting to None.
Example:
def get_current_user(param):
return current_user
class Movie(Entity):
title = Field(Unicode(30))
year = Field(Integer)
acts_as_owned(Unicode(17), get_current_user, 12)
This extension may well be complemented by the excellent 'versioned' extension.
"""
from sqlalchemy import Column, and_
from sqlalchemy.orm import mapper, MapperExtension, EXT_CONTINUE
from elixir.statements import Statement
from elixir.properties import EntityBuilder
__all__ = ['acts_as_owned']
__doc_all__ = []
#
# utility functions
#
def get_entity_where(instance):
clauses = []
for column in instance.table.primary_key.columns:
instance_value = getattr(instance, column.name)
clauses.append(column==instance_value)
return and_(*clauses)
def get_current_user(instance):
gcu = instance.__class__.__get_current_user__
if callable(gcu):
return gcu(instance.__getter_param__)
else:
return gcu
#
# a mapper extension to track the user that insert, or update a record
#
class OwnedMapperExtension(MapperExtension):
def before_insert(self, mapper, connection, instance):
print 'bi' # doesn't show :(
instance.record_inserter = get_current_user(instance)
instance.record_updater = None
return EXT_CONTINUE
def before_update(self, mapper, connection, instance):
print 'bu'
values = instance.table.select(get_entity_where(instance)).execute().fetchone()
instance.record_inserter = values['record_inserter']
instance.record_updater = get_current_user(instance)
return EXT_CONTINUE
owned_mapper_extension = OwnedMapperExtension()
#
# the acts_as_owned statement
#
class OwnedEntityBuilder(EntityBuilder):
def __init__(self, entity, column_type, get_current_user, getter_param=None):
entity._descriptor.add_mapper_extension(owned_mapper_extension)
self.entity = entity
self.add_mapper_extension(owned_mapper_extension)
entity.__column_type__ = column_type
if callable(get_current_user):
entity.__get_current_user__ = staticmethod(get_current_user)
entity.__getter_param__ = getter_param
else: entity.__get_current_user__ = get_current_user
def create_non_pk_cols(self):
# add columns for the user who create or update the record
self.entity._descriptor.add_column(Column('record_inserter', self.entity.__column_type__))
self.entity._descriptor.add_column(Column('record_updater', self.entity.__column_type__))
acts_as_owned = Statement(OwnedEntityBuilder)
from elixir import *
from elixir.ext.versioned import acts_as_versioned
from record_owner import acts_as_owned
metadata.bind = "sqlite:///:memory:"
#metadata.bind.echo = True
current_user = u'alex'
def get_current_user(param=None):
return current_user
class Movie(Entity):
title = Field(Unicode(30))
year = Field(Integer)
acts_as_owned(Unicode(17), get_current_user, 12)
# acts_as_versioned()
def __repr__(self):
return '<Movie "%s" (%d) inserted by "%s", updated by "%s">' % (self.title, self.year, self.record_inserter, self.record_updater)
# return '<Movie "%s" (%d)' % (self.title, self.year)
# return '<Movie "%s" (%d) version %d, since %s>' % (self.title, self.year, self.version, str(self.timestamp))
setup_all(True)
create_all()
#alien = Movie(title=u'Alien', year=1979)
i = Movie.table.insert()
i.execute(dict(title=u'Alien', year=1979))
alien = Movie.query.filter_by(title=u'Alien').one()
session.flush()
print alien
current_user = u'alex2'
alien.title = u'alien2'
session.flush()
alien = Movie.query.filter(Movie.title.like(u'%lien%')).one()
print alien
