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

Question

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

例如，我有下面的pydantic模型，它表示对Movie数据库的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

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)

Answer 2

使用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.