如何基于数据类建立 ORM 模型

Question

我有一堆数据类，想将它们用作我的数据库的 ORM 模型。 为了达到我的目标，我制作了装饰器@sa_orm，它将每个装饰的数据类映射到元数据。

问题：

1. 有些字段是数据类。
2. 一些字段是数据类对象的列表

OneToOne（将 child_id 列添加到父表）关系解决第一个问题，但要解决第二个问题，我需要设置 OneToMany 关系（将 parent_id 添加到子表）。

这里是简化的代码：

from dataclasses import dataclass, field, is_dataclass
from sqlalchemy import Table, MetaData, Column, Integer, String, create_engine, ForeignKey
from sqlalchemy.orm import sessionmaker, mapper, relationship
from sqlalchemy.ext.declarative import declarative_base
from typing import Optional, List
metadata = MetaData()
Base = declarative_base(metadata=metadata)

def sa_orm(metadata):
    def decorator(cls):
        columns = [Column('id', Integer, primary_key=True, autoincrement=True)]
        children = {}
        for field_name, meta in cls.__dataclass_fields__.items():
            try:
                field_class = meta.type.__args__[0]  # If Optional or List
            except AttributeError:
                field_class = meta.type  # If not Optional or List
            if is_dataclass(field_class):
                field_class_name = field_class.__name__
                columns.append(Column(f"{field_class_name}_id".lower(), Integer, ForeignKey(f'{field_class_name}.id'.lower())))
                children[field_name] = relationship(field_class)
            else:
                columns.append(Column(field_name, String))
        table = Table(cls.__name__.lower(), metadata,
                      *columns)
        mapper(cls, table, properties=children)

        return cls
    return decorator


@sa_orm(metadata)
@dataclass
class Profession:
    name: Optional[str] = field(default_factory=str)

@sa_orm(metadata)
@dataclass
class Person:
    profession: Optional[Profession] = field(default=None)
    name: str = field(default_factory=str)

@sa_orm(metadata)
@dataclass
class Company:
    employee: List[Person] = field(default_factory=list)
    name: str = field(default_factory=str)

engine = create_engine(f'postgresql://user:pass@localhost:5432', echo=True)

metadata.create_all(engine)

session = sessionmaker(bind=engine)()

developer = Profession(name="developer")
alex = Person(name="Alexander", profession=developer)
peter = Person(name="Peter")
google = Company(name="Google", employee=[alex, peter])  # OneToMany

session.add(alex)  # this works - insert alex to db, insert developer to db
session.add(peter)  # also works - insert peter to db
session.commit()

session.add(google) # this fails
session.close()

有没有办法从 Company 装饰器中将“company_id”之类的列设置为人员表，或者在不更改数据类的情况下链接这些表？

我有数百个这样的数据类分散在几十个文件中，也许有更好的基于数据类设置数据库的解决方案？

Answer 1

SqlAlchemy 似乎支持通过Imperative 或Declarative 方法使用数据类。

由于您似乎在使用 declarative_base，以下是声明性文档中的示例，可以帮助您实现所需的目标：

from __future__ import annotations

from dataclasses import dataclass
from dataclasses import field
from typing import List
from typing import Optional

from sqlalchemy import Column
from sqlalchemy import ForeignKey
from sqlalchemy import Integer
from sqlalchemy import String
from sqlalchemy import Table
from sqlalchemy.orm import registry
from sqlalchemy.orm import relationship

mapper_registry = registry()


@mapper_registry.mapped
@dataclass
class User:
    __table__ = Table(
        "user",
        mapper_registry.metadata,
        Column("id", Integer, primary_key=True),
        Column("name", String(50)),
        Column("fullname", String(50)),
        Column("nickname", String(12)),
    )
    id: int = field(init=False)
    name: Optional[str] = None
    fullname: Optional[str] = None
    nickname: Optional[str] = None
    addresses: List[Address] = field(default_factory=list)

    __mapper_args__ = {   # type: ignore
        "properties" : {
            "addresses": relationship("Address")
        }
    }

@mapper_registry.mapped
@dataclass
class Address:
    __table__ = Table(
        "address",
        mapper_registry.metadata,
        Column("id", Integer, primary_key=True),
        Column("user_id", Integer, ForeignKey("user.id")),
        Column("email_address", String(50)),
    )
    id: int = field(init=False)
    user_id: int = field(init=False)
    email_address: Optional[str] = None