title: SQLAlchemy - create table from yaml or a dictionary?
body: <p>Is there a way to create a dynamic table from a dictionary specified in a yaml file? I define a lot of the ETL config in the yaml file, so I was curious if I could also add the table creation aspect to it as well so I do not have to modify a separate .sql file in a separate directory.</p>
  
  <pre><code>database:
    table: 'schema.fact_stuff'
    create_columns: [
      {}
    ] #not sure how this section should be
  </code></pre>
  
  <p>I found a solution on stackoverflow which zipped some lists together which is something similar, but I would prefer to define each column explicitly.</p>
  
  <pre><code>{'column_name': 'id', 'column_type': Integer, 'primary_key': False, 'nullable': True}
  </code></pre>
  
  <p>I ended up getting it to work with this:</p>
  
  <pre><code>from sqlalchemy.types import (Integer, NUMERIC, TEXT, BOOLEAN, TIMESTAMP, DATE)
  
  sql_types = {'integer': Integer,
          'numeric': NUMERIC,
          'text': TEXT,
          'date': DATE,
          'timestamp': TIMESTAMP(timezone=False),
          'timestamptz': TIMESTAMP(timezone=True)}
  
  exclude_list = ['original_name']
  table_dict = [{k: v for k, v in d.items() if k not in exclude_list} for d in c[variable]['load']['columns']]
  for column in table_dict:
      for key, val in column.copy().items():
          if key == 'type_':
              column[key] = sql_types[val]
          elif key == 'default':
              column[key] = dt.datetime.utcnow
  
  metadata = sa.MetaData(schema=c[variable]['load']['schema'])
  metadata.reflect(bind=engine, autoload=True)
  fact = sa.Table(c[variable]['load']['table'], metadata, extend_existing=True,
          *(sa.Column(**kwargs) for kwargs in table_dict))
  fact.create_all(engine, checkfirst=True)
  </code></pre>
  
  <p>But then I moved over to having pandas determine the dtypes instead of defining them in a yaml file. This creates the sql with a jinja2 template and I loop through all of my data sources to create the DDL.</p>
  
  <pre><code>def pandas_to_postgres(df):
      dtype_dict = {
        'i': 'integer',
        'O': 'text',
        'f': 'real',
        'b': 'boolean',
        'datetime64[ns]': 'timestamp',
        'datetime64[ns, UTC]': 'timestampz',
      }
      column_list = []
      column_dict = {}
      for k, v in df.dtypes.items():
          column_dict['name'] = k
          column_dict['dtype'] = dtype_dict.get(v.kind, 'text')
          column_list.append(column_dict.copy())
      return column_list
  
  
  def generate_create_table(df, schema, table, table_type, columns, constraint, unique_columns):
      """ Returns a dictionary of coefs from training """
      query = Template(
          template
      ).render(
          schema_name=schema,
          table_name=table,
          table_type=table_type,
          columns=columns,
          constraint=constraint,
          constraint_columns=unique_columns
      )
      print(query)
  </code></pre>