# Path: bamboo/lib/async.py
# def call_async(function, *args, **kwargs):
#     """Potentially asynchronously call `function` with the arguments.
# 
#     :param function: The function to call.
#     :param args: Arguments for the function.
#     :param kwargs: Keyword arguments for the function.
#     """
#     countdown = kwargs.pop('countdown', 0)
# 
#     if is_async():
#         function.__getattribute__('apply_async')(
#             countdown=countdown, args=args, kwargs=kwargs)
#     else:  # pragma: no cover
#         function(*args, **kwargs)
# 
# Path: bamboo/lib/datetools.py
# def recognize_dates(df, schema=None):
#     """Convert data columns to datetimes.
# 
#     Check if object columns in a dataframe can be parsed as dates.
#     If yes, rewrite column with values parsed as dates.
# 
#     If schema is passed, convert columes to datetime if column in *schema* is
#     of type datetime.
# 
#     :param df: The DataFrame to convert columns in.
#     :param schema: Schema to define columns of type datetime.
# 
#     :returns: A DataFrame with column values convert to datetime types.
#     """
#     return __parse_dates_schema(df, schema) if schema else __parse_dates(df)
# 
# Path: bamboo/lib/schema_builder.py
# def filter_schema(schema):
#     """Remove not settable columns."""
#     for column, column_schema in schema.iteritems():
#         if column_schema.get(CARDINALITY):
#             del column_schema[CARDINALITY]
#             schema[column] = column_schema
# 
#     return schema


from functools import partial
from celery.exceptions import RetryTaskError
from celery.task import task
from bamboo.lib.async import call_async
from bamboo.lib.datetools import recognize_dates
from bamboo.lib.schema_builder import filter_schema
import simplejson as json
import os
import tempfile
import pandas as pd




@task(ignore_result=True)
def import_dataset(dataset, file_reader, delete=False):
    """For reading a URL and saving the corresponding dataset.

    Import a DataFrame using the provided `file_reader` function. All
    exceptions are caught and on exception the dataset is marked as failed and
    set for deletion after 24 hours.

    :param dataset: The dataset to import into.
    :param file_reader: Function for reading the dataset.
    :param delete: Delete filepath_or_buffer after import, default False.
    """
    try:
        dframe = file_reader()
        dataset.save_observations(dframe)
    except Exception as e:
        if isinstance(e, RetryTaskError):
            raise e
        else:
            dataset.failed(e.__str__())
            dataset.delete(countdown=86400)


def csv_file_reader(name, na_values=[], delete=False):
    try:

# Path: bamboo/lib/async.py
# def call_async(function, *args, **kwargs):
#     """Potentially asynchronously call `function` with the arguments.
# 
#     :param function: The function to call.
#     :param args: Arguments for the function.
#     :param kwargs: Keyword arguments for the function.
#     """
#     countdown = kwargs.pop('countdown', 0)
# 
#     if is_async():
#         function.__getattribute__('apply_async')(
#             countdown=countdown, args=args, kwargs=kwargs)
#     else:  # pragma: no cover
#         function(*args, **kwargs)
# 
# Path: bamboo/lib/datetools.py
# def recognize_dates(df, schema=None):
#     """Convert data columns to datetimes.
# 
#     Check if object columns in a dataframe can be parsed as dates.
#     If yes, rewrite column with values parsed as dates.
# 
#     If schema is passed, convert columes to datetime if column in *schema* is
#     of type datetime.
# 
#     :param df: The DataFrame to convert columns in.
#     :param schema: Schema to define columns of type datetime.
# 
#     :returns: A DataFrame with column values convert to datetime types.
#     """
#     return __parse_dates_schema(df, schema) if schema else __parse_dates(df)
# 
# Path: bamboo/lib/schema_builder.py
# def filter_schema(schema):
#     """Remove not settable columns."""
#     for column, column_schema in schema.iteritems():
#         if column_schema.get(CARDINALITY):
#             del column_schema[CARDINALITY]
#             schema[column] = column_schema
# 
#     return schema


# Path: bamboo/lib/readers.py
from functools import partial
from celery.exceptions import RetryTaskError
from celery.task import task
from bamboo.lib.async import call_async
from bamboo.lib.datetools import recognize_dates
from bamboo.lib.schema_builder import filter_schema
import simplejson as json
import os
import tempfile
import pandas as pd





@task(ignore_result=True)
def import_dataset(dataset, file_reader, delete=False):
    """For reading a URL and saving the corresponding dataset.

    Import a DataFrame using the provided `file_reader` function. All
    exceptions are caught and on exception the dataset is marked as failed and
    set for deletion after 24 hours.

    :param dataset: The dataset to import into.
    :param file_reader: Function for reading the dataset.
    :param delete: Delete filepath_or_buffer after import, default False.
    """
    try:
        dframe = file_reader()
        dataset.save_observations(dframe)
    except Exception as e:
        if isinstance(e, RetryTaskError):
            raise e
        else:
            dataset.failed(e.__str__())
            dataset.delete(countdown=86400)


def csv_file_reader(name, na_values=[], delete=False):
    try:

        return recognize_dates(