On Jan 13, 2011, at 3:29 PM, Michael Bayer wrote:
> There's mistakes in how this is structured. UserDefinedType represents a
> type object applied to a Column. The actual data handled by such a type is
> not meant to be an instance of the type itself. ACLItem() here would be its
> own class, and UserDefinedType would be the superclass of a class like
> "ACLItemType". ACLItemType() is placed on the Column, and its
> bind_processor() and result_processor() deal with ACLItem objects.
>
> Once you have the roles of "type" and "value" set up, you'd want to mixin
> sqlalchemy.types.MutableType, which alerts the ORM that the value of this
> type can change "inline". Note that MutableType is not recommended for high
> volume applications as it performs terribly, due to the need for flush() to
> scan all "mutables" in the session for changes every time it's called - 0.7
> has a new extension that allows mutable values to send change events in an
> efficient manner.
Thank you for the detailed response!
I have implemented all of your advice, but have hit another roadblock-
specifically, I cannot determine how to properly cast the array type from
within the ACLItemArrayType class.
....../sqlalchemy/engine/default.py", line 299, in do_execute
cursor.execute(statement, parameters)
ProgrammingError: (ProgrammingError) column "aclarray" is of type aclitem[] but
expression is of type text[]
LINE 1: INSERT INTO imageaclarray (aclarray) VALUES (ARRAY[E'user te...
^
HINT: You will need to rewrite or cast the expression.
'INSERT INTO imageaclarray (aclarray) VALUES (%(aclarray)s) RETURNING
imageaclarray.id' {'aclarray': ['user testrole1=rU*/testrole1', 'user
testrole2=Xc/testrole1']}
What is really need is:
insert into imageaclarray (aclarray) values (ARRAY[E'user
agentm=rU*/agentm',E'user agentm=Xc/agentm']::aclitem[]);
or
insert into imageaclarray (aclarray) values (ARRAY[E'user
agentm=rU*/agentm'::aclitem,E'user agentm=Xc/agentm'::aclitem]);
but I can't figure out how to add casting or if it is even possible... clearly
bind_processor is too late, but psycopg2 doesn't seem to offer an option to set
the oid for the container array either. Is there a method during compilation I
can use?
Cheers,
M
Code and unit test follow:
=========================
import sqlalchemy.types as types
import re
import sqlalchemy.exc
from sqlalchemy.dialects.postgresql import ARRAY
#include/utils/acl.h
#define ACL_ALL_RIGHTS_STR "arwdDxtXUCTc"
#update for SQLAlchemy 0.7:
http://www.sqlalchemy.org/docs/07/orm/extensions/mutable.html
class ACLItem(object):
def __init__(self,grantee,permissions,grantor,grant_option=False):
self.grantee = grantee
self.permissions = []
if permissions:
for p in permissions:
self.permissions.append(p)
self.grantor = grantor
self.grant_option = grant_option
def _as_pgsql_string(self):
#convert to string 'user <grantee>=<perms>/<grantor>'
string_perms = ''
for perm in self.permissions:
string_perms += perm
if self.grant_option:
grant_option = '*'
else:
grant_option = ''
return "user %s=%s%s/%s" %
(self.grantee,string_perms,grant_option,self.grantor)
@classmethod
def _from_pgsql_string(klass,aclstring):
"grantee=perms*/grantor"
matches = re.match('([^=]+)=([^/\*]+)(\*?)/(\w+)',aclstring)
if matches is None:
raise sqlalchemy.exc.DataError(aclstring,[],'')
grantee = matches.group(1)
permissions = matches.group(2)
grant_option = bool(len(matches.group(3)))
grantor = matches.group(4)
return ACLItem(grantee,permissions,grantor,grant_option)
def __eq__(self,other_object):
if not isinstance(other_object,self.__class__):
return False
return str(self) == str(other_object)
def has_permission(self,permission_test):
return permission_test in self.permissions
def set_permission(self,permission,on=True):
if not self.has_permission(permission) and on:
self.permissions.append(permission)
elif self.has_permission(permission) and not on:
self.permissions.remove(permission)
def clear_permissions(self):
del self.permissions[:]
def reset_with_acl(self,acl):
"""takes an acl and replaces its own settings with the argument settings
This is useful for cases where an acl in an array is "replaced" without
being creating a new aclitem so that the array order in unmodified
"""
self.grantee = acl.grantee
self.permissions = acl.permissions
self.grantor = acl.grantor
self.grant_option = acl.grant_option
def __str__(self):
return self._as_pgsql_string()
class ACLItemArray(list):
#in an aclitem array, the
def aclitem_for_grantee(self,role):
for acl in self:
if role == acl.grantee:
return acl
return None
def has_permission(self,role,permission):
return self.aclitem_for_grantee(role).has_permission(permission)
def set_permission(self,role,permission):
return self.aclitem_for_grantee(role).set_permission(permission)
def clear_permissions(self):
del self.acls[:]
def append(self,acl):
#if there is already an acl for this user, replace with new acl (so
that there are not duplicate acls for the grantee)
existing_acl = self.aclitem_for_grantee(acl.grantee)
if existing_acl:
existing_acl.reset_with_acl(acl)
else:
super(ACLItemArray,self).append(acl)
def remove_grantee(self,grantee):
existing_acl = self.aclitem_for_grantee(grantee)
if existing_acl:
self.remove(existing_acl)
else:
raise ValueError('ACL for %s not in list' % grantee)
class ACLItemArrayType(ARRAY):
def __init__(self,acls=[]):
self.acls = acls
return super(ACLItemArrayType,self).__init__(ACLItemType)
def dialect_impl(self,dialect):
return self
#def adapt(self,impltype):
def bind_processor(self, dialect):
return super(ACLItemArrayType,self).bind_processor(dialect)
class ACLItemType(types.MutableType,types.UserDefinedType):
#class ACLItemType(types.UserDefinedType,types.MutableType): #FAIL
def get_col_spec(self):
return 'aclitem'
def bind_processor(self,dialect):
def acl2string(aclitem):
return aclitem._as_pgsql_string()
return acl2string
def copy_value(self,value):
return ACLItem(value.grantee,
value.permissions,
value.grantor,
value.grant_option)
def result_processor(self,dialect,column_type):
def string2acl(aclstring):
return ACLItem._from_pgsql_string(aclstring)
return string2acl
=================
from sqlalchemy.schema import Column
from sqlalchemy.types import Integer
from sqlalchemy.orm import scoped_session
from sqlalchemy.engine import create_engine
from sqlalchemy.orm.session import sessionmaker
from copy import deepcopy
#nosetests -s -x
test_aclcolumntype.py:TestACLItemColumnTypeController.test_aclitem
#nosetests -s -x
test_aclcolumntype.py:TestACLItemColumnTypeController.test_aclitemarray
Base = declarative_base()
class ImageACL(Base):
__tablename__ = 'imageacl'
id = Column(Integer,primary_key=True)
acl = Column(ACLItemType,nullable=False)
class ImageACLArray(Base):
__tablename__ = 'imageaclarray'
id = Column(Integer,primary_key=True)
aclarray = Column(ACLItemArrayType,nullable=False)
class TestACLItemColumnTypeController(TestController):
def setUp(self):
self.engine = create_engine('postgresql://localhost/test') #,echo=True)
self.session =
scoped_session(sessionmaker(autocommit=False,bind=self.engine))
Base.metadata.drop_all(self.engine)
Base.metadata.create_all(self.engine)
self.session.execute("CREATE ROLE testrole1;")
self.session.execute("CREATE ROLE testrole2;")
def tearDown(self):
self.session.rollback()
def test_aclitem(self):
session = self.session
#create an aclitem row
row = ImageACL()
row.acl = ACLItem('testrole1','rX','testrole1')
session.add(row)
session.flush()
session.refresh(row)
acl = row.acl
assert(acl.grantee == 'testrole1')
assert(acl.grantor == 'testrole1')
assert(acl.has_permission('r'))
assert(acl.has_permission('X'))
assert(not acl.has_permission('w'))
#set some new permission
acl.grantee = 'testrole2'
acl.grantor = 'testrole2'
acl.clear_permissions()
acl.set_permission('r',True)
acl.set_permission('r',False)
acl.set_permission('X')
acl.grant_option = True
assert(session.is_modified(row))
session.flush()
session.refresh(row)
acl = row.acl
assert(acl.grantee == 'testrole2')
assert(acl.grantor == 'testrole2')
assert(acl.has_permission('r') is False)
assert(acl.has_permission('X') is True)
assert(acl.grant_option is True)
def test_aclitemarray(self):
session = self.session
#create new aclitemarray
row = ImageACLArray()
row_clone = deepcopy(row)
row.aclarray = ACLItemArray()
acl1 = ACLItem('testrole1',['r','U'],'testrole1',True)
acl2 = ACLItem('testrole2',['X','c'],'testrole1',False)
row.aclarray.append(acl1)
row.aclarray.append(acl2)
#save to database
session.add(row)
session.flush()
session.refresh(row)
assert(row == row_clone)
--
You received this message because you are subscribed to the Google Groups
"sqlalchemy" group.
To post to this group, send email to sqlalchemy@googlegroups.com.
To unsubscribe from this group, send email to
sqlalchemy+unsubscr...@googlegroups.com.
For more options, visit this group at
http://groups.google.com/group/sqlalchemy?hl=en.
