https://github.com/frictionlessdata/tableschema-pandas-py#storage
761528
def create(self, bucket, descriptor, force=False):
        

        # Make lists
        buckets = bucket
        if isinstance(bucket, six.string_types):
            buckets = [bucket]
        descriptors = descriptor
        if isinstance(descriptor, dict):
            descriptors = [descriptor]

        # Check buckets for existence
        for bucket in buckets:
            if bucket in self.buckets:
                if not force:
                    message = 'Bucket "%s" already exists' % bucket
                    raise tableschema.exceptions.StorageError(message)
                self.delete(bucket)

        # Define dataframes
        for bucket, descriptor in zip(buckets, descriptors):
            tableschema.validate(descriptor)
            self.__descriptors[bucket] = descriptor
            self.__dataframes[bucket] = pd.DataFrame()
842054
def smart_insert(df, table, engine, minimal_size=5):
    
    from sqlalchemy.exc import IntegrityError

    try:
        table_name = table.name
    except:
        table_name = table

    # 首先进行尝试bulk insert
    try:
        df.to_sql(table_name, engine, index=False, if_exists="append")
    # 失败了
    except IntegrityError:
        # 分析数据量
        n = df.shape[0]
        # 如果数据条数多于一定数量
        if n >= minimal_size ** 2:
            # 则进行分包
            n_chunk = math.floor(math.sqrt(n))
            for sub_df in grouper_df(df, n_chunk):
                smart_insert(
                    sub_df, table_name, engine, minimal_size)
        # 否则则一条条地逐条插入
        else:
            for sub_df in grouper_df(df, 1):
                try:
                    sub_df.to_sql(
                        table_name, engine, index=False, if_exists="append")
                except IntegrityError:
                    pass
793098
def save_load(jid, load, minions=None):
    '''
    
    '''
    with _get_serv(commit=True) as cur:
        try:
            cur.execute(PG_SAVE_LOAD_SQL,
                        {'jid': jid, 'load': psycopg2.extras.Json(load)})
        except psycopg2.IntegrityError:
            # https://github.com/saltstack/salt/issues/22171
            # Without this try/except we get tons of duplicate entry errors
            # which result in job returns not being stored properly
            pass
1362736
def store(self, loc, df):
        

        path = "%s.%s" % (self._root / "processed" / loc, FILE_EXTENSION)
        WRITE_DF(df, path, **WRITE_DF_OPTS)
        self._cache[loc] = df
351040
def schema(self):
        
        attrs = self.attrs.copy()
        parts = ["CREATE", "TABLE", self.name, "(%s," % self.hash_key.schema]
        del attrs[self.hash_key.name]
        if self.range_key:
            parts.append(self.range_key.schema + ",")
            del attrs[self.range_key.name]
        if attrs:
            attr_def = ", ".join([attr.schema for attr in itervalues(attrs)])
            parts.append(attr_def + ",")

        parts.append(
            "THROUGHPUT (%d, %d))" % (self.read_throughput, self.write_throughput)
        )
        parts.extend([g.schema for g in itervalues(self.global_indexes)])
        return " ".join(parts) + ";"
173030
def to_prettytable(df):
    
    pt = PrettyTable()
    pt.field_names = df.columns
    for tp in zip(*(l for col, l in df.iteritems())):
        pt.add_row(tp)
    return pt
537161
def get_version_storage():
    # type: () -> VersionStorage
    
    version_file = conf.get_path('version_file', 'VERSION')
    if version_file.endswith('.py'):
        return PyVersionStorage(version_file)
    elif version_file.endswith('package.json'):
        return NodeVersionStorage(version_file)
    else:
        return RawVersionStorage(version_file)
1371475
def dcounts(self):
        
        print("WARNING: Distinct value count for all tables can take a long time...", file=sys.stderr)
        sys.stderr.flush()

        data = []
        for t in self.tables():
            for c in t.columns():
                data.append([t.name(), c.name(), c.dcount(), t.size(), c.dcount() / float(t.size())])
        df = pd.DataFrame(data, columns=["table", "column", "distinct", "size", "fraction"])
        return df
1038605
def to_sql(cls, qc, **kwargs):
        
        # we first insert an empty DF in order to create the full table in the database
        # This also helps to validate the input against pandas
        # we would like to_sql() to complete only when all rows have been inserted into the database
        # since the mapping operation is non-blocking, each partition will return an empty DF
        # so at the end, the blocking operation will be this empty DF to_pandas

        empty_df = qc.head(1).to_pandas().head(0)
        empty_df.to_sql(**kwargs)
        # so each partition will append its respective DF
        kwargs["if_exists"] = "append"
        columns = qc.columns

        def func(df, **kwargs):
            df.columns = columns
            df.to_sql(**kwargs)
            return pandas.DataFrame()

        map_func = qc._prepare_method(func, **kwargs)
        result = qc._map_across_full_axis(1, map_func)
        # blocking operation
        result.to_pandas()
804891
def _to_df(self, query, conn, index_col=None, coerce_float=True, params=None,
               parse_dates=None, columns=None):
        
        return pd.io.sql.read_sql(str(query), conn, index_col=index_col,
                                  coerce_float=coerce_float, params=params,
                                  parse_dates=parse_dates, columns=columns)
575078
def _data(self, asa=None):
        
        hash_list = []
        stack = [self]
        alldfs = True
        store = None
        while stack:
            node = stack.pop()
            if isinstance(node, GroupNode):
                stack.extend(child for _, child in sorted(node._items(), reverse=True))
            else:
                if node._target() != TargetType.PANDAS:
                    alldfs = False
                if node._store is None or node._hashes is None:
                    msg = "Can only merge built dataframes. Build this package and try again."
                    raise NotImplementedError(msg)
                node_store = node._store
                if store is None:
                    store = node_store
                if node_store != store:
                    raise NotImplementedError("Can only merge dataframes from the same store")
                hash_list += node._hashes

        if asa is None:
            if not hash_list:
                return None
            if not alldfs:
                raise ValueError("Group contains non-dataframe nodes")
            return store.load_dataframe(hash_list)
        else:
            if hash_list:
                assert store is not None
                return asa(self, [store.object_path(obj) for obj in hash_list])
            else:
                return asa(self, [])
992719
def save_to_data(self, in_place=False):
        
        if in_place:
            return [
                list(self._index.items()),
                list(self._undefined_keys.keys())
            ]
        return (
            [(key, values[:]) for key, values in self._index.items()],
            list(self._undefined_keys.keys()),
        )
1049960
def table_as_df(self, table_name):
        """"""
        self.table_must_exist(table_name)
        query = "SELECT * FROM `%s`" % table_name.lower()
        return pandas.read_sql(query, self.own_conn)
214523
def cached_dataframe(self, csv_path, compute_fn):
        
        if not csv_path.endswith(".csv"):
            raise ValueError("Invalid path '%s', must be a CSV file" % csv_path)

        if csv_path in self._memory_cache:
            return self._memory_cache[csv_path]

        if exists(csv_path) and not self.is_empty(csv_path):
            df = self._read_csv(csv_path)
        else:
            df = compute_fn()
            if not isinstance(df, pd.DataFrame):
                raise TypeError(
                    "Expected compute_fn to return DataFrame, got %s : %s" % (
                        df, type(df)))
            self._write_csv(df, csv_path)
        self._memory_cache[csv_path] = df
        return df
1321768
def lipd_to_df(metadata, csvs):
    
    dfs = {}
    logger_dataframes.info("enter lipd_to_df")

    # Flatten the dictionary, but ignore the chron data items
    dict_in_dotted = {}
    logger_dataframes.info("enter dot_notation")
    _dotnotation_for_nested_dictionary(metadata, '', dict_in_dotted)
    dict_in_dotted = collections.OrderedDict(sorted(dict_in_dotted.items()))

    # Create one data frame for metadata items
    dfs["metadata"] = pd.DataFrame(list(dict_in_dotted.items()), columns=["Key", "Value"])

    # Create data frames for paleo data and chron data items. This does not use LiPD data, it uses the csv data
    dfs.update(_get_dfs(csvs))

    return dfs
894525
def fetch_csv_db(
        table_name,
        download_url,
        csv_filename=None,
        db_filename=None,
        subdir=None,
        version=1,
        **pandas_kwargs):
    
    df = fetch_csv_dataframe(
        download_url=download_url,
        filename=csv_filename,
        subdir=subdir,
        **pandas_kwargs)
    base_filename = splitext(csv_filename)[0]
    if db_filename is None:
        db_filename = _db_filename_from_dataframe(base_filename, df)
    return db_from_dataframe(
        db_filename,
        table_name,
        df,
        subdir=subdir,
        version=version)
1133162
def save_to_store(self):
        
        if not self._store:
            raise AttributeError('No datastore defined!')
        saved_data = self.save_to_data(in_place=True)
        data = Serializer.serialize(saved_data)
        self._store.store_blob(data, 'all_keys_with_undefined')