On 7/8/15 9:25 PM, Vincent Catalano wrote:
I've been trying to get SQLAlchemy to work with Postgres composite
types for the last couple days with no luck. I've implemented the
PGCompositeType that Michael created in another post
(https://groups.google.com/forum/#!searchin/sqlalchemy/postgres$20composite/sqlalchemy/f9BPVHfdvbg/HqEbDUO9UOMJ),
however, I cannot seem to get this to work. I'm also a bit unclear on
how to actually query based on the composite type (I'm using
declarative table definitions).
Well that email only went as far as writing SQL expressions.
Here is more detail added which starts to be able to query these types.
We must use psycopg2s facilities for doing so, described at
http://initd.org/psycopg/docs/extras.html#composite-types-casting.
This is only a rudimentary example and can use a lot of improvement.
For joining, you can use it in join of the form query.join(YourClass,
YourClass.composite_attr.attrs.whatever == OtherClass.whatever).
from sqlalchemy.ext.compiler import compiles
from sqlalchemy.types import UserDefinedType, to_instance
from sqlalchemy.sql.expression import ColumnElement, Tuple, bindparam
class PGCompositeElement(ColumnElement):
def __init__(self, base, attrname, type_):
self.base = base
self.attrname = attrname
self.type = to_instance(type_)
@compiles(PGCompositeElement)
def _compile_pgelem(element, compiler, **kw):
return "%s.%s" % (
compiler.process(element.base, **kw),
element.attrname
)
class ROW(Tuple):
pass
@compiles(ROW)
def _compile_row(element, compiler, **kw):
return "ROW(%s)" % (compiler.process(super(ROW, element), **kw))
class _Namespace(object):
def __init__(self, comparator):
self.comparator = comparator
def __getattr__(self, key):
try:
type_ = self.comparator.type.typemap[key]
except KeyError:
raise KeyError(
"Type '%s' doesn't have an attribute: '%s'" %
(self.comparator.type, key))
return PGCompositeElement(
self.comparator.expr,
key,
type_)
class PGCompositeType(UserDefinedType):
def __init__(self, name, typemap):
self.name = name
self.attrnames = [elem[0] for elem in typemap]
self.typemap = dict(typemap)
class comparator_factory(UserDefinedType.Comparator):
@property
def attrs(self):
return _Namespace(self)
def get_col_spec(self):
return self.name
if __name__ == '__main__':
from sqlalchemy import String, Integer, create_engine, Column
from sqlalchemy.orm import Session
from sqlalchemy import event
mytype = PGCompositeType(
"mytype", [("attr1", String(50)), ("attr2", Integer)])
from sqlalchemy.ext.declarative import declarative_base
Base = declarative_base()
class SomeClass(Base):
__tablename__ = 'sometable'
id = Column(Integer, primary_key=True)
composite_data = Column(mytype)
e = create_engine("postgresql://scott:tiger@localhost/test", echo=True)
conn = e.connect()
t = conn.begin()
conn.execute("create type mytype as (attr1 varchar(50), attr2
integer);")
import psycopg2
@event.listens_for(conn, 'before_cursor_execute')
def before_cursor_execute(
conn, cursor, statement, parameters, context, executemany):
psycopg2.extras.register_composite('mytype', cursor)
Base.metadata.create_all(conn)
s = Session(conn)
obj = SomeClass(composite_data=ROW("foo", 50))
s.add(obj)
s.commit()
o1 = s.query(SomeClass).first()
assert o1.composite_data == ('foo', 50)
Here is what I have so far - the column definition is taken from
Michael Bayer's post:
from sqlalchemy.ext.compiler import compiles
from sqlalchemy.types import UserDefinedType, to_instance
from sqlalchemy.sql.expression import ColumnElement
class PGCompositeElement(ColumnElement):
def __init__(self, base, field, type_):
self.name = field
super(PGCompositeElement, self).__init__(base)
self.type = to_instance(type_)
@compiles(PGCompositeElement)
def _compile_pgelem(expr, compiler, **kw):
return '(%s).%s' % (compiler.process(expr.clauses, **kw), expr.name)
class _Namespace(object):
def __init__(self, comparator):
self.comparator = comparator
def __getattr__(self, key):
try:
type_ = self.comparator.type.typemap[key]
except KeyError:
raise KeyError(
"Type '%s' doesn't have an attribute: '%s'" %
(self.comparator.type, key))
return PGCompositeElement(self.comparator.expr, key, type_)
class PGCompositeType(UserDefinedType):
def __init__(self, name, typemap):
self.name = name
self.typemap = typemap
class comparator_factory(UserDefinedType.Comparator):
@property
def attrs(self):
return _Namespace(self)
def get_col_spec(self):
return self.name
# My column and tables definitions:
TopoGeom = PGCompositeType('topo_geom', {'topology_id': Integer,
'layer_id': Integer, 'id': Integer, 'type': Integer})
class Trail(Base):
__tablename__ = 'trail'
id = Column(Integer, primary_key=True)
topo_geom = Column(TopoGeom)
When I query using the following, I get an attribute error since
/topo_geom/ is a string.
result = self.session.query(Trail).first()
print result.topo_geom.attrs.topology_id
How do I actually get the column attributes from a result (for
example, /topology_id/)? Additionally, is it possible to query and
join other tables based on the composite fields?
--
You received this message because you are subscribed to the Google
Groups "sqlalchemy" group.
To unsubscribe from this group and stop receiving emails from it, send
an email to [email protected]
<mailto:[email protected]>.
To post to this group, send email to [email protected]
<mailto:[email protected]>.
Visit this group at http://groups.google.com/group/sqlalchemy.
For more options, visit https://groups.google.com/d/optout.
--
You received this message because you are subscribed to the Google Groups
"sqlalchemy" group.
To unsubscribe from this group and stop receiving emails from it, send an email
to [email protected].
To post to this group, send email to [email protected].
Visit this group at http://groups.google.com/group/sqlalchemy.
For more options, visit https://groups.google.com/d/optout.
