class_docstr: Extends the AdapterBackedSqlClient with the DDL methods necessary for test configuration and execution.
class_docstr_tok: ["Extends","the","AdapterBackedSqlClient","with","the","DDL","methods","necessary","for","test","configuration","and","execution","."]
docstr: Create a table in the data warehouse containing the contents of the dataframe.
  
  Only used in tutorials and tests.
  
  Args:
      sql_table: The SqlTable object representing the table location to use
      df: The Pandas DataFrame object containing the column schema and data to load
      chunk_size: The number of rows to insert per transaction
docstr_tok: ["Create","a","table","in","the","data","warehouse","containing","the","contents","of","the","dataframe",".","Only","used","in","tutorials","and","tests",".","Args",":","sql_table",":","The","SqlTable","object","representing","the","table","location","to","use","df",":","The","Pandas","DataFrame","object","containing","the","column","schema","and","data","to","load","chunk_size",":","The","number","of","rows","to","insert","per","transaction"]
code: def create_table_from_dataframe(
      self,
      sql_table: SqlTable,
      df: pd.DataFrame,
      chunk_size: Optional[int] = None,
  ) -> None:
      """Create a table in the data warehouse containing the contents of the dataframe.
  
      Only used in tutorials and tests.
  
      Args:
          sql_table: The SqlTable object representing the table location to use
          df: The Pandas DataFrame object containing the column schema and data to load
          chunk_size: The number of rows to insert per transaction
      """
      logger.info(
          f"Creating table '{sql_table.sql}' from a DataFrame with {df.shape[0]} row(s)"
      )
      start_time = time.time()
      with self._adapter.connection_named(
          "MetricFlow_create_from_dataframe"
      ):
          # Create table
          # update dtypes to convert None to NA in boolean columns.
          # This mirrors the SQLAlchemy schema detection logic in pandas.io.sql
          df = df.convert_dtypes()
          columns = df.columns
  
          columns_to_insert = []
          for i in range(len(df.columns)):
              # Format as "column_name column_type"
              columns_to_insert.append(
                  f"{columns[i]} {self._get_type_from_pandas_dtype(str(df[columns[i]].dtype).lower())}"
              )
          self._adapter.execute(
              f"CREATE TABLE IF NOT EXISTS {sql_table.sql} ({', '.join(columns_to_insert)})",
              auto_begin=True,
              fetch=False,
          )
          self._adapter.commit_if_has_connection()
  
          # Insert rows
          values = []
          for row in df.itertuples(index=False, name=None):
              cells = []
              for cell in row:
                  if pd.isnull(cell):
                      # use null keyword instead of isNA/None/etc.
                      cells.append("null")
                  elif type(cell) in [str, pd.Timestamp]:
                      # Wrap cell in quotes & escape existing single quotes
                      escaped_cell = self._quote_escape_value(str(cell))
                      # Trino requires timestamp literals to be wrapped in a timestamp() function.
                      # There is probably a better way to handle this.
                      if (
                          self.sql_engine_type is SqlEngine.TRINO
                          and type(cell) is pd.Timestamp
                      ):
                          cells.append(f"timestamp '{escaped_cell}'")
                      else:
                          cells.append(f"'{escaped_cell}'")
                  else:
                      cells.append(str(cell))
  
              values.append(f"({', '.join(cells)})")
              if chunk_size and len(values) == chunk_size:
                  value_string = ",\n".join(values)
                  self._adapter.execute(
                      f"INSERT INTO {sql_table.sql} VALUES {value_string}",
                      auto_begin=True,
                      fetch=False,
                  )
                  values = []
          if values:
              value_string = ",\n".join(values)
              self._adapter.execute(
                  f"INSERT INTO {sql_table.sql} VALUES {value_string}",
                  auto_begin=True,
                  fetch=False,
              )
          # Commit all insert transaction at once
          self._adapter.commit_if_has_connection()
  
      logger.info(
          f"Created table '{sql_table.sql}' from a DataFrame in {time.time() - start_time:.2f}s"
      )
  
file_path: metricflow/metricflow/test/fixtures/sql_clients/adapter_backed_ddl_client.py
imports: from __future__ import annotations
  import logging
  import time
  from typing import Optional
  import pandas
  from metricflow.cli.dbt_connectors.adapter_backed_client import AdapterBackedSqlClient
  from metricflow.dataflow.sql_table import SqlTable
  from metricflow.protocols.sql_client import SqlEngine
label_desc: Use image node_id 1 for calling the AdapterBackedDDLSqlClient obj's underlying member method code with example usage: obj.create_table_from_dataframe(sql_table, df, chunk_size) without return types