Latest Version: 0.9.6.2

Elixir event statements (improved) by beachcoder


Language: Python
Tags: sqlalchemy, elixir
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
from elixir.statements     import Statement
from sqlalchemy.orm.mapper import MapperExtension

import types

def proxy_to_instance(name):
	def new_func(self, mapper, connection, instance):
		if hasattr(instance, name) : getattr(instance, name)()	
	return new_func

def mapper_extension_statement(name):
	class MappingProxyStatement(object):
		def __init__(self, entity, *callbacks):
			def proxy_method(self):
				for callback in callbacks:
					getattr(self, callback)()
			setattr(entity, '__%s__' %name, proxy_method)
			extensions = entity._descriptor.mapper_options.get('extension', [])
			if type(extensions) is not types.ListType: 
				extensions = [extensions]
			if ext_proxy not in extensions:
				extensions.append(ext_proxy)
				entity._descriptor.mapper_options['extension'] = extensions
	return MappingProxyStatement

class MapperExtensionProxy(MapperExtension):
		
	after_delete = proxy_to_instance('__after_delete__')
	after_insert = proxy_to_instance('__after_insert__')
	after_update = proxy_to_instance('__after_update__')
	before_delete = proxy_to_instance('__before_delete__')
	before_insert = proxy_to_instance('__before_insert__')
	before_update = proxy_to_instance('__before_update__')
			
ext_proxy = MapperExtensionProxy()

after_delete = Statement(mapper_extension_statement('after_delete'))			
after_insert = Statement(mapper_extension_statement('after_insert'))			
after_update = Statement(mapper_extension_statement('after_update'))			
before_delete = Statement(mapper_extension_statement('before_delete'))			
before_insert = Statement(mapper_extension_statement('before_insert'))			
before_update = Statement(mapper_extension_statement('before_update'))

if __name__ == '__main__':

	from elixir import *
	
	class Car(Entity):
		has_field('name', String(30))
		before_insert('before_insert_test')
		before_update('before_update_test')
		before_delete('before_delete_test')
		after_insert('after_insert_test')
		after_update('after_update_test')
		after_delete('after_delete_test')
		def before_insert_test(self): print "%s before insert" %self.name
		def before_update_test(self): print "%s before update" %self.name
		def before_delete_test(self): print "%s before delete" %self.name
		def after_insert_test(self): print "%s after insert" %self.name
		def after_update_test(self): print "%s after update" %self.name
		def after_delete_test(self): print "%s after delete" %self.name

	metadata.connect('sqlite:///test_events.db')
	metadata.drop_all()
	metadata.create_all()
	
	car = Car(name = 'Ferrari')
	car.flush()
	car.name = 'Lotus'
	car.flush()
	car.delete()
	car.flush()

Top