/usr/share/pyshared/elixir/ext/perform_ddl.py is in python-elixir 0.7.1-1.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
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 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 | '''
DDL statements for Elixir.
Entities having the perform_ddl statement, will automatically execute the
given DDL statement, at the given moment: ether before or after the table
creation in SQL.
The 'when' argument can be either 'before-create' or 'after-create'.
The 'statement' argument can be one of:
- a single string statement
- a list of string statements, in which case, each of them will be executed
in turn.
- a callable which should take no argument and return either a single string
or a list of strings.
In each string statement, you may use the special '%(fullname)s' construct,
that will be replaced with the real table name including schema, if unknown
to you. Also, self explained '%(table)s' and '%(schema)s' may be used here.
You would use this extension to handle non elixir sql statemts, like triggers
etc.
.. sourcecode:: python
class Movie(Entity):
title = Field(Unicode(30), primary_key=True)
year = Field(Integer)
perform_ddl('after-create',
"insert into %(fullname)s values ('Alien', 1979)")
preload_data is a more specific statement meant to preload data in your
entity table from a list of tuples (of fields values for each row).
.. sourcecode:: python
class Movie(Entity):
title = Field(Unicode(30), primary_key=True)
year = Field(Integer)
preload_data(('title', 'year'),
[(u'Alien', 1979), (u'Star Wars', 1977)])
preload_data(('year', 'title'),
[(1982, u'Blade Runner')])
preload_data(data=[(u'Batman', 1966)])
'''
from elixir.statements import Statement
from elixir.properties import EntityBuilder
from sqlalchemy import DDL
__all__ = ['perform_ddl', 'preload_data']
__doc_all__ = []
#
# the perform_ddl statement
#
class PerformDDLEntityBuilder(EntityBuilder):
def __init__(self, entity, when, statement, on=None, context=None):
self.entity = entity
self.when = when
self.statement = statement
self.on = on
self.context = context
def after_table(self):
statement = self.statement
if hasattr(statement, '__call__'):
statement = statement()
if not isinstance(statement, list):
statement = [statement]
for s in statement:
ddl = DDL(s, self.on, self.context)
ddl.execute_at(self.when, self.entity.table)
perform_ddl = Statement(PerformDDLEntityBuilder)
#
# the preload_data statement
#
class PreloadDataEntityBuilder(EntityBuilder):
def __init__(self, entity, columns=None, data=None):
self.entity = entity
self.columns = columns
self.data = data
def after_table(self):
all_columns = [col.name for col in self.entity.table.columns]
def onload(event, schema_item, connection):
columns = self.columns
if columns is None:
columns = all_columns
data = self.data
if hasattr(data, '__call__'):
data = data()
insert = schema_item.insert()
connection.execute(insert,
[dict(zip(columns, values)) for values in data])
self.entity.table.append_ddl_listener('after-create', onload)
preload_data = Statement(PreloadDataEntityBuilder)
|