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

Question

我想弄清楚是否有一种无需手动设置即可将数据从对象 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.

这样的事情可能吗？

Answer 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.