azure-data-tables-py

Azure Tables SDK for Python

用于结构化数据的 NoSQL 键值存储（Azure Storage Tables 或 Cosmos DB Table API）。

安装

pip install azure-data-tables azure-identity

环境变量

# Azure Storage Tables
AZURE_STORAGE_ACCOUNT_URL=https://<account>.table.core.windows.net

# Cosmos DB Table API
COSMOS_TABLE_ENDPOINT=https://<account>.table.cosmos.azure.com

身份验证

from azure.identity import DefaultAzureCredential
from azure.data.tables import TableServiceClient, TableClient

credential = DefaultAzureCredential()
endpoint = "https://<account>.table.core.windows.net"

# Service client (manage tables)
service_client = TableServiceClient(endpoint=endpoint, credential=credential)

# Table client (work with entities)
table_client = TableClient(endpoint=endpoint, table_name="mytable", credential=credential)

客户端类型

客户端	用途
TableServiceClient	创建/删除表、列出表
TableClient	实体 CRUD、查询

这些片段彼此独立。请确认账户、表、分区、行以及写入/删除的确切范围；保护现有实体，并在并发更改时使用 ETag/匹配条件；重试前先核实超时操作的结果。Storage Table 与 Cosmos Table 的身份验证、功能和事务限制各不相同；请核实所选终端节点，不要假定两者适用相同的身份配置。请限制查询、页面和字节数，并且不要记录实体内容。发生故障时，也要关闭异步凭据和客户端。货币的浮点数示例仅供说明；处理金额时，请选择精确的资产单位。

表操作

# Create table
service_client.create_table("mytable")

# Create if not exists
service_client.create_table_if_not_exists("mytable")

# Delete table
service_client.delete_table("mytable")

# List tables
for table in service_client.list_tables():
    print(table.name)

# Get table client
table_client = service_client.get_table_client("mytable")

实体操作

重要：每个实体都必须包含 PartitionKey 和 RowKey（两者共同构成唯一 ID）。

创建实体

entity = {
    "PartitionKey": "sales",
    "RowKey": "order-001",
    "product": "Widget",
    "quantity": 5,
    "price": 9.99,
    "shipped": False
}

# Create (fails if exists)
table_client.create_entity(entity=entity)

# Upsert explicitly replaces; omitted mode defaults can merge instead.
from azure.data.tables import UpdateMode
table_client.upsert_entity(entity=entity, mode=UpdateMode.REPLACE)

获取实体

# Get by key (fastest)
entity = table_client.get_entity(
    partition_key="sales",
    row_key="order-001"
)
print(f"Product: {entity['product']}")

更新实体

# Replace entire entity
entity["quantity"] = 10
table_client.update_entity(entity=entity, mode="replace")

# Merge (update specific fields only)
update = {
    "PartitionKey": "sales",
    "RowKey": "order-001",
    "shipped": True
}
table_client.update_entity(entity=update, mode="merge")

删除实体

table_client.delete_entity(
    partition_key="sales",
    row_key="order-001"
)

查询实体

在分区内查询

# Query by partition (efficient)
entities = table_client.query_entities(
    query_filter="PartitionKey eq 'sales'"
)
for entity in entities:
    print(entity["RowKey"])  # Do not dump property payloads

使用筛选条件查询

# Filter by properties
entities = table_client.query_entities(
    query_filter="PartitionKey eq 'sales' and quantity gt 3"
)

# With parameters (safer)
entities = table_client.query_entities(
    query_filter="PartitionKey eq @pk and price lt @max_price",
    parameters={"pk": "sales", "max_price": 50.0}
)

选择特定属性

entities = table_client.query_entities(
    query_filter="PartitionKey eq 'sales'",
    select=["RowKey", "product", "price"]
)

列出所有实体

# List all (cross-partition - use sparingly)
for entity in table_client.list_entities():
    print(entity["RowKey"])  # Do not dump property payloads

批处理操作

from azure.data.tables import TableTransactionError

# Batch operations (same partition only!)
operations = [
    ("create", {"PartitionKey": "batch", "RowKey": "1", "data": "first"}),
    ("create", {"PartitionKey": "batch", "RowKey": "2", "data": "second"}),
    ("upsert", {"PartitionKey": "batch", "RowKey": "3", "data": "third"}),
]

try:
    table_client.submit_transaction(operations)
except TableTransactionError as e:
    print(f"Transaction failed: {e}")

异步客户端

from azure.data.tables.aio import TableServiceClient, TableClient
from azure.identity.aio import DefaultAzureCredential

async def table_operations():
    credential = DefaultAzureCredential()
    
    async with TableClient(
        endpoint="https://<account>.table.core.windows.net",
        table_name="mytable",
        credential=credential
    ) as client:
        # Create
        await client.create_entity(entity={
            "PartitionKey": "async",
            "RowKey": "1",
            "data": "test"
        })
        
        # Query
        async for entity in client.query_entities("PartitionKey eq 'async'"):
            print(entity["RowKey"])  # Do not dump property payloads

import asyncio
asyncio.run(table_operations())

数据类型

Python 类型	表存储类型
str	String
int	Int64
float	Double
bool	Boolean
datetime	DateTime
bytes	Binary
UUID	Guid

最佳实践

1. 设计分区键，以适应查询模式并实现均匀分布
2. 尽可能在分区内查询（跨分区查询开销较大）
3. 对同一分区中的多个实体使用批处理操作
4. 审慎选择 upsert 模式；幂等性还取决于业务影响和并发情况
5. 使用参数化查询以防止注入
6. 保持实体较小——每个实体最大 1 MB
7. 在高吞吐量场景中使用异步客户端

适用场景

当任务涉及执行概述中描述的工作流或操作时，可使用此技能。

限制

• 仅当任务明确符合上述范围时，才使用此技能。
• 不要将输出视为环境特定验证、测试或专家审核的替代品。
• 如果缺少必要的输入、权限、安全边界或成功标准，请暂停并请求澄清。