def test_acs_columns_run():
    task = Columns()
    assert_equals(False, task.complete())
    assert_equals(0, len(current_session().dirty))
    worker = runtask(task)
    assert_equals(True, task.complete())
    assert_equals(True, worker.run_succeeded)
    assert_equals(0, len(current_session().dirty))

def test_table_task_increment_version_runs_again():
    task = TestTableTask()
    runtask(task)
    output = task.output()
    assert_true(output.exists())

    task = TestTableTask()
    task.version = lambda: 10
    output = task.output()
    assert_false(output.exists())

    current_session().rollback()
    runtask(task)
    assert_true(output.exists())

    def populate(self):
        session = current_session()

        insert = True
        for hometype, _, measure, _, _ in hometype_measures():
            col_id = hometype + '_' + measure
            input_table = self.input()[col_id].table
            if insert:
                stmt = 'INSERT INTO {output} (region_name, {col_id}) ' \
                        'SELECT "RegionName", "{year}-{month}"::NUMERIC ' \
                        'FROM {input_table} '
            else:
                stmt = 'UPDATE {output} ' \
                        'SET {col_id} = "{year}-{month}"::NUMERIC ' \
                        'FROM {input_table} WHERE ' \
                        '{input_table}."RegionName" = {output}.region_name '
            session.execute(stmt.format(
                output=self.output().table,
                year=str(self.year).zfill(2),
                month=str(self.month).zfill(2),
                col_id=col_id,
                input_table=input_table))
            if insert:
                session.execute('ALTER TABLE {output} ADD PRIMARY KEY (region_name)'.format(
                    output=self.output().table))
            insert = False

 def populate(self):
     session = current_session()
     session.execute('INSERT INTO {output} (the_geom, cvegeo) '
                     'SELECT wkb_geometry, cvegeo '
                     'FROM {input} '.format(
                         output=self.output().table,
                         input=self.input()['data'].table))

 def run(self):
     session = current_session()
     try:
         session.execute('DROP TABLE IF EXISTS observatory.obs_meta')
         session.execute(self.FIRST_AGGREGATE)
         session.execute('CREATE TABLE observatory.obs_meta AS {select}'.format(
             select=self.QUERY.replace('the_geom_webmercator', 'the_geom')
         ))
         # confirm that there won't be ambiguity with selection of geom
         session.execute('CREATE UNIQUE INDEX ON observatory.obs_meta '
                         '(numer_id, denom_id, numer_timespan, geom_weight)')
         session.execute('CREATE INDEX ON observatory.obs_meta USING gist '
                         '(the_geom)')
         for dimension, query in self.DIMENSIONS.iteritems():
             session.execute('DROP TABLE IF EXISTS observatory.obs_meta_{dimension}'.format(
                 dimension=dimension))
             session.execute('CREATE TABLE observatory.obs_meta_{dimension} '
                             'AS {select}'.format(
                                 dimension=dimension,
                                 select=query.replace('the_geom_webmercator', 'the_geom') \
                                             .replace('3857', '4326')
                             ))
             session.execute('CREATE INDEX ON observatory.obs_meta_{dimension} USING gist '
                             '(the_geom)'.format(dimension=dimension))
         session.commit()
         self._complete = True
     except:
         session.rollback()
         raise

    def populate(self):
        session = current_session()

        session.execute('INSERT INTO {output} '
                        'SELECT "wof:id", '
                        'CASE WHEN ST_Npoints(wkb_geometry) > 1000000 '
                        '     THEN ST_Simplify(wkb_geometry, 0.0001) '
                        '     ELSE wkb_geometry '
                        'END, '
                        '"wof:name" '
                        'FROM {input} '.format(
                            output=self.output().table,
                            input=self.input()['data'].table
                        ))

        # replace default WOF US states with our clipped versions
        if self.resolution == 'region':
            for geoid, statename in session.execute('SELECT geoid, name FROM tiger2014.state'):
                session.execute('UPDATE {output} out '
                                'SET the_geom = shoreline.the_geom '
                                'FROM {shoreline} shoreline '
                                'WHERE shoreline.geoid = \'{geoid}\' '
                                '  AND out.name ILIKE \'{statename}\' '.format(
                                    shoreline=self.input()['shoreline'].table,
                                    output=self.output().table,
                                    geoid=geoid,
                                    statename=statename))

    def populate(self):
        session = current_session()
        columns = self.columns()
        out_colnames = columns.keys()
        in_table = self.input()['data']
        in_colnames = [ct._id.split('.')[-1] for ct in columns.values()]
        in_colnames[0] = 'geo_code'
        for i, in_c in enumerate(in_colnames):
            cmd =   "SELECT 'exists' FROM information_schema.columns " \
                    "WHERE table_schema = '{schema}' " \
                    "  AND table_name = '{tablename}' " \
                    "  AND column_name = '{colname}' " \
                    "  LIMIT 1".format(
                        schema=in_table.schema,
                        tablename=in_table.tablename.lower(),
                        colname=in_c.lower())
            # remove columns that aren't in input table
            if session.execute(cmd).fetchone() is None:
                in_colnames[i] = None
                out_colnames[i] = None
        in_colnames = [
            "CASE {ic}::TEXT WHEN '-6' THEN NULL ELSE {ic} END".format(ic=ic) for ic in in_colnames if ic is not None]
        out_colnames = [oc for oc in out_colnames if oc is not None]

        cmd =   'INSERT INTO {output} ({out_colnames}) ' \
                'SELECT {in_colnames} FROM {input} '.format(
                    output=self.output().table,
                    input=in_table.table,
                    in_colnames=', '.join(in_colnames),
                    out_colnames=', '.join(out_colnames))
        session.execute(cmd)

    def populate(self):
        session = current_session()

        insert = True
        input_ = self.input()
        output = self.output()
        session.execute('ALTER TABLE {output} ADD PRIMARY KEY (region_name)'.format(
            output=output.table))
        session.flush()
        for hometype, _, measure, _, _ in hometype_measures():
            col_id = hometype + '_' + measure
            input_table = input_[col_id].table
            stmt = '''INSERT INTO {output} (region_name, {col_id})
                      SELECT "RegionName", "{year}-{month}"::NUMERIC
                      FROM {input_table}
                      ON CONFLICT (region_name)
                         DO UPDATE SET {col_id} = EXCLUDED.{col_id}'''
            session.execute(stmt.format(
                output=output.table,
                year=str(self.year).zfill(2),
                month=str(self.month).zfill(2),
                col_id=col_id,
                input_table=input_table))
            if insert:
                insert = False

 def populate(self):
     # This select statement transforms the input table, taking advantage of our
     # new column names.
     # The session is automatically committed if there are no errors.
     session = current_session()
     columns = self.columns()
     colnames = columns.keys()
     select_colnames = []
     for naics_code, naics_columns in self.input()['naics'].iteritems():
         for colname in naics_columns.keys():
             select_colnames.append('''MAX(CASE
                 WHEN industry_code = '{naics_code}' THEN {colname} ELSE NULL
             END)::Numeric'''.format(naics_code=naics_code,
                                     colname=colname
                                    ))
     insert = '''INSERT INTO {output} ({colnames})
                 SELECT area_fips, {select_colnames}
                 FROM {input}
                 GROUP BY area_fips '''.format(
                     output=self.output().table,
                     input=self.input()['data'].table,
                     colnames=', '.join(colnames),
                     select_colnames=', '.join(select_colnames),
                 )
     session.execute(insert)

 def populate(self):
     session = current_session()
     session.execute(' INSERT INTO {output} '
                     ' SELECT the_geom, id_2, pop2010, name_2 '
                     ' FROM {input} '.format(
                         output=self.output().table,
                         input=self.input()['data'].table
                     ))

def session_scope():
    """Provide a transactional scope around a series of operations."""
    try:
        yield current_session()
        session_commit(None)
    except Exception as e:
        session_rollback(None, e)
        raise

 def populate(self):
     session = current_session()
     session.execute('INSERT INTO {output} '
                     'SELECT ST_MakeValid(wkb_geometry), oa_sa, sprgrp, grp, subgrp '
                     'FROM {input}'.format(
                         output=self.output().table,
                         input=self.input()['data'].table,
                     ))

 def run(self):
     session = current_session()
     session.execute('CREATE TABLE {output} AS '
                     'SELECT geoid, ST_Union(ST_MakeValid(the_geom)) AS the_geom, '
                     '       MAX(aland) aland, MAX(awater) awater '
                     'FROM {input} '
                     'GROUP BY geoid'.format(
                         output=self.output().table,
                         input=self.input().table))

 def populate(self):
     session = current_session()
     session.execute('INSERT INTO {output} '
                     'SELECT {code}uid as geom_id, '
                     '       wkb_geometry as geom '
                     'FROM {input} '.format(
                         output=self.output().table,
                         code=self.resolution.replace('_', ''),
                         input=self.input()['data'].table))

    def requires(self):
        tables = {}
        session = current_session()
        for table in session.query(OBSTable):
            if should_upload(table):
                tables[table.id] = table.tablename

        for table_id, tablename in tables.iteritems():
            yield TableToCartoViaImportAPI(table=tablename, force=self.force)

def test_column_task(klass):
    # Ensure every column task runs and produces some kind of independent
    # metadata.
    # TODO: test columnstasks that have params
    if klass.get_param_names():
        raise SkipTest("Cannot test ColumnsTask with params")
    task = klass()
    runtask(task)
    assert_greater(current_session().query(OBSColumn).filter(OBSColumn.id.startswith(classpath(task))).count(), 0)

 def requires(self):
     session = current_session()
     if self.exact_id:
         table = session.query(OBSTable).get(self.exact_id)
     elif self.tablename:
         table = session.query(OBSTable).filter(OBSTable.tablename == self.tablename).one()
     elif self.id:
         table = session.query(OBSTable).filter(OBSTable.id.ilike('%' + self.id + '%')).one()
     else:
         raise Exception('Need id or exact_id for SyncData')
     return TableToCarto(table=table.tablename, force=self.force)

    def populate(self):
        table_name = self.output().table
        shell(r"psql -c '\copy {table} FROM {file_path} WITH CSV HEADER'".format(
            table=table_name,
            file_path=self.input()['data_file'].path
        ))
        for name, segment_id in SpielmanSingletonColumns.x10_mapping.iteritems():
            current_session().execute("update {table} set X10 = '{name}' "
                                      "where X10 ='{segment_id}'; ".format(
                                          table=table_name,
                                          name=name,
                                          segment_id=segment_id
                                      ))

        for name, segment_id in SpielmanSingletonColumns.x55_mapping.iteritems():
            current_session().execute("update {table} set X55 = '{name}' "
                                      "where X55 ='{segment_id}'; ".format(
                                          table=table_name,
                                          name=name,
                                          segment_id=segment_id
                                      ))

 def run(self):
     session = current_session()
     session.execute("CREATE TABLE {output} AS "
                     "SELECT * FROM {input} "
                     "WHERE agglvl_code IN ('74', '73', '71') "
                     "  AND year = '{year}' "
                     "  AND qtr = '{qtr}' "
                     "  AND own_code = '5' ".format(
                         input=self.input().table,
                         output=self.output().table,
                         year=self.year,
                         qtr=self.qtr,
                     ))

 def populate(self):
     session = current_session()
     stmt = ('INSERT INTO {output} '
             'SELECT geoid, ST_Union(ST_MakePolygon(ST_ExteriorRing(the_geom))) AS the_geom, '
             '       MAX(aland) aland '
             'FROM ( '
             '    SELECT geoid, (ST_Dump(the_geom)).geom AS the_geom, '
             '           aland '
             '    FROM {input} '
             ") holes WHERE GeometryType(the_geom) = 'POLYGON' "
             'GROUP BY geoid'.format(
                 output=self.output().table,
                 input=self.input()['data'].table), )[0]
     session.execute(stmt)