Hi, I'm fairly new to sqlalchemy and I noticed that I've written some utility
functions which I seek advice upon (acting on the assumption that they're
fairly silly and somebody must just enlighten my ways :D).
Thanks for having a look at the atachment,
Florian
# The following two functions help me define a mapper
# which I couldn't do otherwise (due to inheritance)
def find_fk_column(table, parent_table):
for column in parent_table.columns:
foreign_key = column.foreign_key
if foreign_key:
if foreign_key.column.table == table:
return column
def add_assoc(name, own, other, on_table):
table = own.mapper.tables[-1]
other_table = other.mapper.tables[-1]
on_inter = table.c.id == find_fk_column(table, on_table)
on_assoc = other_table.c.id == find_fk_column(other_table, on_table)
def do_assoc(self):
on_id = table.c.id==self.id
on = on_id & on_inter & on_assoc
return other.mapper.select(on)
setattr(own, name, property(do_assoc))
# I use add_assoc like:
add_assoc('pages', Linklist, Page, model.linklist_page)
# The following function is what I use for amending columns (insert or update)
# * It customly binds the params in order not to exceed maximum identifier
length
# * It prevents updating if all fetched data is equivalent to the existing data
# * It returns the id of the fetched table
def amend(table, keys, **data):
if isinstance(keys, str):
keys = [keys]
params = dict((key,data[key]) for key in keys)
clauses = [table.c[name]==sqlalchemy.bindparam(name,
type=table.c[name].type) for name in keys]
where = sqlalchemy.and_(*clauses)
sql = table.select(where)
result = sql.execute(**params).fetchone()
if result:
is_different = [(name,value, result[name]) for name, value in
data.items() if value != result[name]]
if len(is_different):
table.update(where).execute(**data)
return result.id
else:
insert_sql = table.insert()
insert_sql.execute(**data)
return table.engine.last_inserted_ids()[0]
# I use it like:
amend(sometable, ['foo', 'bar'], foo=1, bar='asdf', blubber=3)
