On Thursday, September 29, 2016 at 9:45:24 AM UTC-4, Mike Bayer wrote: > > you can add your own types to do these things also, especially > read-only, just make any subclass of UserDefinedType and apply whatever > result-row handling is needed for how cx_Oracle is returning the data. > > The hard part about types is the elaborate expression support (e.g. like > JSON foo ->> bar vs. foo -> bar in PG for example). Reading and > writing a value is not that hard and especially if the type is just > specific to what you need right now, you don't have the burden of making > sure your type works for all versions / flags / settings of Oracle / > cx_Oracle etc. >
The following seems to work fine for my purposes. (It uses alchy, https://github.com/dgilland/alchy). Is there a way to specify that any column of type VARRAY should always be bound literally, so that one doesn't need to specify explicitly .compile(compile_kwargs={"literal_binds": True}) (and thereby literally binding all fields)? Also, is there a way, inside VARRAY.__init__() or some other place that is called before table creation to specify the sa.event.listen(<table>, "before_create", self.create_ddl().execute_if(dialect='oracle'))? import six import sqlalchemy as sa class VARRAY(sa.types.UserDefinedType): def __init__(self, type_name, size_limit, item_type, nullable=True, as_tuple=False): super(VARRAY, self).__init__() self.type_name = type_name self.size_limit = size_limit self.item_type = item_type self.nullable = nullable self.as_tuple = as_tuple def compile(self, dialect=None): return self.type_name def get_col_spec(self, **kw): return self.type_name def create_ddl(self, dialect=None, or_replace=True): sql = "CREATE " if or_replace: sql += "OR REPLACE " sql += "TYPE %(schema)s.{} AS VARRAY({}) OF {}".format(self.type_name, self.size_limit, self.item_type.compile(dialect=dialect)) if not self.nullable: sql += " NOT NULL" return sa.DDL(sql) def process_literal_param(self, value, dialect): return "{}({})".format(self.type_name, ','.join("NULL" if x is None else ("'%s'" % x) if isinstance(x, six.string_types) else str(x) for x in value)) def literal_processor(self, dialect): def processor(value): return self.process_literal_param(value, dialect) return processor def process_result_value(self, value, dialect): if self.as_tuple: value = tuple(value) return value def result_processor(self, dialect, coltype): def processor(value): return self.process_result_value(value, dialect) return processor def copy(self): return VARRAY(self.type_name, self.size_limit, self.item_type, nullable=self.nullable, as_tuple=self.as_tuple) if __name__ == '__main__': uri = "oracle://user:password@host" import alchy import sqlalchemy.dialects.oracle as oc db = alchy.Manager(config={'SQLALCHEMY_DATABASE_URI': uri}) class TestVarray(db.Model): __tablename__ = 'test_varray' __table_args__ = { 'schema': 'myschema' } idx = sa.Column(sa.Integer, primary_key=True) label = sa.Column(sa.String(20), nullable=False) words = sa.Column(VARRAY("tp_test_varray_words", 3000, sa.String(8), nullable=True), nullable=False) numbers = sa.Column(VARRAY("tp_test_varray_numbers", 3000, oc.NUMBER(), nullable=True), nullable=False) sa.event.listen(TestVarray.__table__, "before_create", TestVarray.column_attrs()['words'].columns[0].type.create_ddl().execute_if(dialect='oracle')) sa.event.listen(TestVarray.__table__, "before_create", TestVarray.column_attrs()['numbers'].columns[0].type.create_ddl().execute_if(dialect='oracle')) db.drop_all() db.create_all() db.engine.execute(TestVarray.__table__.insert({'idx': 1, 'label': 'One', 'words': ['Once', 'upon', 'a', 'time'], 'numbers': [1.1, 1.2]}). compile(compile_kwargs={"literal_binds": True})) db.engine.execute(TestVarray.__table__.insert({'idx': 2, 'label': 'Two', 'words': ['To', 'be', 'or', 'not'], 'numbers': [2.1, 2.2]}). compile(compile_kwargs={"literal_binds": True})) print TestVarray.query.all() print db.session().query(TestVarray.label, TestVarray.words, TestVarray.numbers).all() [<TestVarray(idx=1, label='One', words=['Once', 'upon', 'a', 'time'], numbers=[1.1, 1.2])>, <TestVarray(idx=2, label='Two', words=['To', 'be', 'or', 'not'], numbers=[2.1, 2.2])>] [('One', ['Once', 'upon', 'a', 'time'], [1.1, 1.2]), ('Two', ['To', 'be', 'or', 'not'], [2.1, 2.2])] -- 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 sqlalchemy+unsubscr...@googlegroups.com. To post to this group, send email to sqlalchemy@googlegroups.com. Visit this group at https://groups.google.com/group/sqlalchemy. For more options, visit https://groups.google.com/d/optout.
