SQLAlchemy 创建动态表和列

Question

我正在尝试根据我检索的数据动态创建数据库表和列。 我检索数据库列表、列名称和属性列表，例如列类型、primary_key / 唯一、可为空以及其他元数据。 我正在尝试使用这些信息来动态创建表格，并且一直在使用论坛帖子来更好地了解如何实现这一点。所以我想根据我检索到的信息创建表 - 数据库和列信息（列名和列类型、主键和可为空的信息。检索到的信息可能每天或每周更改。 论坛帖子 #1 - Sqlalchemy dynamically create table and mapped class

postgresql_db = engine(...)

post_meta = sql.MetaData(bind=postgresql_db.engine)

post_meta.reflect(schema='customers')

connection = postgresql_db.engine.connect()

col_names = ['id', 'fname', 'lname', 'age']
ctype = ['Integer', 'String', 'String', 'Integer']
pk = ['True', 'False', 'False', 'False']
nulls = ['No', 'No', 'No', 'No']

class test(object):

     test = Table('customers', post_meta,
              *(Column(col, ctype, primary_key=pk, nullable=nulls)
           for col, ctype, pk, nulls in zip(col_names, ctype, pk, nulls))

test.create()

有一个错误信息： AttributeError: 'list' object has no attribute _set_parent_with_dispatch 似乎无法确定这个错误到底指的是什么。

追溯：

Traceback (most recent call last):
  File "C:/Users/xxx/db.py", line 247, in <module>
    main()
  File "C:/Users/xxx/db.py", line 168, in main
    for col, ctype, pk, nulls in zip(col_names, ctype, pk, nulls)
  File "C:/Users/xxx/apidb.py", line 168, in <genexpr>
    for col, ctype, pk, nulls in zip(col_names, ctype, pk, nulls)
  File "C:\Python27\lib\site-packages\sqlalchemy\sql\schema.py", line 1234, in __init__
    self._init_items(*args)
  File "C:\Python27\lib\site-packages\sqlalchemy\sql\schema.py", line 79, in _init_items
    item._set_parent_with_dispatch(self)
AttributeError: 'list' object has no attribute '_set_parent_with_dispatch'

任何想法我做错了什么？

Answer 1

这里有很多不正确的地方。

Column 初始化器中的nullable 参数应该具有类型bool，但您正在尝试传递str 对象nulls，pk 和primary_key 参数也是如此。

此外，您试图在理解中覆盖名称ctype、pk、nulls，这是不正确的并引发给定异常。 您应该在理解中重命名从zip 生成的对象。

SQLAlchemy 无法识别字符串 'Integer'、'String'，它们不是有效的 Column 类型。

如果要反映名为'customers' 的特定表，应该使用参数only，而不是schema，它应该是名称中的list。

你也不需要test类。

您的代码可能看起来像

from sqlalchemy import MetaData, Table, Column, Integer, String

postgresql_db = engine(...)

post_meta = MetaData(bind=postgresql_db.engine)

post_meta.reflect(only=['customers'])

connection = postgresql_db.engine.connect()

columns_names = ['id', 'fname', 'lname', 'age']
columns_types = [Integer, String, String, Integer]
primary_key_flags = [True, False, False, False]
nullable_flags = [False, False, False, False]

test = Table('customers', post_meta,
             *(Column(column_name, column_type,
                      primary_key=primary_key_flag,
                      nullable=nullable_flag)
               for column_name,
                   column_type,
                   primary_key_flag,
                   nullable_flag in zip(columns_names,
                                        columns_types,
                                        primary_key_flags,
                                        nullable_flags)))

test.create()

最后，如果你做post_meta.reflect(only=['customers'])并且它有效，给定的表可以简单地通过

test = post_meta.tables['customers']

无需从头开始构建。