azure-mgmt-fabric-py

Azure Fabric Management SDK for Python

以编程方式管理 Microsoft Fabric 容量和资源。

安装

pip install azure-mgmt-fabric
pip install azure-identity

环境变量

AZURE_SUBSCRIPTION_ID=<your-subscription-id>
AZURE_RESOURCE_GROUP=<your-resource-group>

身份验证

from azure.identity import DefaultAzureCredential
from azure.mgmt.fabric import FabricMgmtClient
import os

credential = DefaultAzureCredential()
client = FabricMgmtClient(
    credential=credential,
    subscription_id=os.environ["AZURE_SUBSCRIPTION_ID"]
)

创建 Fabric 容量

from azure.mgmt.fabric import FabricMgmtClient
from azure.mgmt.fabric.models import FabricCapacity, FabricCapacityProperties, FabricCapacityAdministration, CapacitySku
from azure.identity import DefaultAzureCredential
import os

credential = DefaultAzureCredential()
client = FabricMgmtClient(
    credential=credential,
    subscription_id=os.environ["AZURE_SUBSCRIPTION_ID"]
)

resource_group = os.environ["AZURE_RESOURCE_GROUP"]
capacity_name = "myfabriccapacity"

capacity = client.fabric_capacities.begin_create_or_update(
    resource_group_name=resource_group,
    capacity_name=capacity_name,
    resource=FabricCapacity(
        location="eastus",
        sku=CapacitySku(
            name="F2",  # Fabric SKU
            tier="Fabric"
        ),
        properties=FabricCapacityProperties(
            administration=FabricCapacityAdministration(
                members=["[email protected]"]
            )
        )
    )
).result()

print(f"Capacity created: {capacity.name}")

获取容量详细信息

capacity = client.fabric_capacities.get(
    resource_group_name=resource_group,
    capacity_name=capacity_name
)

print(f"Capacity: {capacity.name}")
print(f"SKU: {capacity.sku.name}")
print(f"State: {capacity.properties.state}")
print(f"Location: {capacity.location}")

列出资源组中的容量

capacities = client.fabric_capacities.list_by_resource_group(
    resource_group_name=resource_group
)

for capacity in capacities:
    print(f"Capacity: {capacity.name} - SKU: {capacity.sku.name}")

列出订阅中的所有容量

all_capacities = client.fabric_capacities.list_by_subscription()

for capacity in all_capacities:
    print(f"Capacity: {capacity.name} in {capacity.location}")

更新容量

from azure.mgmt.fabric.models import FabricCapacityUpdate, CapacitySku

updated = client.fabric_capacities.begin_update(
    resource_group_name=resource_group,
    capacity_name=capacity_name,
    properties=FabricCapacityUpdate(
        sku=CapacitySku(
            name="F4",  # Scale up
            tier="Fabric"
        ),
        tags={"environment": "production"}
    )
).result()

print(f"Updated SKU: {updated.sku.name}")

暂停容量

评估活动工作负载后，仅暂停已获批准的容量；核实实际计费和承诺条款，不要承诺所有费用都会停止：

client.fabric_capacities.begin_suspend(
    resource_group_name=resource_group,
    capacity_name=capacity_name
).result()

print("Capacity suspended")

恢复容量

恢复已暂停的容量：

client.fabric_capacities.begin_resume(
    resource_group_name=resource_group,
    capacity_name=capacity_name
).result()

print("Capacity resumed")

删除容量

client.fabric_capacities.begin_delete(
    resource_group_name=resource_group,
    capacity_name=capacity_name
).result()

print("Capacity deleted")

检查名称是否可用

from azure.mgmt.fabric.models import CheckNameAvailabilityRequest

result = client.fabric_capacities.check_name_availability(
    location="eastus",
    body=CheckNameAvailabilityRequest(
        name="my-new-capacity",
        type="Microsoft.Fabric/capacities"
    )
)

if result.name_available:
    print("Name is available")
else:
    print(f"Name not available: {result.reason}")

列出可用 SKU

skus = client.fabric_capacities.list_skus(
    resource_group_name=resource_group,
    capacity_name=capacity_name
)

for sku in skus:
    print(f"SKU: {sku.name} - Tier: {sku.tier}")

客户端操作

操作	方法
client.fabric_capacities	容量 CRUD 操作
client.operations	列出可用操作

Fabric SKU

SKU	说明	CUs
F2	入门级	2 个容量单位
F4	小型	4 个容量单位
F8	中型	8 个容量单位
F16	大型	16 个容量单位
F32	特大型	32 个容量单位
F64	2 倍特大型	64 个容量单位
F128	4 倍特大型	128 个容量单位
F256	8 倍特大型	256 个容量单位
F512	16 倍特大型	512 个容量单位
F1024	32 倍特大型	1024 个容量单位
F2048	64 倍特大型	2048 个容量单位

容量状态

状态	说明
Active	容量正在运行
Paused	容量已暂停（不计费）
Provisioning	正在创建
Updating	正在修改
Deleting	正在删除
Failed	操作失败

长时间运行的操作

所有修改操作都是长时间运行的操作（LRO）。使用 .result() 等待：

# Synchronous wait
capacity = client.fabric_capacities.begin_create_or_update(...).result()

# Or poll manually
poller = client.fabric_capacities.begin_create_or_update(...)
while not poller.done():
    print(f"Status: {poller.status()}")
    time.sleep(5)  # import time before this optional manual polling fragment
capacity = poller.result()

最佳实践

1. 使用 DefaultAzureCredential 进行身份验证
2. 暂停未使用的容量以降低成本
3. 从较小的 SKU 开始，并根据需要扩展
4. 使用标记进行成本跟踪和组织管理
5. 创建容量前检查名称是否可用
6. 正确处理 LRO——不要假定操作会立即完成
7. 设置容量管理员——指定有权管理工作区的用户
8. 通过 Azure Monitor 指标监控容量使用情况

适用场景

当需要执行概述中所述的工作流或操作时，可使用此技能。

限制

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