Azure Data Tables Python SDK

azure-data-tables-py
分类编程
作者Agentic Awesome Skills 社区
许可MIT
评分4.60/5
使用6.4K

Azure Tables SDK for Python

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

安装

bash
pip install azure-data-tables azure-identity

环境变量

bash
# 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

身份验证

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

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


服务客户端（管理表）

service_client = TableServiceClient(endpoint=endpoint, credential=credential)

表客户端（操作实体）

table_client = TableClient(endpoint=endpoint, table_name="mytable", credential=credential)

客户端类型

| 客户端 | 用途 |
|--------|---------|
| TableServiceClient | 创建/删除表，列出所有表 |
| TableClient | 实体 CRUD，查询 |

表操作

python
# 创建表
service_client.create_table("mytable")

如果不存在则创建

service_client.create_table_if_not_exists("mytable")

删除表

service_client.delete_table("mytable")

列出所有表

for table in service_client.list_tables():
    print(table.name)

获取表客户端

table_client = service_client.get_table_client("mytable")

实体操作

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

创建实体

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

创建（如果已存在则失败）

table_client.create_entity(entity=entity)

Upsert（创建或替换）

table_client.upsert_entity(entity=entity)

获取实体

python
# 通过键获取（速度最快）
entity = table_client.get_entity(
    partition_key="sales",
    row_key="order-001"
)
print(f"Product: {entity['product']}")

更新实体

python
# 替换整个实体
entity["quantity"] = 10
table_client.update_entity(entity=entity, mode="replace")

合并（仅更新特定字段）

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

删除实体

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

查询实体

分区内查询

python
# 按分区查询（高效）
entities = table_client.query_entities(
    query_filter="PartitionKey eq 'sales'"
)
for entity in entities:
    print(entity)

使用过滤器查询

python
# 按属性过滤
entities = table_client.query_entities(
    query_filter="PartitionKey eq 'sales' and quantity gt 3"
)

使用参数（更安全）

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

选择特定属性

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

列出所有实体

python

列出所有实体（跨分区 - 请谨慎使用）

for entity in table_client.list_entities(): print(entity)
code
## 批量操作
python from azure.data.tables import TableTransactionError

批量操作（仅限同一分区！）

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}")

code
## 异步客户端
python
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:
# 创建
await client.create_entity(entity={
"PartitionKey": "async",
"RowKey": "1",
"data": "test"
})

# 查询
async for entity in client.query_entities("PartitionKey eq 'async'"):
print(entity)

import asyncio
asyncio.run(table_operations())
``

数据类型

| Python 类型 | Table Storage 类型 |
|-------------|-------------------|
| str | String |
| int | Int64 |
| float | Double |
| bool | Boolean |
| datetime | DateTime |
| bytes | Binary |
| UUID | Guid |

最佳实践

1. 设计分区键：根据查询模式设计，确保数据均匀分布。
2. 分区内查询：尽可能在分区内进行查询（跨分区查询成本较高）。
3. 使用批量操作：针对同一分区中的多个实体使用批量操作。
4. 使用 upsert_entity`：实现幂等写入。
5. 使用参数化查询：防止注入攻击。
6. 保持实体精简：单个实体最大 1MB。
7. 使用异步客户端：适用于高吞吐量场景。

适用场景

本技能适用于执行概览中所描述的工作流或操作。

局限性

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