Example creating some extra postgres types

extra_postgrers_types.py

from sqlalchemy.sql.expression import ColumnElement, _literal_as_text
from sqlalchemy.ext.compiler import compiles
from sqlalchemy.dialects.postgresql import ARRAY

class StaticArray(sa.types.TypeDecorator):
    impl = sa.types.TypeEngine

    def __init__(self):
        super(StaticArray, self).__init__()
        import sqlalchemy.dialects.postgresql.base as pg
        self.__supported = {pg.PGDialect:pg.PGArray}
        del pg

    def load_dialect_impl(self, dialect):
        if dialect.__class__ in self.__supported:
            return self.__supported[dialect.__class__](sa.String)
        else:
            return dialect.type_descriptor(sa.String)

    def process_bind_param(self, value, dialect):
        return value

    def process_result_value(self, value, dialect):
        if value is not None:
            return tuple(value)
        else:
            return None
            
    def is_mutable(self):
        return False

class array_agg(ColumnElement):
    type = StaticArray()
    
    def __init__(self, expr, order_by=None):
        self.expr = _literal_as_text(expr)

        if order_by is not None:
            self.order_by = _literal_as_text(order_by)
        else:
            self.order_by = None

@compiles(array_agg, 'postgresql')
def _compile_array_agg_postgresql(element, compiler, **kw):
    buf = StringIO()
    buf.write('array_agg(')
    buf.write(compiler.process(element.expr))
    if element.order_by is not None:
        buf.write(' ORDER BY ')
        buf.write(compiler.process(element.order_by))
    buf.write(')')
    return buf.getvalue()