On 09/30/2016 04:32 PM, Seth P wrote:
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)?
ha see you *are* doing the write side.
the bind_expression() hook is here to allow you to re-render the
expression. assuming value-bound bindparam() objects (e.g. not like
you'd get with an INSERT or UPDATE usually), the value should be present
and you can do this (had to work up a POC):
from sqlalchemy import *
from sqlalchemy.types import UserDefinedType
class T(UserDefinedType):
def bind_expression(self, colexpr):
return literal_column(colexpr.value) # or whatever is needed here
t = table('t', column('x', T()))
print t.select().where(t.c.x == 'hi')
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'))?
look into adding SchemaType as a mixin, it signals to the owning Column
that it should receive events. You can then add the events to your
type itself like before_parent_attach which should fire for the Column.
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
<mailto:sqlalchemy+unsubscr...@googlegroups.com>.
To post to this group, send email to sqlalchemy@googlegroups.com
<mailto:sqlalchemy@googlegroups.com>.
Visit this group at https://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 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.
