SQLAlchemy - 从 yaml 或字典创建表？

Question

有没有办法从 yaml 文件中指定的字典创建动态表？我在 yaml 文件中定义了很多 ETL 配置，所以我很好奇我是否也可以向其中添加表创建方面，这样我就不必在单独的目录中修改单独的 .sql 文件。

database:
  table: 'schema.fact_stuff'
  create_columns: [
    {}
  ] #not sure how this section should be

我在 stackoverflow 上找到了一个解决方案，它将一些列表压缩在一起，这是类似的，但我更愿意明确定义每一列。

{'column_name': 'id', 'column_type': Integer, 'primary_key': False, 'nullable': True}

我最终让它与这个一起工作：

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)

但是后来我转移到让熊猫确定 dtypes 而不是在 yaml 文件中定义它们。这将使用 jinja2 模板创建 sql，然后我遍历所有数据源以创建 DDL。

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)

Answer 1

今天发布的SQLAthanor (v.0.3.0) 正是支持这一点。使用SQLAthanorTable ，您可以使用以下代码以编程方式生成 SQLAlchemy对象（假设metadata包含您的 SQLAlchemyMetaData对象）：

from sqlathanor import Table

my_table = Table.from_yaml('yaml_file.yaml', 
                           'my_table_name', 
                           metadata, 
                           primary_key = 'id')

ETA：请注意，您还可以使用、和创建Table对象。Table.from_json()Table.from_dict()Table.from_csv()

以下是其工作原理的文档（通常）：https://sqlathanor.readthedocs.io/en/latest/using.html#generate-sqlalchemy-tables-from-serialized-data

这是特定方法的文档的链接Table.from_yaml()：https://sqlathanor.readthedocs.io/en/latest/api.html#sqlathanor.schema.Table.from_yaml

Table（我建议查看方法文档 - 它涉及以编程方式从序列化数据构造对象的一些“陷阱” ）

预计到达时间：

基本上，编程生成的工作方式Table是 SQLAthanor：

1. 首先将序列化字符串（或文件）转换为 Python dict。对于 YAML，默认的反序列化器是PyYAML。对于 JSON，默认的反序列化器是simplejson（两个默认的反序列化器都可以使用deserialize_function参数覆盖）。

2. dict生成Python 后，SQLAthanor 会读取其中的每个键dict来确定列名称。它读取每个键的值，并根据值的数据类型尝试“猜测”SQLAlchemy 数据类型。

3. 根据在步骤 2 中找到的内容，它会创建一个Table包含对象的Column对象，其中每个Column对象对应于反序列化的dict.

如果您需要对每个进行更精确的控制Column，您可以：

• 使用参数覆盖其 SQLAlchemy 数据类型type_mapping（type_mapping接收一个dict，其中顶级键对应于列名，每个值都是要应用于的数据类型Column）
• Column使用参数column_kwargs(column_kwargs接收一个dict，其中顶级键对应于列名，并且每个值都是一个dictwith 关键字参数，将提供给该列的构造函数。

默认情况下，Table.from_<format>()不支持嵌套数据结构。默认情况下，skip_nested设置为True，这意味着反序列化中dict包含嵌套对象（可迭代对象或dict）的键将被跳过（即不会收到相应的Column）。如果您Table需要存储嵌套数据，您可以设置skip_nested为False并激活default_to_str为True。这会将嵌套数据（可迭代或dict对象）转换为字符串，从而将它们保留在Text列中（除非被 覆盖type_mapping）。

Table.from_dict()例子

以下是dict可以提供给 的示例Table.from_dict()：

sample_dict = {
    'id': 123,
    'some_column_name': 'Some Column Value',
    'created_datetime': datetime.utcnow()
}

my_table = Table.from_dict(sample_dict, 
                           'my_table', 
                           metadata, 
                           primary_key = 'id')

当提供给Table.from_dict()它时dict​​，将生成一个Table具有数据库表名称的对象my_table，其中包含三列：

• id其类型设置Integer为表的主键
• some_column_name其中将有类型Text
• created_datetime其中将有类型DateTime

Table.from_yaml()例子

以下是相同的示例，但使用可提供给的 YAML 字符串/文档Table.from_yaml()：

sample_yaml = """
    id: 123
    some_column_name: Test Value
    created_timestamp: 2018-01-01T01:23:45.67890
"""

my_table = Table.from_yaml(sample_yaml, 
                           'my_table', 
                           metadata, 
                           primary_key = 'id')

当提供给Table.from_yaml()它时​​，将首先反序列化为sample_yaml就像dict前面的示例中一样，然后生成一个具有包含三列的Table数据库表名称的对象：my_table

• id其类型设置Integer为表的主键
• some_column_name其中将有类型Text
• created_datetime其中将有类型DateTime

希望这可以帮助！