def _dataframe_to_sql()

in data_validation/result_handlers/postgres.py [0:0]

• 22 lines of code
• 5 McCabe index (conditional complexity)

    def _dataframe_to_sql(self, schema_name, table_name, result_df):
        """Inserts Dataframe data into PostgreSQL table using pandas.Dataframe.to_sql() method."""

        def label_to_string(label) -> str:
            if isinstance(label, (list, tuple, numpy.ndarray)) and len(label) == 2:
                # This is the expected format
                return f"'{label[0]}={label[1]}'"
            else:
                # Anything else
                return f"'{label}'"

        def labels_to_array_literal(labels):
            """Convert Pandas labels array into a PostgreSQL array literal."""
            if not labels:
                return "{}"
            return "{" + ",".join(label_to_string(_) for _ in labels) + "}"

        result_df[consts.CONFIG_LABELS] = result_df.labels.apply(
            lambda x: labels_to_array_literal(x)
        )
        result_df.to_sql(
            table_name,
            self._client.con,
            schema=schema_name,
            if_exists="append",
            index=False,
            chunksize=1000,
            method=_psql_insert_copy,
        )