#!/usr/bin/env python
from sqlalchemy import (
Column,
Integer,
DateTime,
Unicode,
ForeignKey,
Boolean,
Sequence,
Table
)
from sqlalchemy import orm
from sqlalchemy.orm.collections import collection
from sqlalchemy import create_engine
from sqlalchemy.orm import (
scoped_session,
sessionmaker,
)
from sqlalchemy.ext.declarative import declarative_base
Base = declarative_base()
class User(Base):
__tablename__ = 'user'
__model__ = u'User'
id = Column(Integer,
Sequence('user_seq_id', optional=True),
primary_key=True)
username = Column(Unicode(128), nullable=False, unique=True)
def __init__(self, name):
self.username = name
class Permission(Base):
__tablename__ = 'permission'
__model__ = u'Permission'
id = Column(Integer,
Sequence('permission_seq_id', optional=True),
primary_key=True)
user_id = Column(Integer, ForeignKey(User.id))
user = orm.relationship('User')
permission = Column(Unicode(64), index=True)
assoc_id = Column(Integer,
ForeignKey('permission_assoc.id', ondelete="CASCADE"),
nullable=False)
member = property(lambda self: getattr(self.association,
'_backref_%s' % self.association.model))
def __init__(self, user, permission):
if isinstance(user, int):
self.user_id = user
else:
self.user = user
self.permission = permission
class MyMappedCollection(dict):
def __init__(self, data=None):
if data:
self.update(data)
@collection.appender
def __setitem__(self, permission, pobj):
l = self.setdefault(permission, [])
l.append(pobj)
@collection.remover
def _remove(self, perm):
self[perm.permission].remove(perm)
@collection.iterator
def __iter__(self):
for vals in self.itervalues():
for perm in vals:
yield perm
def __repr__(self):
return '%s(%r)' % (type(self).__name__, self)
class PermissionAssoc(Base):
__tablename__ = 'permission_assoc'
__model__ = u'PermissionAssoc'
id = Column(Integer,
Sequence('permission_assoc_seq_id', optional=True),
primary_key=True)
model = Column(Unicode(64), index=True)
permissions = orm.relationship(Permission,
backref='association',
lazy=False,
collection_class=MyMappedCollection)
def __init__(self, model):
self.model = model
class ULine(Base):
""" Line base class """
__tablename__ = 'uline'
__model__ = u'ULine'
id = Column(Integer,
Sequence('uline_seq_id', optional=True),
primary_key=True)
name = Column(Unicode(256))
assoc_id = Column(Integer, ForeignKey('permission_assoc.id'))
permission_rel = orm.relation(PermissionAssoc,
backref=orm.backref('_backref_%s' % __model__,
uselist=False))
@property
def permissions(self):
if self.permission_rel is None:
self.permission_rel = PermissionAssoc(self.__model__)
return self.permission_rel.permissions
def __init__(self, name=''):
self.name = name
def initTestingDB():
engine = create_engine('sqlite:///:memory:', echo=False)
DBSession = scoped_session(sessionmaker())
DBSession.configure(bind=engine)
Base.metadata.bind = engine
Base.metadata.create_all(engine)
return DBSession
if __name__ == '__main__':
session = initTestingDB()
user1 = User('john')
user2 = User('kratos')
line = ULine('new line')
line.permissions['read'] = Permission(user1, 'read')
line.permissions['read'] = Permission(user2, 'read')
session.add(line)
session.flush()