sql Alchemy 模型继承

Question

我使用Python和SQLAlchemy 2.0.23。我的目标是基于这些模型在 postgre 数据库中动态创建 SQLAlchemy 模型和表。我有几个国家/地区，这些表对于所有国家/地区来说都是类似的表，因此我使用动态模型类，为每个国家/地区创建单独的模型

from sqlalchemy import Column, DateTime, Float, BigInteger, String, Boolean, ForeignKey, Sequence
from sqlalchemy.ext.declarative import declared_attr, declarative_base


Base = declarative_base()

SCHEMA = 'schema'

class ModelBase(Base):
    __abstract__ = True
    __table_args__ = {'schema': SCHEMA}

class Model(ModelBase):
    __abstract__ = True
    
    id = Column(BigInteger, primary_key=True)
    provider_id = Column(BigInteger)
    card_bin = Column(String(10))
    bank = Column(String(100))
    card = Column(String(50))
    card_type = Column(String(10))


class DynamicModel:
    @staticmethod
    def create(country_code: str):
        class_name = f"model{country_code.upper()}"
        table_name = f"model_table{country_code}"
        id_seq = Sequence(f'{table_name}_id_seq', schema=SCHEMA)
        class_attrs = {
            '__tablename__': table_name,
            'id': Column(BigInteger, id_seq, primary_key=True, server_default=id_seq.next_value()),
            'provider_id': Column(BigInteger, ForeignKey(f'{SCHEMA}.provider__table_{country_code}.id'))
        }
        for key, value in Model.__dict__.items():
            if isinstance(value, Column) and key not in class_attrs:
                class_attrs[key] = value

        return type(class_name, (Model,), class_attrs)

我无法克服一个问题 - 我希望列 id 和provider_id 成为创建的表中的第一个，因为它是在 Model 类中定义的，但它们始终是表中的最后一个。

我尝试了各种解决方案，但没有任何帮助让列按正确的顺序

Answer 1

import os

from sqlalchemy import (
    Column,
    String,
    BigInteger,
    create_engine,
    ForeignKey,
    Sequence,
)
from sqlalchemy.orm import (
    declarative_base,
)
from sqlalchemy.schema import (
    Table,
)
from sqlalchemy import event
from sqlalchemy.schema import CreateSchema


def get_engine(env):
    PG_URI = f"postgresql+psycopg://{env['DB_USER']}:{env['DB_PASSWORD']}@{env['DB_HOST']}:{env['DB_PORT']}/{env['DB_NAME']}"
    return create_engine(PG_URI,
                         echo=True,
    )

Base = declarative_base()

SCHEMA = 'our_schema'


event.listen(Base.metadata, 'before_create', CreateSchema(SCHEMA))


class ModelBase(Base):
    __abstract__ = True
    __table_args__ = {'schema': SCHEMA}

class Model(ModelBase):
    __abstract__ = True

    id = Column(BigInteger, primary_key=True)
    provider_id = Column(BigInteger)
    card_bin = Column(String(10))
    bank = Column(String(100))
    card = Column(String(50))
    card_type = Column(String(10))


class DynamicModel:
    @staticmethod
    def create(country_code: str):
        class_name = f"model{country_code.upper()}"
        table_name = f"model_table{country_code}"

        provider_t = Table(
            f"provider__table_{country_code}",
            Base.metadata,
            Column('id', BigInteger, primary_key=True), schema=SCHEMA)

        id_seq = Sequence(f'{table_name}_id_seq', schema=SCHEMA)
        class_attrs = {
            '__tablename__': table_name,
            'id': Column(BigInteger, id_seq, primary_key=True, server_default=id_seq.next_value()),
            'provider_id': Column(BigInteger, ForeignKey(f'{SCHEMA}.provider__table_{country_code}.id'))
        }
        for key, value in Model.__dict__.items():
            print(f"{key}")
            if isinstance(value, Column) and key not in class_attrs:
                class_attrs[key] = value

        print(class_attrs.keys())
        return provider_t, id_seq, type(class_name, (Model,), class_attrs)


country_classes = dict([(k, DynamicModel.create(k)) for k in ('us',)])


print (country_classes['us'][2].__dict__.keys())
def main():
    engine = get_engine(os.environ)

    with engine.begin() as conn:
        Base.metadata.create_all(conn)


if __name__ == '__main__':
    main()

这会生成一个像这样的

CREATE TABLE

：

CREATE TABLE our_schema.model_tableus (
    id BIGINT DEFAULT nextval('our_schema.model_tableus_id_seq') NOT NULL, 
    provider_id BIGINT, 
    card_bin VARCHAR(10), 
    bank VARCHAR(100), 
    card VARCHAR(50), 
    card_type VARCHAR(10), 
    PRIMARY KEY (id), 
    FOREIGN KEY(provider_id) REFERENCES our_schema.provider__table_us (id)
)