如何在 Python 中将对象 A 中的数据合并到对象 B 中？答案

【问题标题】：How to merge data from object A into object B in Python?如何在 Python 中将对象 A 中的数据合并到对象 B 中？
【发布时间】：2022-01-16 15:01:12
【问题描述】：

我试图弄清楚是否有一种程序方法可以将数据从对象 A 合并到对象 B，而无需手动设置。

例如，我有以下 pydantic 模型，它表示对电影数据库的 API 调用结果：

class PersonScraperReply(BaseModel):
    """Represents a Person Scraper Reply"""

    scraper_name: str
    """Name of the scraper used to scrape this data"""

    local_person_id: int
    """Id of person in local database"""

    local_person_name: str
    """name of person in local database"""

    aliases: Optional[list[str]] = None
    """list of strings that represent the person's aliases obtained from scraper"""

    description: Optional[str] = None
    """String description of the person obtained from scraper"""

    date_of_birth: Optional[date] = None
    """Date of birth of the person obtained from scraper"""

    date_of_death: Optional[date] = None
    """Date the person passed away obtained from scraper"""

    gender: Optional[GenderEnum] = None
    """Gender of the person obtained from scraper"""

    homepage: Optional[str] = None
    """Person's official homepage obtained from scraper"""

    place_of_birth: Optional[str] = None
    """Location where the person wsa born obtained from scraper"""

    profile_image_url: Optional[str] = None
    """Url for person's profile image obtained from scraper"""

    additional_images: Optional[list[str]] = None
    """List of urls for additional images for the person obtained from scraper"""

    scrape_status: ScrapeStatus
    """status of scraping. Success or failure"""

我还有一个代表我数据库中的人的 SQLAlchemy 类：

class PersonInDatabase(Base):

    id: int
    """Person Id"""

    name: str
    """Person Name"""
    
    description: str = Column(String)
    """Description of the person"""

    gender: GenderEnum = Column(Enum(GenderEnum), nullable=False, default=GenderEnum.unspecified)
    """Person's gender, 0=unspecified, 1=male, 2=female, 3=non-binary"""

    tmdb_id: int = Column(Integer)
    """Tmdb id"""

    imdb_id: str = Column(String)
    """IMDB id, in the format of nn[alphanumeric id]"""

    place_of_birth: str = Column(String)
    """Place of person's birth"""

    # dates
    date_of_birth: DateTime = Column(DateTime)
    """Date the person was born"""

    date_of_death: DateTime = Column(DateTime)
    """Date the person passed away"""

    date_last_person_scrape: DateTime = Column(DateTime)
    """Date last time the person was scraped"""

我的目标是将我从 API 调用收到的数据合并到数据库对象。当我说合并时，我的意思是分配两个对象中都存在的字段，其余的不做任何事情。大致如下：

person_scrape_reply = PersonScraperReply()
person_in_db = PersonInDatabase()


for field_in_API_name, field_in_API_value in person_scrape_reply.fields: #for field in API response
    if field_in_API_name in person_in_db.field_names and field_in_API_value is not None: #if field exists in PersonInDatabase and the value is not none
        person_in_db.fields[field_in_API_name] = field_in_API_value #assign API response value to field in database class.

这样的事情可能吗？

【问题讨论】：

标签： python pydantic python-dataclasses

【解决方案1】：

使用attrs 包。

from attrs import define, asdict

@define
class PersonScraperReply(BaseModel):
    """Represents a Person Scraper Reply"""

    scraper_name: str
    """Name of the scraper used to scrape this data"""

    local_person_id: int
    """Id of person in local database"""

    local_person_name: str
    """name of person in local database"""

    aliases: Optional[list[str]] = None
    """list of strings that represent the person's aliases obtained from scraper"""

    description: Optional[str] = None
    """String description of the person obtained from scraper"""

    date_of_birth: Optional[date] = None
    """Date of birth of the person obtained from scraper"""

    date_of_death: Optional[date] = None
    """Date the person passed away obtained from scraper"""

    gender: Optional[GenderEnum] = None
    """Gender of the person obtained from scraper"""

    homepage: Optional[str] = None
    """Person's official homepage obtained from scraper"""

    place_of_birth: Optional[str] = None
    """Location where the person wsa born obtained from scraper"""

    profile_image_url: Optional[str] = None
    """Url for person's profile image obtained from scraper"""

    additional_images: Optional[list[str]] = None
    """List of urls for additional images for the person obtained from scraper"""

    scrape_status: ScrapeStatus
    """status of scraping. Success or failure"""

@define
class PersonInDatabase(Base):

    id: int
    """Person Id"""

    name: str
    """Person Name"""
    
    description: str = Column(String)
    """Description of the person"""

    gender: GenderEnum = Column(Enum(GenderEnum), nullable=False, default=GenderEnum.unspecified)
    """Person's gender, 0=unspecified, 1=male, 2=female, 3=non-binary"""

    tmdb_id: int = Column(Integer)
    """Tmdb id"""

    imdb_id: str = Column(String)
    """IMDB id, in the format of nn[alphanumeric id]"""

    place_of_birth: str = Column(String)
    """Place of person's birth"""

    # dates
    date_of_birth: DateTime = Column(DateTime)
    """Date the person was born"""

    date_of_death: DateTime = Column(DateTime)
    """Date the person passed away"""

    date_last_person_scrape: DateTime = Column(DateTime)
    """Date last time the person was scraped"""


person_scrape_reply = PersonScraperReply()
person_in_db = PersonInDatabase()
scrape_asdict = asdict(person_scrape_reply)
db_asdict = asdict(person_in_db)

for field_in_API_name, field_in_API_value in scrape_asdict.items(): #for field in API response
    if field_in_API_name in db_asdict.keys() and field_in_API_value is not None: #if field exists in PersonInDatabase and the value is not none
        setattr(person_in_db, field_in_API_name, field_in_API_value) #assign API response value to field in database class.

【讨论】：

这听起来不错，正是我所需要的，遗憾的是它不能很好地与 SQLAlchemy 配合使用。当我用@define 装饰数据库类时，我得到`assert manager.registry is None AssertionError`
我想通了，请看下面我的回答，感谢您为我指明正确的方向！
@Curtwagner1984 在我看到你的第一条评论后，我试图找到与 sqlalchemy 兼容的东西并找到了一些有趣的包，但我想确保它对你有用，所以我没有编辑我的答案，但我很高兴我帮助找到了解决方案:)

【解决方案2】：

@Daniel 建议的方法（使用 attrs）对我来说是一个错误，我确信它适用于常规类，但它会导致 SQLAlchemy 和 Pydantic 类都出错。

在使用调试器后，我看到 Pydantic 和 SQLAchemy 都提供了一种以字符串格式访问其字段名称的方法。在 SQLAchemy 中，它是 inspect([SQLALCHEMY MAPPED CLASS]).attrs.key，而 Pydantic 只是有一个内置的 dict() 方法。当 pydantic 的一大卖点是它可以将数据类序列化为 JSON 时，我忘记它有点傻。

无论如何，使用这两种方法，这对我有用：

def assing_empty(person_to_assign: Person, scrape_result: PersonScraperReply):
    blacklisted_fields = ["aliases"] #fields to ignore
    person_to_assign_fields = [x.key for x in inspect(person_to_assign).attrs] #SQLAlchemy fields
    scrape_result_fields = [x for x in scrape_result.dict().keys() if x not in blacklisted_fields] #Pydantic fields

    for field_name in scrape_result_fields:
        if field_name in person_to_assign_fields:
            person_to_assign_value = getattr(person_to_assign, field_name)
            scrape_result_value = getattr(scrape_result, field_name)

            if scrape_result_value is not None and person_to_assign_value is None:
                setattr(person_to_assign, field_name, scrape_result_value)

【讨论】：