动态表创建 SQLAlchemy ORM

Question

我是 SQLAlchemy ORM 的新手。我正在尝试构建一个 AWS S3 摄取程序，该程序将通过 ORM 将任何 CSV 文件从 S3 存储桶摄取到 Postgres。我正在尝试读取 CSV 文件的第一行并将结果存储到列表（columns_names）中。代码报错：

无法为映射表组装任何主键列。

只有在声明 PRIMARY KEY 列后才会在数据库中创建表。通过 ORM 创建表时必须使用主键吗？另外，如何从列表 columns_names 动态创建列？

这是我的代码：

import boto
import boto3
import botocore
import os
from datetime import datetime
import s3fs
import pandas as pd 
import configparser
import re
from sqlalchemy import create_engine
from sqlalchemy import MetaData, Table, Column, Integer, String
from sqlalchemy.orm.session import sessionmaker
from sqlalchemy.orm import relationship
from sqlalchemy.ext.declarative import declarative_base


config = configparser.ConfigParser(allow_no_value=True)
config.read('IngestionConfig.config')
table_name = config.get('db-settings','table_name')
S3Bucket = config.get('AWS-settings','BucketName')
S3Key = config.get('AWS-settings','filename')
s3_client = boto3.client('s3')
response = s3_client.get_object(Bucket = S3Bucket, Key= S3Key)
file = response["Body"]
filedata = file.read() 
contents = filedata.decode('utf-8')
first_line = contents.split('\n',1)[0]
col_names = re.sub(r"\s+", '_', first_line).replace('"', r'')
columns_names= []
columns_names = col_names.split(',') 

postgresql_db = create_engine('postgresql://ayan.putatunda@localhost/postgres',echo = True)

Base = declarative_base()

class test(Base):
    __tablename__ = table_name
    for name in columns_names:
        name = Column(String)

Base.metadata.create_all(postgresql_db)

Answer 1

SQLAlchemy ORM 确实需要一个主键，因为它的设计需要一种方法来识别对应于对象的行，所以在 ORM 中不能使用没有主键的表。

您可以通过首先使用您的表格信息创建一个字典来动态创建表格：

col_lst = ['col_1', 'col_2', 'col_3']

attr_dict = {'__tablename__': 'myTableName'}
for col in col_lst:
    attr_dict[col] = Column(Integer)

接下来使用 type 函数使用 SQLAlchemy 的 declarative_base 方法创建表 Class：

Base = declarative_base()

MyTableClass = type('MyTableClass', (Base,), attr_dict)