def _insert_postgres()

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

• 19 lines of code
• 6 McCabe index (conditional complexity)

    def _insert_postgres(self, result_df: "DataFrame"):
        """Store the validation results Dataframe to an Ibis Backend."""

        if "." in self._table_id:
            schema_name, table_name = self._table_id.split(".")
        else:
            schema_name, table_name = None, self._table_id

        if schema_name:
            self._set_current_schema(schema_name)
        try:
            _ = clients.get_ibis_table(self._client, schema_name, table_name)
            # Do nothing, the table exists.
        except sqlalchemy.exc.NoSuchTableError:
            self._client.create_table(table_name, schema=RESULTS_TABLE_SCHEMA)

        if not result_df.empty:
            self._dataframe_to_sql(schema_name, table_name, result_df)

        if result_df.empty:
            logging.info(RH_NO_WRITE_MESSAGE)
        else:
            logging.info(
                f"{RH_WRITE_MESSAGE} to {self._table_id}, run id: {result_df.iloc[0][consts.CONFIG_RUN_ID]}"
            )