使用Python SDK向Cosmos SQL DB创建Item的报错问题咨询

安全阿岳

2026-10-17

问题根因分析

第一个AttributeError: 'str' object has no attribute 'get'报错：你之前调用create_item时错误传入了字符串类型的str(data_dict)，Cosmos SDK要求body参数必须是Python字典类型，不能是字符串；另外你把DataFrame转字典的方式也存在错误，json.dumps(str(df))会把整个DataFrame转成描述性字符串再序列化，无法得到正确的键值对结构。
第二个id字段缺失报错：Cosmos SQL API不会自动生成文档的id字段，该字段是必填项，取值为字符串类型，同一个逻辑分区下id必须唯一。

具体实现代码

import uuid
import pandas as pd
from azure.cosmos import CosmosClient

# 初始化Cosmos客户端（原有配置保持不变）
url = "<my-url>"
key = "my-key"
client = CosmosClient(url, key)
database_name = "My-Database"
container_name = "Table"
database = client.get_database_client(database_name)
container = database.get_container_client(container_name)

# 1. 正确将DataFrame转为字典列表
items = df.to_dict("records")

# 2. 为每条数据补充必填的id字段
for item in items:
    # 也可以用业务唯一字段组合生成id，比如f"{item['userID']}_{item['_timestamp']}"，避免重复
    item["id"] = str(uuid.uuid4())
    # 注意：item必须包含你创建Cosmos容器时指定的分区键字段，否则会触发分区键相关报错

# 3. 写入数据
# 单条写入（适合300条以内的小规模数据）
for item in items:
    container.create_item(body=item)

# 批量写入（更高效率的可选方案）
# with container.batch() as batch:
#     for item in items:
#         batch.upsert_item(item)