如何将Pydantic嵌套模型转为字典写入DynamoDB并保留原始嵌套结构？

我用Pydantic定义了嵌套模型，读取数据后打印的结构如下：

{
   "Name": "jhon",
   "Address": {
      "HouseNumber": 20,
      "current": true,
      "street": "easy street"
   }
}

但写入DynamoDB后，Address字段被保存为带类型标识的结构：

Name | Address
jhon | {"HouseNumber":{"N":"20"},"current": {"BOOL": true}, ...}

我希望Address字段直接保存为{"HouseNumber": "20", "current": true, "street": "easy street"}这样的原始嵌套字典结构，请问该怎么做？

我的写入代码：

import boto3
import json

dynamo_resource = boto3.Session().resource('dynamodb', region_name='us-west-2')
person_table = dynamo_resource.Table('person')

with person_table.batch_writer() as batch:
    for i in person_list:
        # 转换为JSON再解析回字典
        json_string = i.model_dump_json()
        record = json.loads(json_string)
        batch.put_item(Item=record)

我的Pydantic模型定义：

from pydantic import BaseModel
from typing import Optional

class Address(BaseModel):
    HouseNumber: int
    current: bool
    street: str

class Person(BaseModel):
    Name: str
    Address: Optional[Address]

解决方案 #

要让Address字段以原始嵌套字典格式保存到DynamoDB，有两种可行方案：

方案1：将Address序列化为JSON字符串存储 #

直接把Address字段转为JSON字符串，存入DynamoDB的字符串类型字段，读取时再反序列化回字典，这种方式简单直接，适合不需要对子字段做查询的场景：

修改写入代码：

import boto3

dynamo_resource = boto3.Session().resource('dynamodb', region_name='us-west-2')
person_table = dynamo_resource.Table('person')

with person_table.batch_writer() as batch:
    for i in person_list:
        record = i.model_dump()
        # 将Address对象转为JSON字符串
        if record.get('Address'):
            record['Address'] = record['Address'].model_dump_json()
        batch.put_item(Item=record)

读取时还原Address结构：

from pydantic import TypeAdapter

response = person_table.get_item(Key={'Name': 'jhon'})
item = response['Item']
# 将JSON字符串转回Address对象或字典
item['Address'] = TypeAdapter(Address).json_loads(item['Address'])

方案2：使用boto3客户端构造原生Map类型 #

如果需要基于Address的子字段做查询、排序等操作，可将Address存储为DynamoDB原生的Map类型，通过boto3客户端手动构造字段类型：

import boto3

dynamo_client = boto3.Session().client('dynamodb', region_name='us-west-2')

for i in person_list:
    record = i.model_dump()
    # 构造符合DynamoDB格式的Item
    dynamodb_item = {
        'Name': {'S': record['Name']},
    }
    if record.get('Address'):
        dynamodb_item['Address'] = {
            'M': {
                'HouseNumber': {'N': str(record['Address']['HouseNumber'])},
                'current': {'BOOL': record['Address']['current']},
                'street': {'S': record['Address']['street']}
            }
        }
    dynamo_client.put_item(TableName='person', Item=dynamodb_item)

这种方式下，DynamoDB控制台会将Address显示为正常嵌套字典；通过boto3 resource读取时，会自动转换为Python原生字典，无需手动处理类型标识。

内容的提问来源于stack exchange，提问作者kishi