def __init__(self, *args, **kwargs):
args = [_literal_as_binds(c) for c in args]
+ default_array_type = kwargs.pop('_default_array_type', sqltypes.ARRAY)
if 'type_' not in kwargs:
type_from_args = _type_from_args(args)
if isinstance(type_from_args, sqltypes.ARRAY):
kwargs['type_'] = type_from_args
else:
- kwargs['type_'] = sqltypes.ARRAY(type_from_args)
+ kwargs['type_'] = default_array_type(type_from_args)
kwargs['_parsed_args'] = args
super(array_agg, self).__init__(*args, **kwargs)
from sqlalchemy import Sequence, Table, Column, Integer, update, String,\
func, MetaData, Enum, Index, and_, delete, select, cast, text, \
Text, null
+from sqlalchemy import types as sqltypes
from sqlalchemy.dialects.postgresql import ExcludeConstraint, array
from sqlalchemy import exc, schema
from sqlalchemy.dialects import postgresql
from sqlalchemy.sql import util as sql_util
from sqlalchemy.util import u, OrderedDict
from sqlalchemy.dialects.postgresql import aggregate_order_by, insert
+from sqlalchemy.dialects.postgresql import array_agg as pg_array_agg
+from sqlalchemy.dialects.postgresql import ARRAY as PG_ARRAY
class SequenceTest(fixtures.TestBase, AssertsCompiledSQL):
"AS string_agg_1 FROM table1"
)
+ def test_pg_array_agg_implicit_pg_array(self):
+
+ expr = pg_array_agg(column('data', Integer))
+ assert isinstance(expr.type, PG_ARRAY)
+ is_(expr.type.item_type._type_affinity, Integer)
+
+ def test_pg_array_agg_uses_base_array(self):
+
+ expr = pg_array_agg(column('data', sqltypes.ARRAY(Integer)))
+ assert isinstance(expr.type, sqltypes.ARRAY)
+ assert not isinstance(expr.type, PG_ARRAY)
+ is_(expr.type.item_type._type_affinity, Integer)
+
+ def test_pg_array_agg_uses_pg_array(self):
+
+ expr = pg_array_agg(column('data', PG_ARRAY(Integer)))
+ assert isinstance(expr.type, PG_ARRAY)
+ is_(expr.type.item_type._type_affinity, Integer)
+
+ def test_pg_array_agg_explicit_base_array(self):
+
+ expr = pg_array_agg(column(
+ 'data', sqltypes.ARRAY(Integer)), type_=sqltypes.ARRAY(Integer))
+ assert isinstance(expr.type, sqltypes.ARRAY)
+ assert not isinstance(expr.type, PG_ARRAY)
+ is_(expr.type.item_type._type_affinity, Integer)
+
+ def test_pg_array_agg_explicit_pg_array(self):
+
+ expr = pg_array_agg(column(
+ 'data', sqltypes.ARRAY(Integer)), type_=PG_ARRAY(Integer))
+ assert isinstance(expr.type, PG_ARRAY)
+ is_(expr.type.item_type._type_affinity, Integer)
+
def test_aggregate_order_by_adapt(self):
m = MetaData()
table = Table('table1', m, Column('a', Integer), Column('b', Integer))