hello friends,
i've done a couple of extensions i found useful, and since their
simplicity i'm sure they might also serve as examples.
all my scarce extensions knowledge comes from analyze of elixir and
existing extensions, and these help with simplifying my code (without
giving up functionality) and avoid patching elixir in order to achieve
some pervert solutions.
please enjoy my work as, we all enjoy the work of others, especially
in the freeware world.
1. inverse_orphans:
in persons.py you have
class Person(Entity):
id = Field(Integer, pk=True)
name = Field(Unicode(20))
in books.py you have
from person import *
class Book(Entity):
id = Field(Integer, pk=True)
name = Field(Unicode(50))
author = ManyToOne('Person', inverse='books_authored')
inverse_orphans()
since a Person is a quite generic entity, you can't be sure he should
author books or cut throats or both, but in a suitable application, i
might ask for the books a person has authored etc. hence, will
silently add Person a books_authored OneToMany fully functional
relationship.
2. record_owned
is an extension made after acts_as_versioned, just that it will add
the facility of auto recording the person that inserted and updated
each record in an Entity.
there is also some advance over the versioned extension, in that you
may also fill manually defined fields, to allow them be part of a
relationship and/or have additional information.
i will also patch the versioned extension to allow this customization,
keeping the present functionality as default.
thanks for your consideration,
alex
--~--~---------~--~----~------------~-------~--~----~
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
-~----------~----~----~----~------~----~------~--~---
"""
inverse_orphans Elixir Statement Generator
===============
inverse_orphans
===============
i am using an identity model module from a third party,
having, among others, an User class.
this model file might well be upgraded by it's supplier.
in another model file, which imports the identity model entities,
i have a Person class, which has a ManyToOne relationship with the User table.
However i'd like to also know the Person which references a given User, adding
OneToMany relationships to User, will need to be maintained everytime the
supplier upgrades identity model.
to implement this, i am giving an inverse name on the ManyToOne side, and
adding an after_mapper action to create the OneToMany relationship.
"""
from elixir.statements import Statement
from elixir import OneToOne, OneToMany, ManyToOne, ManyToMany
__all__ = ['inverse_orphans']
__doc_all__ = __all__
#TODO: inherit from entity builder
class inverse_orphans_entity_builder(object):
"""An inverse_orphans Elixir Statement object"""
def __init__(self, entity):
self.entity = entity
def before_table(self):
'''
if we name an inverse relationship which is not already defined on the
target, here we create the inverse relationship on the target.
should run this for each relationship property.
'''
for r in self.entity._descriptor.relationships:
desc = r.target._descriptor
if r.inverse_name and desc.find_relationship(r.inverse_name) is None:
if type(r) == ManyToOne:
if 'unique' in r.column_kwargs and r.column_kwargs['unique']:
invclass = OneToOne
else: invclass = OneToMany
elif type(r) == ManyToMany:
invclass = ManyToMany
else: invclass = None
else: invclass = None
if invclass:
invprop = invclass(r.entity.__name__, inverse=r.name)
invprop._target = r.entity
setattr(r.target, r.inverse_name, invprop)
invprop.attach(r.target, r.inverse_name)
inverse_orphans = Statement(inverse_orphans_entity_builder)
"""
A plugin for Elixir, which adds and maintains columns for the user that creates,
and respectivelly updates, a record.
The user may be supplied directly, or a function getting an argument to fetch
the user may be supplied. the argument (getter_param) is optional, defaulting
to None.
The field names may be optionally supplied in the column_field_names parameter.
If column_field_names is None (by default), ['record_inserter', 'record_updater'],
will be assumed. The field names may be already defined, in which case they will
be used as they are. This feature will allow those fields belonging to entity
relationships. If only one field will be supplied, the same field name will be
set for inserter and updater, the effect will be to keep the last toucher. In
combination with the versioned extension, the inserter will be the user who
touched version 1.
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, ['toucher'])
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):
owner_columns = instance.__class__.__column_field_names__
setattr(instance, owner_columns[1], None)
setattr(instance, owner_columns[0], get_current_user(instance))
return EXT_CONTINUE
def before_update(self, mapper, connection, instance):
owner_columns = instance.__class__.__column_field_names__
values = instance.table.select(get_entity_where(instance)).execute().fetchone()
setattr(instance, owner_columns[0], values[owner_columns[0]])
setattr(instance, owner_columns[1], 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, column_field_names=None):
entity._descriptor.add_mapper_extension(owned_mapper_extension)
self.entity = entity
self.add_mapper_extension(owned_mapper_extension)
entity.__current_user_column_type__ = column_type
if column_field_names in [None, []]:
column_field_names = ['record_inserter', 'record_updater']
entity.__column_field_names__ = column_field_names
if len(entity.__column_field_names__) < 2:
entity.__column_field_names__.append(entity.__column_field_names__[0])
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 inserted or updated the record, if fields
# don't exist yet. this to allow their definition as relationships etc.
for fieldname in self.entity.__column_field_names__:
if fieldname not in [col.name for col in self.entity._descriptor.columns]:
self.entity._descriptor.add_column(Column(fieldname, self.entity.__current_user_column_type__))
acts_as_owned = Statement(OwnedEntityBuilder)
from elixir import *
from inverse_orphans import inverse_orphans
metadata.bind = "sqlite:///:memory:"
metadata.bind.echo = True
class Movie(Entity):
title = Field(Unicode(30))
year = Field(Integer)
description = Field(Text)
director = ManyToOne('Director', inverse='movies')
main_genre = ManyToOne('Genre', inverse='movies_with_main')
inverse_orphans()
def __repr__(self):
return '<Movie "%s" (%d)>' % (self.title, self.year)
class Director(Entity):
name = Field(Unicode(60))
#movies = OneToMany('Movie', inverse='director')
def __repr__(self):
return '<Director "%s">' % self.name
class Genre(Entity):
name = Field(Unicode(15), primary_key=True)
abbrev = Field(Unicode(15), primary_key=True)
movies = ManyToMany('Movie', inverse='genres')
inverse_orphans()
def __repr__(self):
return '<Genre "%s">' % self.name
setup_all(True)
#create_all()
scifi = Genre(name="Science-Fiction", abbrev='sf')
rscott = Director(name="Ridley Scott")
glucas = Director(name="George Lucas")
alien = Movie(title="Alien", year=1979, director=rscott,
genres=[scifi, Genre(name="Horror", abbrev='hr')],
main_genre=scifi)
swars = Movie(title="Star Wars", year=1977, director=glucas, genres=[scifi],
main_genre=scifi)
brunner = Movie(title="Blade Runner", year=1982, director=rscott, genres=[scifi],
main_genre=scifi)
#rscott.movies.append(brunner)
#rscott.movies.append(alien)
#swars.director = glucas
print rscott.movies
print glucas.movies
print alien.genres
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)
inserter = Field(Unicode(30)) # the field may be manually defined
# acts_as_owned(Unicode(17), get_current_user, 12, ['inserter', 'updater'])
acts_as_owned(Unicode(17), get_current_user, column_field_names=['toucher'])
# 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" version %d, since %s>' % \
# (self.title, self.year, self.inserter, self.updater, self.version, str(self.timestamp))
return '<Movie "%s" (%d) touched by "%s" version %d, since %s>' % \
(self.title, self.year, self.toucher, self.version, str(self.timestamp))
# return '<Movie "%s" (%d) inserted by "%s", updated by "%s" version %d, since %s>' % \
# (self.title, self.year, self.record_inserter, self.record_updater, self.version, str(self.timestamp))
setup_all(True)
create_all()
alien = Movie(title=u'Alien', year=1979)
#this way, the inserter will not be set
#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
