like this:
class LiteralBindParam(BindParameter):
pass
@compiles(LiteralBindParam)
def literal_bind(element, compiler, **kw):
kw['literal_binds'] = True
return compiler.visit_bindparam(element, **kw)
class ArrayType(UserDefinedType):
def get_col_spec(self):
return "ARRAY"
def literal_processor(self, dialect):
def process(value):
return 'int_array(' + ", ".join(str(v) for v in value) + ')'
return process
from sqlalchemy.orm import Session
engine = sqlalchemy.create_engine("sqlite://", echo=True)
metadata = MetaData(engine)
session = Session(engine)
a = LiteralBindParam(None, [1,2,3,4,5], type_=ArrayType)
ret = session.execute(func.some_db_func(a)).scalar()
On Feb 13, 2014, at 1:03 PM, Amos Smith <asm...@aseg.com> wrote:
> I found the missing space before literal_processor in my earlier listing,
> poor font selection hid that pretty well - still not sure why I couldn't see
> that earlier apologies again for that post.
>
> The new listing below now runs to completion but literal_processor method is
> not called for my UDT 'ArrayType', it's just being bound as the value NULL.
>
> It's still not clear to me if there is an issue with literal_processor being
> called, or I'm misusing the LiteralBindParam recipe from earlier post, or
> possibly still and issue with the array type udt.
>
> I'm not sure what type should be used in following line; currently just None.
>
> a = LiteralBindParam(None, ArrayType([1,2,3,4,5]))
>
> I'm also not sure if the return from ArrayType get_col_spec is valid; or if
> this needs to be registered or defined somewhere.
>
> def get_col_spec(self):
> return "ARRAY"
>
> ------------------------
>
> import os
>
> import sqlalchemy
> from sqlalchemy.orm import sessionmaker
> from sqlalchemy import MetaData
> from sqlalchemy.ext.declarative import declarative_base
>
> from sqlalchemy import func
> from sqlalchemy.types import UserDefinedType
>
> from sqlalchemy.sql.expression import BindParameter
> from sqlalchemy.ext.compiler import compiles
>
> #================================================
> class LiteralBindParam(BindParameter):
> pass
>
> @compiles(LiteralBindParam)
> def literal_bind(element, compiler, **kw):
> kw['literal_binds'] = True
> return compiler.visit_bindparam(element, **kw)
>
> #================================================
> class ArrayType(UserDefinedType):
> def __init__(self, listVal):
> self.listVal = listVal
> return
>
> def get_col_spec(self):
> return "ARRAY"
>
> def literal_processor(self, dialect):
> print 'got called'
> def process(value):
> return 'int_array(' + self.listValue + ')'
> return process
>
> #================================================
> dbUser = os.environ.get('uid')
> dbPwd = os.environ.get('pwd')
> oraSID = os.environ.get('sid')
> connstr = 'oracle://%s:%s@%s' % (dbUser, dbPwd, oraSID)
>
> #================================================
> engine = sqlalchemy.create_engine(connstr)
> metadata = MetaData(engine)
> Base = declarative_base(metadata=metadata)
>
> sessionMaker = sessionmaker(bind=engine)
> session = sessionMaker()
>
> a = LiteralBindParam(None, ArrayType([1,2,3,4,5]))
>
> ret = session.execute(func.some_db_func(a)).scalar()
>
> print ret
>
> ----- Original Message -----
> From: "Amos Smith" <asm...@aseg.com>
> To: sqlalchemy@googlegroups.com
> Sent: Wednesday, February 12, 2014 3:57:40 PM
> Subject: Re: [sqlalchemy] array types using OracleDialect
>
> I migrated to 0.9.2 as I understood the literal_processor was new feature.
>
> I always get this error from listing shown below:
>
> def literal_processor(self, dialect):
> ^
> IndentationError: unindent does not match any outer indentation level
>
>
> ArrayType is obviously just a stub in this listing.
>
> Apologies in advance if this is some obvious Python formatting issue of mine,
> I'm still learning this language.
>
>
>
>
> #===== Begin Listing =============================
>
> import os
>
> import sqlalchemy
> from sqlalchemy.orm import sessionmaker
> from sqlalchemy import MetaData
> from sqlalchemy.ext.declarative import declarative_base
>
> from sqlalchemy import func
> from sqlalchemy.types import UserDefinedType
>
> from sqlalchemy.sql.expression import BindParameter
> from sqlalchemy.ext.compiler import compiles
>
> #================================================
> class LiteralBindParam(BindParameter):
> pass
>
> @compiles(LiteralBindParam)
> def literal_bind(element, compiler, **kw):
> kw['literal_binds'] = True
> return compiler.visit_bindparam(element, **kw)
>
> #================================================
> class ArrayType(UserDefinedType):
> def get_col_spec(self):
> return "ARRAY"
>
> def column_expression(self, col):
> return None
>
> def literal_processor(self, dialect):
> def process(value):
> return "int_array(1, 2, 3, 4, 5)"
> return process
>
> #================================================
> dbUser = os.environ.get('uid')
> dbPwd = os.environ.get('pwd')
> oraSID = os.environ.get('sid')
> connstr = 'oracle://%s:%s@%s' % (dbUser, dbPwd, oraSID)
>
> #================================================
> engine = sqlalchemy.create_engine(connstr)
> metadata = MetaData(engine)
> Base = declarative_base(metadata=metadata)
> dialectMgr = DialectManager()
> sessionMaker = sessionmaker(bind=engine)
> session = sessionMaker()
>
> a = LiteralBindParam(None, ArrayType())
>
> session.execute(func.some_db_func(a)).scalar()
>
> #===== End Listing ===============================
>
> ----- Original Message -----
>
>
>> Regarding the following:
>>> if the type of the LiteralBindParameter implements “literal_processor()”,
>>> that controls how the literal value is rendered into the statement.
>>
>> How does one implement the "literal_processor()" for a new type? Is
>> literal_processor() method applicable for UserDefinedTypes?
>
> the method is literal_processor:
> http://docs.sqlalchemy.org/en/rel_0_9/core/types.html#sqlalchemy.types.TypeEngine.literal_processor
>
>
> its new as of 0.9 and applies to any type, including UserDefinedType. If you
> don’t see it taking effect, that’s a bug.
>
> --
> 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 http://groups.google.com/group/sqlalchemy.
> For more options, visit https://groups.google.com/groups/opt_out.
>
> --
> 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 http://groups.google.com/group/sqlalchemy.
> For more options, visit https://groups.google.com/groups/opt_out.
signature.asc
Description: Message signed with OpenPGP using GPGMail
