def get_logger(name):
    '''
    Obtain a logger outputing to stderr with specified name. Defaults to INFO
    log level.
    '''
    logger = logging.getLogger(name)
    if LUIGI_LOG_ARG in sys.argv:
        index = sys.argv.index(LUIGI_LOG_ARG) + 1
        if len(sys.argv) > index:
            logger.setLevel(sys.argv[index])

    return logger class TableToCarto(Task):

    force = BoolParameter(default=False, significant=False)
    schema = Parameter(default='observatory')
    table = Parameter()
    outname = Parameter(default=None)

    def run(self):
        json_colnames = []
        table = '.'.join([self.schema, self.table])
        if table in metadata.tables:
            cols = metadata.tables[table].columns
            for colname, coldef in list(cols.items()):
                coltype = coldef.type
                if isinstance(coltype, JSON):
                    json_colnames.append(colname)

        sql_to_cartodb_table(self.output().tablename, self.table, json_colnames,
                             schema=self.schema)
        self.force = False

    def output(self):
        if self.schema != 'observatory':
            table = '.'.join([self.schema, self.table])
        else:
            table = self.table
        if self.outname is None:
            self.outname = underscore_slugify(table)
        target = CartoDBTarget(self.outname)
        if self.force and target.exists():
            target.remove()
            self.force = False
        return target class TableToCartoViaImportAPI(Task):
    '''
    This task wraps :func:`~.util.sql_to_cartodb_table` to upload a table to
    a CARTO account specified in the ``.env`` file quickly using the Import
    API.

    :param table: The name of the table to upload, exclusive of schema.
    :param schema: Optional. The schema of the table to upload, defaults to
                   ``observatory``.
    :param force: Optional boolean.  Defaults to ``False``.  If ``True``, a
                  table of the same name existing remotely will be overwritten.
    '''

    force = BoolParameter(default=False, significant=False)
    schema = Parameter(default='observatory')
    username = Parameter(default=None, significant=False)
    api_key = Parameter(default=None, significant=False)
    outname = Parameter(default=None, significant=False)
    table = Parameter()
    columns = ListParameter(default=[])

    def run(self):
        carto_url = 'https://{}.carto.com'.format(self.username) if self.username else os.environ['CARTODB_URL']
        api_key = self.api_key if self.api_key else os.environ['CARTODB_API_KEY']
        try:
            os.makedirs(os.path.join('tmp', classpath(self)))
        except OSError:
            pass
        outname = self.outname or self.table
        tmp_file_path = os.path.join('tmp', classpath(self), outname + '.csv')
        if not self.columns:
            shell(r'''psql -c '\copy "{schema}".{tablename} TO '"'"{tmp_file_path}"'"'
                  WITH CSV HEADER' '''.format(
                      schema=self.schema,
                      tablename=self.table,
                      tmp_file_path=tmp_file_path,
                  ))
        else:
            shell(r'''psql -c '\copy (SELECT {columns} FROM "{schema}".{tablename}) TO '"'"{tmp_file_path}"'"'
                  WITH CSV HEADER' '''.format(
                      schema=self.schema,
                      tablename=self.table,
                      tmp_file_path=tmp_file_path,
                      columns=', '.join(self.columns),
                  ))
        curl_resp = shell(
            'curl -s -F privacy=public -F type_guessing=false '
            '  -F file=@{tmp_file_path} "{url}/api/v1/imports/?api_key={api_key}"'.format(
                tmp_file_path=tmp_file_path,
                url=carto_url,
                api_key=api_key
            ))
        try:
            import_id = json.loads(curl_resp)["item_queue_id"]
        except ValueError:
            raise Exception(curl_resp)
        while True:
            resp = requests.get('{url}/api/v1/imports/{import_id}?api_key={api_key}'.format(
                url=carto_url,
                import_id=import_id,
                api_key=api_key
            ))
            if resp.json()['state'] == 'complete':
                LOGGER.info("Waiting for import %s for %s", import_id, outname)
                break
            elif resp.json()['state'] == 'failure':
                raise Exception('Import failed: {}'.format(resp.json()))

            print(resp.json()['state'])
            time.sleep(1)

        # If CARTO still renames our table to _1, just force alter it
        if resp.json()['table_name'] != outname:
            query_cartodb('ALTER TABLE {oldname} RENAME TO {newname}'.format(
                oldname=resp.json()['table_name'],
                newname=outname,
                carto_url=carto_url,
                api_key=api_key,
            ))
            assert resp.status_code == 200

        # fix broken column data types -- alter everything that's not character
        # varying back to it
        try:
            session = current_session()
            resp = session.execute(
                '''
                SELECT att.attname,
                       pg_catalog.format_type(atttypid, NULL) AS display_type,
                       att.attndims
                FROM pg_attribute att
                  JOIN pg_class tbl ON tbl.oid = att.attrelid
                  JOIN pg_namespace ns ON tbl.relnamespace = ns.oid
                WHERE tbl.relname = '{tablename}'
                  AND pg_catalog.format_type(atttypid, NULL) NOT IN
                      ('character varying', 'text', 'user-defined', 'geometry')
                  AND att.attname IN (SELECT column_name from information_schema.columns
                                      WHERE table_schema='{schema}'
                                        AND table_name='{tablename}')
                  AND ns.nspname = '{schema}';
                '''.format(schema=self.schema,
                           tablename=self.table.lower())).fetchall()
            alter = ', '.join([
                " ALTER COLUMN {colname} SET DATA TYPE {data_type} "
                " USING NULLIF({colname}, '')::{data_type}".format(
                    colname=colname, data_type=data_type
                ) for colname, data_type, _ in resp])
            if alter:
                alter_stmt = 'ALTER TABLE {tablename} {alter}'.format(
                    tablename=outname,
                    alter=alter)
                LOGGER.info(alter_stmt)
                resp = query_cartodb(alter_stmt, api_key=api_key, carto_url=carto_url)
                if resp.status_code != 200:
                    raise Exception('could not alter columns for "{tablename}":'
                                    '{err}'.format(tablename=outname,
                                                   err=resp.text))
        except Exception as err:
            # in case of error, delete the uploaded but not-yet-properly typed
            # table
            self.output().remove(carto_url=carto_url, api_key=api_key)
            raise err

    def output(self):
        carto_url = 'https://{}.carto.com'.format(self.username) if self.username else os.environ['CARTODB_URL']
        api_key = self.api_key if self.api_key else os.environ['CARTODB_API_KEY']
        target = CartoDBTarget(self.outname or self.table, api_key=api_key, carto_url=carto_url)
        if self.force:
            target.remove(carto_url=carto_url, api_key=api_key)
            self.force = False
        return target def current_session():
    '''
    Returns the session relevant to the currently operating :class:`Task`, if
    any.  Outside the context of a :class:`Task`, this can still be used for
    manual session management.
    '''
    return _current_session.get() class OBSTable(Base):
    '''
    Describes a physical table in our database.

    These should *never* be instantiated manually.  They are automatically
    created by :meth:`~.tasks.TableTask.output`.  The unique key is
    :attr:~.meta.OBSTable.id:.

    .. py:attribute:: id

       The unique identifier for this table.  Is always qualified by
       the module name of the class that created it.

    .. py:attribute:: columns

       An iterable of all the :class:`~.meta.OBSColumnTable` instances
       contained in this table.

    .. py:attribute:: tablename

       The automatically generated name of this table, which can be used
       directly in select statements.  Of the format ``obs_<hash>``.

    .. py:attribute:: timespan

       An OBSTimespan instance containing information about the timespan
       this table applies to.
       Obtained from :meth:`~.tasks.TableTask.timespan`.

    .. py:attribute:: the_geom

       A simple geometry approximating the boundaries of the data in this
       table.

    .. py:attribute:: description

       A description of the table.  Not used.

    .. py:attribute:: version

       A version control number, used to determine whether the table and its
       metadata should be updated.

    '''
    __tablename__ = 'obs_table'

    id = Column(Text, primary_key=True)  # fully-qualified id like 'us.census.acs.extract_2013_5yr_state_18357fba'

    columns = relationship("OBSColumnTable", back_populates="table",
                           cascade="all,delete")

    tablename = Column(Text, nullable=False)
    timespan = Column(Text, ForeignKey('obs_timespan.id'))
    table_timespan = relationship("OBSTimespan")

    the_geom = Column(Geometry)
    description = Column(Text)

    targets = association_proxy('tgts', 'reltype', creator=tabletotabletargets_creator)

    version = Column(Numeric, default=0, nullable=False)

    def geom_column(self):
        '''
        Return the column geometry column for this table, if it has one.

        Returns None if there is none.
        '''
        for col in self.columns:
            if lower(col.type) == 'geometry':
                return col

    def geomref_column(self):
        '''
        Return the geomref column for this table, if it has one.

        Returns None if there is none.
        '''
        for col in self.columns:
            col = col.column
            if col.is_geomref():
                return col class OBSColumn(Base):
    '''
    Describes the characteristics of data in a column, which can exist in
    multiple physical tables.

    These should *only* be instantiated to generate the return value of
    :meth:`.ColumnsTask.columns`.  Any other usage could have unexpected
    consequences.

    .. py:attribute:: id

       The unique identifier for this column.  Should be qualified by
       the module name of the class that created it.  Is automatically
       generated by :class:`~.tasks.ColumnsTask` if left unspecified,
       and automatically qualified by module name either way.

    .. py:attribute:: type

       The type of this column -- for example, ``Text``, ``Numeric``,
       ``Geometry``, etc.  This is used to generate a schema for any
       table using this column.

    .. py:attribute:: name

       The human-readable name of this column.  This is used in the
       catalog and API.

    .. py:attribute:: description

       The human-readable description of this column.  THis is
       used in the catalog.

    .. py:attribute:: weight

       A numeric weight for this column.  Higher weights are favored
       in certain rankings done by the API. Defaults to zero, which hides
       the column from catalog.

    .. py:attribute:: aggregate

       How this column can be aggregated.  For example,
       populations can be summed, so a population column should
       be ``sum``. Should be left blank if it is not to
       aggregate.

    .. py:attribute:: tables

       Iterable of all linked :class:`~.meta.OBSColumnTable`s, which could be
       traversed to find all tables with this column.

    .. py:attribute:: tags

       Iterable of all linked :class:`~.meta.OBSColumnTag`s, which could be
       traversed to find all tags applying to this column.

    .. py:attribute:: targets

       Dict of all related columns.  Format is ``<OBSColumn>: <reltype>``.

    .. py:attribute:: version

       A version control number, used to determine whether the column and its
       metadata should be updated.

    .. py:attribute:: extra

       Arbitrary additional information about this column stored as JSON.

    '''
    __tablename__ = 'obs_column'

    id = Column(Text, primary_key=True) # fully-qualified id like '"us.census.acs".b01001001'

    type = Column(Text, nullable=False) # postgres type, IE numeric, string, geometry, etc.
    name = Column(Text) # human-readable name to provide in bigmetadata

    description = Column(Text) # human-readable description to provide in
                                 # bigmetadata

    weight = Column(Numeric, default=1)
    aggregate = Column(Text) # what aggregate operation to use when adding
                               # these together across geoms: AVG, SUM etc.

    tables = relationship("OBSColumnTable", back_populates="column", cascade="all,delete")
    session = current_session()
    # Passing the same session object to the creator saves thousands
    # of `select 1` queries, because each use of the session was pinging
    tag_creator_with_session = functools.partial(tag_creator, session=session)
    tags = association_proxy('column_column_tags', 'tag', creator=tag_creator_with_session)

    targets = association_proxy('tgts', 'reltype', creator=coltocoltargets_creator)

    version = Column(Numeric, default=0, nullable=False)
    extra = Column(JSON)

    @property
    def sources(self):
        sources = {}
        session = current_session()
        for c2c in session.query(OBSColumnToColumn).filter_by(target_id=self.id):
            sources[c2c.source] = c2c.reltype
        return sources

    @property
    def index_type(self):
        if hasattr(self, '_index_type'):
            pass
        elif 'geom_ref' in list(self.targets.values()):
            self._index_type = 'btree'
        elif self.type.lower() == 'geometry':
            self._index_type = 'gist'
        else:
            self._index_type = None

        return self._index_type

    def children(self):
        children = [col for col, reltype in self.sources.items() if reltype == DENOMINATOR]
        children.sort(key=lambda x: natural_sort_key(x.name))
        return children

    def is_cartographic(self):
        '''
        Returns True if this column is a geometry that can be used for cartography.
        '''
        for tag in self.tags:
            if 'cartographic_boundary' in tag.id:
                return True
        return False

    def is_interpolation(self):
        '''
        Returns True if this column is a geometry that can be used for interpolation.
        '''
        for tag in self.tags:
            if 'interpolation_boundary' in tag.id:
                return True
        return False

    def is_geomref(self):
        '''
        Returns True if the column is a geomref, else Null
        '''
        return GEOM_REF in list(self.targets.values())

    def has_children(self):
        '''
        Returns True if this column has children, False otherwise.
        '''
        return len(self.children()) > 0

    def has_denominators(self):
        '''
        Returns True if this column has no denominator, False otherwise.
        '''
        return DENOMINATOR in list(self.targets.values())

    def has_catalog_image(self):
        '''
        Returns True if this column has a pre-generated image for the catalog.
        '''
        return os.path.exists(os.path.join('catalog', 'img', self.id + '.png'))

    def denominators(self):
        '''
        Return the :class:`~.meta.OBSColumn` denominator of this column.
        '''
        if not self.has_denominators():
            return []
        return [k for k, v in self.targets.items() if v == DENOMINATOR]

    def unit(self):
        '''
        Return the :class:`~.meta.OBSTag` unit of this column.
        '''
        units = [tag for tag in self.tags if tag.type == 'unit']
        if units:
            return units[0]

    def summable(self):
        '''
        Returns True if we can sum this column and calculate for arbitrary
        areas.
        '''
        return self.aggregate == 'sum'

    def catalog_lonlat(self):
        '''
        Return tuple (longitude, latitude) for the catalog for this measurement.
        '''
        return catalog_lonlat(self.id)

    def geom_timespans(self):
        '''
        Return a dict of geom columns and timespans that this measure is
        available for.
        '''
        geom_timespans = {}
        for table in self.tables:
            table = table.table
            geomref_column = table.geomref_column()
            geom_column = [t for t, rt in geomref_column.targets.items()
                           if rt == GEOM_REF][0]
            if geom_column not in geom_timespans:
                geom_timespans[geom_column] = []
            geom_timespans[geom_column].append(table.timespan)
        return geom_timespans

    def source_tags(self):
        '''
        Return source tags.
        '''
        return [tag for tag in self.tags if tag.type.lower() == 'source']

    def license_tags(self):
        '''
        Return license tags.
        '''
        return [tag for tag in self.tags if tag.type.lower() == 'license'] class UpdatedMetaTarget(Target):
    def exists(self):
        session = current_session()

        # identify tags that have appeared, changed or disappeared
        # version shifts won't crop up in and of themselves, but name changes will
        changed_tags = '''
        with tags as (
            SELECT jsonb_each_text(numer_tags)
            FROM observatory.obs_meta_numer
            UNION ALL
            SELECT jsonb_each_text(denom_tags)
            FROM observatory.obs_meta_denom
            UNION ALL
            SELECT jsonb_each_text(geom_tags)
            FROM observatory.obs_meta_geom
        ),
        distinct_tags as (
            SELECT DISTINCT
            SPLIT_PART((jsonb_each_text).key, '/', 1) AS type,
            SPLIT_PART((jsonb_each_text).key, '/', 2) AS id,
            ((jsonb_each_text).value)::TEXT AS name
            FROM tags
        ),
        meta as (
            SELECT distinct t.id, t.type, t.name
            FROM observatory.obs_tag t
            JOIN observatory.obs_column_tag ct ON t.id = ct.tag_id
            JOIN observatory.obs_column c ON ct.column_id = c.id
            JOIN observatory.obs_column_table ctab ON c.id = ctab.column_id
        )
        SELECT meta.id, meta.type, meta.name,
                distinct_tags.id, distinct_tags.type, distinct_tags.name
        FROM distinct_tags FULL JOIN meta
        ON meta.id = distinct_tags.id AND
            meta.type = distinct_tags.type AND
            meta.name = distinct_tags.name
        WHERE meta.id IS NULL OR distinct_tags.id IS NULL
        '''
        resp = session.execute(changed_tags)
        if resp.fetchone():
            return False

        # identify columns that have appeared, changed, or disappeared
        changed_columns = '''
        WITH columns AS (
            SELECT DISTINCT c.id, c.version
            FROM observatory.obs_column c
                LEFT JOIN observatory.obs_column_to_column c2c ON c.id = c2c.source_id,
                observatory.obs_column_tag ct,
                observatory.obs_tag t,
                observatory.obs_column_table ctab
            WHERE c.id = ct.column_id
            AND ct.tag_id = t.id
            AND c.type NOT ILIKE 'geometry%'
            AND c.weight > 0
            AND (c2c.reltype != 'geom_ref' OR c2c.reltype IS NULL)
            AND c.id = ctab.column_id
        ),
        meta AS (
            SELECT numer_id, numer_version FROM observatory.obs_meta_numer
        )
        SELECT id, version, numer_id, numer_version
        FROM columns FULL JOIN meta
        ON id = numer_id AND version = numer_version
        WHERE numer_id IS NULL OR id IS NULL;
        '''
        resp = session.execute(changed_columns)
        if resp.fetchone():
            return False

        # identify tables that have appeared, changed, or disappeared
        changed_tables = '''
        with numer_tables as (
            select distinct tab.id, tab.version
            from observatory.obs_table tab,
                observatory.obs_column_table ctab,
                observatory.obs_column c
                    left join observatory.obs_column_to_column c2c on c.id = c2c.source_id,
                observatory.obs_column_tag ct,
                observatory.obs_tag t
                where c.id = ct.column_id
                    and ct.tag_id = t.id
                    and ctab.column_id = c.id
                    and ctab.table_id = tab.id
                    and c.type not ilike 'geometry%'
                    and c.weight > 0
                    and (c2c.reltype != 'geom_ref' or c2c.reltype is null)
        ),
        geom_tables as (
            select tab.id, tab.version,
                    rank() over (partition by c.id order by tab.timespan desc)
            from observatory.obs_table tab,
                observatory.obs_column_table ctab,
                observatory.obs_column c,
                observatory.obs_column_to_column c2c,
                observatory.obs_column_tag ct,
                observatory.obs_tag t,
                observatory.obs_column c2,
                observatory.obs_column_table ctab2,
                numer_tables
                where c.id = ct.column_id
                    and ct.tag_id = t.id
                    and ctab.column_id = c.id
                    and ctab.table_id = tab.id
                    and c.type ilike 'geometry%'
                    and c.weight > 0
                    and c2c.reltype = 'geom_ref'
                    and c.id = c2c.target_id
                    and c2.id = c2c.source_id
                    and c2.id = ctab2.column_id
                    and numer_tables.id = ctab2.table_id
            group by c.id, tab.id, tab.version
        ),
        numer_meta as (
            select distinct numer_tid, numer_t_version
            from observatory.obs_meta
        ),
        geom_meta as (
            select distinct geom_tid, geom_t_version
            from observatory.obs_meta
        )
        select distinct id, version, numer_tid, numer_t_version
        from numer_tables full join numer_meta on
            id = numer_tid and version = numer_t_version
        where numer_tid is null or id is null
        union all
        select distinct id, version, geom_tid, geom_t_version
        from geom_tables full join geom_meta on
            id = geom_tid and version = geom_t_version
        where rank = 1 and (geom_tid is null or id is null)
        ;
        '''
        resp = session.execute(changed_tables)
        if resp.fetchone():
            return False
        return True def underscore_slugify(txt):
    '''
    Given a string, converts from camelcase to underscore style, and then
    slugifies.

    >>> underscore_slugify('FooBarBaz')
    'foo_bar_baz'

    >>> underscore_slugify('Foo, bar, baz')
    'foo_bar_baz'

    >>> underscore_slugify('Foo_Bar_Baz')
    'foo_bar_baz'

    >>> underscore_slugify('Foo:barBaz')
    'foo_bar_baz'
    '''
    return slugify(camel_to_underscore(re.sub(
        r'[^a-zA-Z0-9]+', '_', txt))).replace('-', '_') def query_cartodb(query, carto_url=None, api_key=None):
    '''
    Convenience function to query CARTO's SQL API with an arbitrary SQL string.
    The account connected via ``.env`` is queried.

    Returns the raw ``Response`` object.  Will not raise an exception in case
    of non-200 response.

    :param query: The query to execute on CARTO.
    '''
    carto_url = (carto_url or os.environ['CARTODB_URL']) + '/api/v2/sql'
    resp = requests.post(carto_url, data={
        'api_key': api_key or os.environ['CARTODB_API_KEY'],
        'q': query
    })
    return resp def classpath(obj):
    '''
    Returns the path to this object's class, relevant to current working dir.
    Excludes the first element of the path.  If there is only one element,
    returns ``tmp``.

    >>> classpath(object)
    'tmp'
    >>> from tasks.base_tasks import ColumnsTask
    >>> classpath(ColumnsTask())
    'util'
    >>> from tasks.es.ine import FiveYearPopulation
    >>> classpath(FiveYearPopulation())
    'es.ine'
    '''
    classpath_ = '.'.join(obj.__module__.split('.')[1:])
    return classpath_ if classpath_ else 'tmp' def shell(cmd, encoding='utf-8'):
    '''
    Run a shell command, uses :py:func:`subprocess.check_output(cmd,
    shell=True)` under the hood.

    Returns the ``STDOUT`` output, and raises an error if there is a
    none-zero exit code.
    '''
    try:
        return subprocess.check_output(cmd, shell=True, stderr=subprocess.STDOUT).decode(encoding)
    except subprocess.CalledProcessError as err:
        LOGGER.error(err.output)
        raise def unqualified_task_id(task_id):
    '''
    Returns the name of the task from the task_id.
    '''
    return task_id.split('.')[-1] class PostgresTarget(Target):
    '''
    PostgresTarget which by default uses command-line specified login.
    '''

    def __init__(self, schema, tablename, non_empty=True, where="1 = 1"):
        self._schema = schema
        self._tablename = tablename
        self._non_empty = non_empty
        self._where = where

    @property
    def table(self):
        return '"{schema}".{tablename}'.format(schema=self._schema,
                                               tablename=self._tablename)

    @property
    def tablename(self):
        return self._tablename

    @property
    def schema(self):
        return self._schema

    @property
    def qualified_tablename(self):
        return '"{}".{}'.format(self.schema, self.tablename)

    def _existenceness(self):
        '''
        Returns 0 if the table does not exist, 1 if it exists but has no
        rows (is empty), and 2 if it exists and has one or more rows.
        '''
        session = current_session()
        sql = '''
                SELECT COUNT(*) FROM information_schema.tables
                WHERE table_schema ILIKE '{schema}'
                  AND table_name ILIKE '{tablename}'
              '''.format(
            schema=self._schema,
            tablename=self._tablename)
        resp = session.execute(sql)
        if int(resp.fetchone()[0]) == 0:
            return 0
        resp = session.execute(
            'SELECT row_number() over () FROM "{schema}".{tablename} WHERE {where} LIMIT 1'.format(
                schema=self._schema, tablename=self._tablename,
                where=self._where))
        if resp.fetchone() is None:
            return 1
        else:
            return 2

    def empty(self):
        '''
        Returns True if the table exists but has no rows in it.
        '''
        return self._existenceness() == 1

    def exists(self):
        '''
        Returns True if the table exists and has at least one row in it.
        '''
        if self._non_empty:
            return self._existenceness() == 2
        else:
            return self._existenceness() >= 1

    def exists_or_empty(self):
        '''
        Returns True if the table exists, even if it is empty.
        '''
        return self._existenceness() >= 1 class CartoDBTarget(Target):
    '''
    Target which is a CartoDB table
    '''

    def __init__(self, tablename, carto_url=None, api_key=None):
        self.tablename = tablename
        self.carto_url = carto_url
        self.api_key = api_key

    def __str__(self):
        return self.tablename

    def exists(self):
        resp = query_cartodb(
            'SELECT row_number() over () FROM "{tablename}" LIMIT 1'.format(
                tablename=self.tablename),
            api_key=self.api_key,
            carto_url=self.carto_url)
        if resp.status_code != 200:
            return False
        return resp.json()['total_rows'] > 0

    def remove(self, carto_url=None, api_key=None):
        api_key = api_key or os.environ['CARTODB_API_KEY']

        try:
            while True:
                resp = requests.get('{url}/api/v1/tables/{tablename}?api_key={api_key}'.format(
                    url=carto_url,
                    tablename=self.tablename,
                    api_key=api_key
                ))
                viz_id = resp.json()['id']
                # delete dataset by id DELETE
                # https://observatory.cartodb.com/api/v1/viz/ed483a0b-7842-4610-9f6c-8591273b8e5c
                try:
                    requests.delete('{url}/api/v1/viz/{viz_id}?api_key={api_key}'.format(
                        url=carto_url,
                        viz_id=viz_id,
                        api_key=api_key
                    ), timeout=1)
                except requests.Timeout:
                    pass
        except ValueError:
            pass
        query_cartodb('DROP TABLE IF EXISTS {tablename}'.format(tablename=self.tablename))
        assert not self.exists()
from lib.logger import get_logger
from tasks.base_tasks import TableToCarto, TableToCartoViaImportAPI
from tasks.meta import current_session, OBSTable, OBSColumn, UpdatedMetaTarget
from tasks.util import underscore_slugify, query_cartodb, classpath, shell, unqualified_task_id
from tasks.targets import PostgresTarget, CartoDBTarget
from luigi import (WrapperTask, BoolParameter, Parameter, Task, LocalTarget,
                   DateParameter, IntParameter)
from luigi.task_register import Register
from luigi.contrib.s3 import S3Target
from datetime import date
import time
import os
                'SELECT * FROM obs_table'
            ).json()['rows']
        ])
        tables = dict([(k, v) for k, v in current_session().execute(
            '''
            SELECT tablename, t.version
            FROM observatory.obs_table t,
                 observatory.obs_column_table ct,
                 observatory.obs_column c
            WHERE t.id = ct.table_id
              AND c.id = ct.column_id
              AND t.tablename NOT IN ('obs_ffebc3eb689edab4faa757f75ca02c65d7db7327')
              AND c.weight > 0
            '''
        )])

        for tablename, version in tables.items():
            if version > existing_table_versions.get(tablename):
                force = True
            else:
                force = self.force
            yield TableToCartoViaImportAPI(table=tablename, force=force)


class PurgeMetadataTasks(Task):
    '''
    Purge local metadata tables that no longer have tasks linking to them
    '''
    pass


class PurgeMetadataColumns(Task):
    '''
    Purge local metadata tables that no longer have tasks linking to them
    '''
    pass


class PurgeUndocumentedTables(Task):
    '''
    Purge tables that should be in metadata but are not.
    '''

    def run(self):
        session = current_session()
        resp = session.execute('SELECT table_schema, table_name '
                               'FROM information_schema.tables '
                               "WHERE table_schema ILIKE 'observatory' ")
        for _, tablename in resp:
            if tablename in ('obs_table', 'obs_column_table', 'obs_column',
                             'obs_tag', 'obs_column_to_column', 'obs_column_tag'):
                continue
            if session.query(OBSTable).filter_by(tablename=tablename).count() == 0:
                cnt = session.execute('SELECT COUNT(*) FROM observatory.{tablename}'.format(
                    tablename=tablename)).fetchone()[0]
                if cnt == 0:
                    stmt = 'DROP TABLE observatory.{tablename} CASCADE'.format(
                        tablename=tablename)
                    LOGGER.info(stmt)
                    session.execute(stmt)
                    session.commit()
                else:
                    raise Exception("Will not automatically drop table {tablename} "
                                    "with data in it".format(tablename=tablename))


class PurgeMetadataTables(Task):
    '''
    Purge local metadata tables that no longer have tasks linking to them,
    as well as entries in obs_table that do not link to any table.
    '''

    def run(self):
        session = current_session()
        for _output in self.output():
            if not _output.exists():
                resp = session.execute("SELECT id from observatory.obs_table "
                                       "WHERE tablename = '{tablename}'".format(
                                           tablename=_output.tablename))
                _id = resp.fetchall()[0][0]
                stmt = "DELETE FROM observatory.obs_table " \
                        "WHERE id = '{id}'".format(id=_id)
                LOGGER.info(stmt)
                session.execute(stmt)
                session.commit()

    def output(self):
        session = current_session()
        for table in session.query(OBSTable):
            split = table.id.split('.')
            schema, task_id = split[0:-1], split[-1]
            modname = 'tasks.' + '.'.join(schema)
            module = __import__(modname, fromlist=['*'])
            exists = False
            for name in dir(module):
                kls = getattr(module, name)
                if not isinstance(kls, Register):
                    continue
                if task_id.startswith(underscore_slugify(name)):
                    exists = True
            if exists is True:
                LOGGER.info('{table} exists'.format(table=table))
            else:
                # TODO drop table
                LOGGER.info(table)
            yield PostgresTarget(schema='observatory', tablename=table.tablename)


class ConfirmTableExists(Task):
    '''
    Confirm a table exists
    '''

    schema = Parameter(default='observatory')
    tablename = Parameter()

    def run(self):
        raise Exception('Table {} does not exist'.format(self.tablename))

    def output(self):

        return PostgresTarget(self.schema, self.tablename)