def test_python_fn_default(self):
        metadata = MetaData()
        table = Table(
            "sometable",
            metadata,
            Column("id", Integer, primary_key=True),
            Column("data", String),
            Column("foo", Integer, default=lambda: 10),
        )

        values = [
            {"id": 1, "data": "data1"},
            {"id": 2, "data": "data2", "foo": 15},
            {"id": 3, "data": "data3"},
        ]

        checkparams = {
            "id_m0": 1,
            "id_m1": 2,
            "id_m2": 3,
            "data_m0": "data1",
            "data_m1": "data2",
            "data_m2": "data3",
            "foo": None,  # evaluated later
            "foo_m1": 15,
            "foo_m2": None,  # evaluated later
        }

        stmt = table.insert().values(values)
        eq_(
            dict(
                [
                    (k, v.type._type_affinity)
                    for (k, v) in stmt.compile(
                        dialect=postgresql.dialect()
                    ).binds.items()
                ]
            ),
            {
                "foo": Integer,
                "data_m2": String,
                "id_m0": Integer,
                "id_m2": Integer,
                "foo_m1": Integer,
                "data_m1": String,
                "id_m1": Integer,
                "foo_m2": Integer,
                "data_m0": String,
            },
        )

        self.assert_compile(
            stmt,
            "INSERT INTO sometable (id, data, foo) VALUES "
            "(%(id_m0)s, %(data_m0)s, %(foo)s), "
            "(%(id_m1)s, %(data_m1)s, %(foo_m1)s), "
            "(%(id_m2)s, %(data_m2)s, %(foo_m2)s)",
            checkparams=checkparams,
            dialect=postgresql.dialect(),
        )

    def test_sql_expression_pk_noautoinc_returning(self):
        # test that return_defaults() works with a primary key where we are
        # sending a SQL expression, and we need to get the server-calculated
        # value back.  [ticket:3133]
        metadata = MetaData()
        table = Table(
            "sometable",
            metadata,
            Column("id", Integer, autoincrement=False, primary_key=True),
            Column("data", String),
        )

        stmt = table.insert().return_defaults().values(id=func.foobar())
        returning_dialect = postgresql.dialect()
        returning_dialect.implicit_returning = True
        compiled = stmt.compile(
            dialect=returning_dialect, column_keys=["data"]
        )
        eq_(compiled.postfetch, [])
        eq_(compiled.returning, [table.c.id])

        self.assert_compile(
            stmt,
            "INSERT INTO sometable (id, data) VALUES "
            "(foobar(), %(data)s) RETURNING sometable.id",
            checkparams={"data": "foo"},
            params={"data": "foo"},
            dialect=returning_dialect,
        )

def print_query(query_result):
    try:
        return str(query_result.statement.compile(dialect=postgresql.dialect(), compile_kwargs={"literal_binds": True}))
    except Exception as e:
        logging.getLogger(__name__).warning(str(e))
        return str(query_result)
        pass

    def test_modify_metadata(self):

        metadata = ConceptNameTransform.modify_metadata(self.metadata)

        indexes = metadata.tables['table1'].indexes
        self.assertEqual(len(indexes), 2, 'Indexes created')

        for index in indexes:
            index_sql = str(CreateIndex(index).compile(
                dialect=postgresql.dialect()))
            if index.name == 'tab_fcn_81387a6132eed7f69b5_ix':
                expected = clean("""
                  CREATE INDEX tab_fcn_81387a6132eed7f69b5_ix
                    ON table1 (foo_concept_name)
                """)
                self.assertEqual(index_sql, expected)
            elif index.name == 'tab_bcn_7baa5e16ad1f8129b90_ix':
                expected = clean("""
                  CREATE INDEX tab_bcn_7baa5e16ad1f8129b90_ix
                    ON table1 (bar_concept_name)
                """)
                self.assertEqual(index_sql, expected)
            else:
                self.fail(
                    'Unexpected index encountered: {}'.format(index.name))

    def test_modify_select(self):

        select_obj = select([self.table1])
        join_obj = self.table1

        select_obj, join_obj = ConceptNameTransform.modify_select(
            self.metadata,
            'user',
            select_obj,
            join_obj)

        select_obj = select_obj.select_from(join_obj)

        new_sql = str(select_obj.compile(dialect=postgresql.dialect()))

        expected = clean("""
          SELECT table1.foo_concept_id, table1.bar_concept_id,
          concept_1.concept_name AS foo_concept_name, concept_2.concept_name
          AS bar_concept_name
          {NL}FROM table1
          LEFT OUTER JOIN concept AS concept_1
              ON concept_1.concept_id = foo_concept_id
          LEFT OUTER JOIN concept AS concept_2
              ON concept_2.concept_id = bar_concept_id
        """)

        self.maxDiff = None
        self.assertEqual(expected, new_sql)

    def test_insert_returning(self):
        dialect = postgresql.dialect()
        table1 = table(
            "mytable", column("myid", Integer), column("name", String(128)), column("description", String(128))
        )

        i = insert(table1, values=dict(name="foo")).returning(table1.c.myid, table1.c.name)
        self.assert_compile(
            i,
            "INSERT INTO mytable (name) VALUES " "(%(name)s) RETURNING mytable.myid, " "mytable.name",
            dialect=dialect,
        )
        i = insert(table1, values=dict(name="foo")).returning(table1)
        self.assert_compile(
            i,
            "INSERT INTO mytable (name) VALUES "
            "(%(name)s) RETURNING mytable.myid, "
            "mytable.name, mytable.description",
            dialect=dialect,
        )
        i = insert(table1, values=dict(name="foo")).returning(func.length(table1.c.name))
        self.assert_compile(
            i,
            "INSERT INTO mytable (name) VALUES " "(%(name)s) RETURNING length(mytable.name) " "AS length_1",
            dialect=dialect,
        )

    def _mquery(self, query_obj=None, *filter, **filter_by):
        """ Get a MongoQuery with initial filtering applied

        :param query_obj: Query Object
        :type query_obj: dict|None
        :param filter: Additional filter() criteria
        :param filter_by: Additional filter_by() criteria
        :rtype: sqlalchemy.orm.Query, list of fields
        """
        mongo_query = self._getCrudHelper().mquery(
            self._query().filter(*filter).filter_by(**filter_by),
            query_obj
        )
        sqlalchemy_query = mongo_query.end()
        try:
            dialect = pg.dialect()
            sql_query = sqlalchemy_query.statement.compile(dialect=dialect)
            if PY2:
                sql_str = (sql_query.string.encode(dialect.encoding) % sql_query.params).decode(dialect.encoding)
            else:
                sql_str = sql_query.string % sql_query.params
            self.sqlaclhemy_queries.append(sql_str)
        except Exception as e:
            logging.error('Error generate SQL string %e', e)
        return sqlalchemy_query, mongo_query.get_project()

 def test_render_explain_analyze(self, session, Article):
     assert str(
         explain_analyze(session.query(Article))
         .compile(
             dialect=postgresql.dialect()
         )
     ).startswith('EXPLAIN (ANALYZE true) SELECT')

    def test_inline_default(self):
        metadata = MetaData()
        table = Table('sometable', metadata,
                      Column('id', Integer, primary_key=True),
                      Column('data', String),
                      Column('foo', Integer, default=func.foobar()))

        values = [
            {'id': 1, 'data': 'data1'},
            {'id': 2, 'data': 'data2', 'foo': 'plainfoo'},
            {'id': 3, 'data': 'data3'},
        ]

        checkparams = {
            'id_m0': 1,
            'id_m1': 2,
            'id_m2': 3,
            'data_m0': 'data1',
            'data_m1': 'data2',
            'data_m2': 'data3',
            'foo_m1': 'plainfoo',
        }

        self.assert_compile(
            table.insert().values(values),
            'INSERT INTO sometable (id, data, foo) VALUES '
            '(%(id_m0)s, %(data_m0)s, foobar()), '
            '(%(id_m1)s, %(data_m1)s, %(foo_m1)s), '
            '(%(id_m2)s, %(data_m2)s, foobar())',
            checkparams=checkparams,
            dialect=postgresql.dialect())

    def test_server_default(self):
        metadata = MetaData()
        table = Table(
            "sometable",
            metadata,
            Column("id", Integer, primary_key=True),
            Column("data", String),
            Column("foo", Integer, server_default=func.foobar()),
        )

        values = [
            {"id": 1, "data": "data1"},
            {"id": 2, "data": "data2", "foo": "plainfoo"},
            {"id": 3, "data": "data3"},
        ]

        checkparams = {
            "id_m0": 1,
            "id_m1": 2,
            "id_m2": 3,
            "data_m0": "data1",
            "data_m1": "data2",
            "data_m2": "data3",
        }

        self.assert_compile(
            table.insert().values(values),
            "INSERT INTO sometable (id, data) VALUES "
            "(%(id_m0)s, %(data_m0)s), "
            "(%(id_m1)s, %(data_m1)s), "
            "(%(id_m2)s, %(data_m2)s)",
            checkparams=checkparams,
            dialect=postgresql.dialect(),
        )

 def test_update_returning(self):
     dialect = postgresql.dialect()
     table1 = table(
         'mytable',
         column(
             'myid', Integer),
         column(
             'name', String(128)),
         column(
             'description', String(128)))
     u = update(
         table1,
         values=dict(
             name='foo')).returning(
         table1.c.myid,
         table1.c.name)
     self.assert_compile(u,
                         'UPDATE mytable SET name=%(name)s '
                         'RETURNING mytable.myid, mytable.name',
                         dialect=dialect)
     u = update(table1, values=dict(name='foo')).returning(table1)
     self.assert_compile(u,
                         'UPDATE mytable SET name=%(name)s '
                         'RETURNING mytable.myid, mytable.name, '
                         'mytable.description', dialect=dialect)
     u = update(table1, values=dict(name='foo'
                                    )).returning(func.length(table1.c.name))
     self.assert_compile(
         u,
         'UPDATE mytable SET name=%(name)s '
         'RETURNING length(mytable.name) AS length_1',
         dialect=dialect)

    def test_server_default_absent_value(self):
        metadata = MetaData()
        table = Table('sometable', metadata,
            Column('id', Integer, primary_key=True),
            Column('data', String),
            Column('foo', Integer, server_default=func.foobar()))

        values = [
            {'id': 1, 'data': 'data1', 'foo': 'plainfoo'},
            {'id': 2, 'data': 'data2'},
            {'id': 3, 'data': 'data3', 'foo': 'otherfoo'},
        ]

        checkparams = {
            'id_0': 1,
            'id_1': 2,
            'id_2': 3,
            'data_0': 'data1',
            'data_1': 'data2',
            'data_2': 'data3',
            'foo_0': 'plainfoo',
            'foo_2': 'otherfoo',
        }

        # note the effect here is that the first set of params
        # takes effect for the rest of them, when one is absent
        self.assert_compile(table.insert().values(values),
            'INSERT INTO sometable (id, data, foo) VALUES '
            '(%(id_0)s, %(data_0)s, %(foo_0)s), '
            '(%(id_1)s, %(data_1)s, %(foo_0)s), '
            '(%(id_2)s, %(data_2)s, %(foo_2)s)',
            checkparams=checkparams, dialect=postgresql.dialect())

    def test_insert_returning(self):
        dialect = postgresql.dialect()
        table1 = table('mytable',
                       column('myid', Integer),
                       column('name', String(128)),
                       column('description', String(128)),
                       )

        i = insert(
            table1,
            values=dict(
                name='foo')).returning(
            table1.c.myid,
            table1.c.name)
        self.assert_compile(i,
                            'INSERT INTO mytable (name) VALUES '
                            '(%(name)s) RETURNING mytable.myid, '
                            'mytable.name', dialect=dialect)
        i = insert(table1, values=dict(name='foo')).returning(table1)
        self.assert_compile(i,
                            'INSERT INTO mytable (name) VALUES '
                            '(%(name)s) RETURNING mytable.myid, '
                            'mytable.name, mytable.description',
                            dialect=dialect)
        i = insert(table1, values=dict(name='foo'
                                       )).returning(func.length(table1.c.name))
        self.assert_compile(i,
                            'INSERT INTO mytable (name) VALUES '
                            '(%(name)s) RETURNING length(mytable.name) '
                            'AS length_1', dialect=dialect)

    def test_postgres(self):
        from sqlalchemy.dialects import postgresql as diale_
        dialect = diale_.dialect()
        query = str(self.group_concat_md.compile(dialect=dialect))

        expected = (
            '''SELECT metric_dimension.dimension_set_id, STRING_AGG(metric_dimension.name '''
            '''|| '=' || metric_dimension.value, ',' ) AS dimensions '''
            '''
FROM metric_dimension GROUP BY metric_dimension.dimension_set_id''')
        self.assertEqual(expected, query)

        query = str(self.group_concat_md_order.compile(dialect=dialect))

        expected = (
            '''SELECT metric_dimension.dimension_set_id, STRING_AGG(metric_dimension.name '''
            '''|| '=' || metric_dimension.value, ',' ORDER BY metric_dimension.name ASC) '''
            '''AS dimensions '''
            '''
FROM metric_dimension GROUP BY metric_dimension.dimension_set_id''')
        self.assertEqual(expected, query)

        expected = (
            """SELECT metric_dimension.dimension_set_id \n"""
            """FROM metric_dimension ORDER BY CASE WHEN metric_dimension.dimension_set_id='A'"""
            """ THEN 0 WHEN metric_dimension.dimension_set_id='B' THEN 1 WHEN"""
            """ metric_dimension.dimension_set_id='C' THEN 2 ELSE 3 END ASC""")
        query = str(self.order_by_field.compile(dialect=dialect))
        self.assertEqual(expected, query)

 def _raise_query(self, q):
     """
         useful for debugging. just do...
         self._raise_query(q)
     """
     c = q.compile(dialect=postgresql.dialect())
     raise ValueError(c)

    def test_python_fn_default(self):
        metadata = MetaData()
        table = Table('sometable', metadata,
                      Column('id', Integer, primary_key=True),
                      Column('data', String),
                      Column('foo', Integer, default=lambda: 10))

        values = [
            {'id': 1, 'data': 'data1'},
            {'id': 2, 'data': 'data2', 'foo': 15},
            {'id': 3, 'data': 'data3'},
        ]

        checkparams = {
            'id_0': 1,
            'id_1': 2,
            'id_2': 3,
            'data_0': 'data1',
            'data_1': 'data2',
            'data_2': 'data3',
            'foo': None,  # evaluated later
            'foo_1': 15,
            'foo_2': None,  # evaluated later
        }

        self.assert_compile(
            table.insert().values(values),
            "INSERT INTO sometable (id, data, foo) VALUES "
            "(%(id_0)s, %(data_0)s, %(foo)s), "
            "(%(id_1)s, %(data_1)s, %(foo_1)s), "
            "(%(id_2)s, %(data_2)s, %(foo_2)s)",
            checkparams=checkparams,
            dialect=postgresql.dialect())

    def test_mysql(self):
        from sqlalchemy.dialects import mysql as diale_
        dialect = diale_.dialect()
        query = str(self.group_concat_md.compile(dialect=dialect))

        expected = (
            '''SELECT metric_dimension.dimension_set_id, '''
            '''GROUP_CONCAT(concat(concat(metric_dimension.name, '''
            ''''='), metric_dimension.value)  SEPARATOR ',') AS dimensions '''
            '''
FROM metric_dimension GROUP BY metric_dimension.dimension_set_id''')
        self.assertEqual(expected, query)

        query = str(self.group_concat_md_order.compile(dialect=dialect))

        expected = (
            '''SELECT metric_dimension.dimension_set_id, '''
            '''GROUP_CONCAT(concat(concat(metric_dimension.name, '''
            ''''='), metric_dimension.value) ORDER BY metric_dimension.name ASC '''
            '''SEPARATOR ',') AS dimensions '''
            '''
FROM metric_dimension GROUP BY metric_dimension.dimension_set_id''')
        self.assertEqual(expected, query)

        expected = (
            '''SELECT metric_dimension.dimension_set_id \n'''
            '''FROM metric_dimension '''
            '''ORDER BY FIELD(metric_dimension.dimension_set_id, 'A', 'B', 'C') ASC''')

        query = str(self.order_by_field.compile(dialect=dialect))
        self.assertEqual(expected, query)

def copy_to(source, dest, engine_or_conn, **flags):
    """Export a query or select to a file. For flags, see the PostgreSQL
    documentation at http://www.postgresql.org/docs/9.5/static/sql-copy.html.

    Examples: ::
        select = MyTable.select()
        with open('/path/to/file.tsv', 'w') as fp:
            copy_to(select, fp, conn)

        query = session.query(MyModel)
        with open('/path/to/file/csv', 'w') as fp:
            copy_to(query, fp, engine, format='csv', null='.')

    :param source: SQLAlchemy query or select
    :param dest: Destination file pointer, in write mode
    :param engine_or_conn: SQLAlchemy engine, connection, or raw_connection
    :param **flags: Options passed through to COPY

    If an existing connection is passed to `engine_or_conn`, it is the caller's
    responsibility to commit and close.
    """
    dialect = postgresql.dialect()
    statement = getattr(source, 'statement', source)
    compiled = statement.compile(dialect=dialect)
    conn, autoclose = raw_connection_from(engine_or_conn)
    cursor = conn.cursor()
    query = cursor.mogrify(compiled.string, compiled.params).decode()
    formatted_flags = '({})'.format(format_flags(flags)) if flags else ''
    copy = 'COPY ({}) TO STDOUT {}'.format(query, formatted_flags)
    cursor.copy_expert(copy, dest)
    if autoclose:
        conn.close()

    def test_modify_metadata(self):

        metadata = AgeTransform.modify_metadata(self.metadata)

        indexes = metadata.tables['table1'].indexes
        self.assertEqual(len(indexes), 2, 'Indexes created')

        for index in indexes:
            index_sql = str(CreateIndex(index).compile(
                dialect=postgresql.dialect()))
            if index.name == 'tab_fsaim_107eee9e009461416_ix':
                expected = clean("""
                  CREATE INDEX tab_fsaim_107eee9e009461416_ix
                    ON table1 (foo_start_age_in_months)
                """)
                self.assertEqual(index_sql, expected)
            elif index.name == 'tab_bsaim_ca07fdbcdf9bfef7a_ix':
                expected = clean("""
                  CREATE INDEX tab_bsaim_ca07fdbcdf9bfef7a_ix
                    ON table1 (bar_start_age_in_months)
                """)
                self.assertEqual(index_sql, expected)
            else:
                self.fail(
                    'Unexpected index encountered: {}'.format(index.name))

    def test_modify_select(self):

        select_obj = select([self.table1])
        join_obj = self.table1

        select_obj, join_obj = AgeTransform.modify_select(
            self.metadata,
            'table1',
            select_obj,
            join_obj)

        select_obj = select_obj.select_from(join_obj)

        new_sql = str(select_obj.compile(dialect=postgresql.dialect()))

        expected = clean("""
          SELECT table1.foo_start_time,
          table1.bar_start_time,
          table1.person_id,
          months_in_interval(person.time_of_birth, table1.foo_start_time)
            AS foo_start_age_in_months,
          months_in_interval(person.time_of_birth, table1.bar_start_time)
            AS bar_start_age_in_months
          {NL}FROM table1
            JOIN person ON person.person_id = table1.person_id
          """)

        self.maxDiff = None
        self.assertEqual(new_sql, expected)