🏢 Multi-Tenant Workspace Architecture for Microsoft Fabric¶

# Hybrid topology decision engine
def determine_workspace_topology(tenant: dict) -> str:
    """Determine the appropriate workspace topology for a tenant."""
    if tenant.get("regulatory_isolation_required"):
        return "PER_TENANT_FULL"  # Separate capacity if needed

    if tenant.get("ci_cd_required"):
        return "PER_ENVIRONMENT"  # Dev/Stg/Prod workspaces

    if tenant.get("data_volume_tb") > 10:
        return "PER_TENANT_FULL"  # Large tenants get dedicated workspaces

    return "PER_LAYER"  # Standard tenants share Bronze

# Example classification
tenants = [
    {"name": "Casino-Vegas", "regulatory_isolation_required": True,
     "ci_cd_required": True, "data_volume_tb": 25},     # → PER_TENANT_FULL
    {"name": "Casino-Regional", "regulatory_isolation_required": False,
     "ci_cd_required": True, "data_volume_tb": 2},       # → PER_ENVIRONMENT
    {"name": "Federal-EPA", "regulatory_isolation_required": True,
     "ci_cd_required": True, "data_volume_tb": 15},      # → PER_TENANT_FULL
    {"name": "Federal-SBA", "regulatory_isolation_required": False,
     "ci_cd_required": False, "data_volume_tb": 1},      # → PER_LAYER
]

# Workspace role assignments per tenant
workspace_roles = {
    "ws-casino-vegas-prod": {
        "Admin": ["sg-casino-platform-admins"],
        "Member": ["sg-casino-vegas-engineers"],
        "Contributor": ["sg-casino-vegas-analysts"],
        "Viewer": ["sg-casino-vegas-viewers"]
    },
    "ws-federal-usda-prod": {
        "Admin": ["sg-federal-platform-admins"],
        "Member": ["sg-usda-data-engineers"],
        "Contributor": ["sg-usda-analysts"],
        "Viewer": ["sg-usda-stakeholders"]
    }
}

# OneLake security configuration via REST API
onelake_security = {
    "workspace": "ws-federal-shared",
    "lakehouse": "lh_bronze_shared",
    "folder_permissions": [
        {
            "folder": "Tables/usda_crop_production",
            "read": ["sg-usda-data-engineers", "sg-usda-analysts"],
            "write": ["sg-usda-data-engineers"]
        },
        {
            "folder": "Tables/epa_air_quality",
            "read": ["sg-epa-data-engineers", "sg-epa-analysts"],
            "write": ["sg-epa-data-engineers"]
        }
    ]
}

import requests
from azure.identity import ClientSecretCredential

class FabricWorkspaceProvisioner:
    """Provision Fabric workspaces for new tenants."""

    def __init__(self, tenant_id: str, client_id: str, client_secret: str):
        self.credential = ClientSecretCredential(
            tenant_id=tenant_id,
            client_id=client_id,
            client_secret=client_secret
        )
        self.base_url = "https://api.fabric.microsoft.com/v1"

    def _get_headers(self) -> dict:
        token = self.credential.get_token("https://api.fabric.microsoft.com/.default")
        return {
            "Authorization": f"Bearer {token.token}",
            "Content-Type": "application/json"
        }

    def create_workspace(
        self,
        tenant_name: str,
        environment: str,
        capacity_id: str,
        description: str = ""
    ) -> dict:
        """Create a new workspace for a tenant."""
        workspace_name = f"ws-{tenant_name}-{environment}"
        payload = {
            "displayName": workspace_name,
            "description": description or f"Workspace for {tenant_name} ({environment})",
            "capacityId": capacity_id
        }

        response = requests.post(
            f"{self.base_url}/workspaces",
            headers=self._get_headers(),
            json=payload
        )
        response.raise_for_status()
        return response.json()

    def assign_roles(
        self,
        workspace_id: str,
        role_assignments: dict[str, list[str]]
    ):
        """Assign RBAC roles to a workspace."""
        for role, principals in role_assignments.items():
            for principal_id in principals:
                payload = {
                    "principal": {
                        "id": principal_id,
                        "type": "Group"
                    },
                    "role": role
                }
                requests.post(
                    f"{self.base_url}/workspaces/{workspace_id}/roleAssignments",
                    headers=self._get_headers(),
                    json=payload
                ).raise_for_status()

    def create_lakehouse(
        self,
        workspace_id: str,
        lakehouse_name: str
    ) -> dict:
        """Create a Lakehouse in the workspace."""
        payload = {
            "displayName": lakehouse_name,
            "type": "Lakehouse"
        }
        response = requests.post(
            f"{self.base_url}/workspaces/{workspace_id}/items",
            headers=self._get_headers(),
            json=payload
        )
        response.raise_for_status()
        return response.json()

    def provision_tenant(
        self,
        tenant_config: dict
    ) -> dict:
        """Full tenant provisioning pipeline."""
        results = {}
        tenant_name = tenant_config["name"]
        capacity_id = tenant_config["capacity_id"]

        for env in tenant_config.get("environments", ["prod"]):
            # Create workspace
            ws = self.create_workspace(tenant_name, env, capacity_id)
            ws_id = ws["id"]

            # Create lakehouses (Bronze, Silver, Gold)
            for layer in ["bronze", "silver", "gold"]:
                lh = self.create_lakehouse(ws_id, f"lh_{layer}_{tenant_name}")
                results[f"{env}_{layer}"] = lh["id"]

            # Assign roles
            self.assign_roles(ws_id, tenant_config["role_assignments"])

            results[f"{env}_workspace_id"] = ws_id

        return results


# Usage
provisioner = FabricWorkspaceProvisioner(
    tenant_id="...",
    client_id="...",
    client_secret="..."
)

provisioner.provision_tenant({
    "name": "casino-vegas",
    "capacity_id": "cap-casino-prod-id",
    "environments": ["dev", "stg", "prod"],
    "role_assignments": {
        "Admin": ["sg-casino-platform-admins-oid"],
        "Member": ["sg-casino-vegas-engineers-oid"],
        "Contributor": ["sg-casino-vegas-analysts-oid"],
        "Viewer": ["sg-casino-vegas-viewers-oid"]
    }
})

# Cross-workspace deployment for multi-tenant environments
from fabric_cicd import FabricWorkspace, publish_all_items

def deploy_to_tenant(
    tenant_name: str,
    source_env: str,
    target_env: str,
    item_types: list[str]
):
    """Deploy Fabric items from source to target environment workspace."""
    source_workspace_name = f"ws-{tenant_name}-{source_env}"
    target_workspace_name = f"ws-{tenant_name}-{target_env}"

    # Initialize target workspace
    target = FabricWorkspace(
        workspace_id=get_workspace_id(target_workspace_name),
        repository_directory=f"./fabric-items/{tenant_name}",
        item_type_in_scope=item_types
    )

    # Publish items to target
    publish_all_items(target)
    print(f"✅ Deployed {item_types} from {source_env} → {target_env} for {tenant_name}")


# Deploy to all tenants
tenants = ["casino-vegas", "casino-atlantic", "federal-usda", "federal-epa"]
for tenant in tenants:
    deploy_to_tenant(
        tenant_name=tenant,
        source_env="stg",
        target_env="prod",
        item_types=["Notebook", "SemanticModel", "SQLEndpoint"]
    )

casino_property_template = {
    "workspace_items": [
        "lh_bronze_{property}",        # Raw ingestion
        "lh_silver_{property}",        # Cleansed + validated
        "lh_gold_{property}",          # Property-level KPIs
        "wh_{property}",              # SQL analytics
        "sm_{property}_operations",    # Direct Lake model
        "sm_{property}_compliance",    # Compliance model
        "pipe_{property}_daily_ingest",
        "pipe_{property}_compliance_check"
    ],
    "required_tables": {
        "bronze": [
            "slot_transactions", "table_game_transactions",
            "player_sessions", "machine_events"
        ],
        "silver": [
            "slot_transactions_cleansed", "player_profiles",
            "ctr_candidates", "sar_candidates"
        ],
        "gold": [
            "slot_performance_daily", "revenue_by_denomination",
            "ctr_filings", "sar_alerts", "w2g_records",
            "player_lifetime_value"
        ]
    }
}

federal_agencies = [
    {
        "name": "usda",
        "display_name": "U.S. Department of Agriculture",
        "capacity_id": "cap-federal-prod",
        "environments": ["dev", "prod"],
        "datasets": ["crop_production", "livestock_survey", "agricultural_census"],
        "compliance": ["FISMA", "FedRAMP", "NIST 800-53"],
        "data_classification": "CUI",  # Controlled Unclassified Information
        "retention_years": 7,
        "role_assignments": {
            "Admin": ["sg-federal-platform-admins"],
            "Member": ["sg-usda-data-engineers"],
            "Contributor": ["sg-usda-analysts"],
            "Viewer": ["sg-usda-leadership"]
        }
    },
    {
        "name": "epa",
        "display_name": "Environmental Protection Agency",
        "capacity_id": "cap-federal-prod",
        "environments": ["dev", "prod"],
        "datasets": ["air_quality", "water_quality", "toxic_release"],
        "compliance": ["FISMA", "FedRAMP", "EPA_QA/QC"],
        "data_classification": "Public",
        "retention_years": 10,
        "role_assignments": {
            "Admin": ["sg-federal-platform-admins"],
            "Member": ["sg-epa-data-engineers"],
            "Contributor": ["sg-epa-scientists"],
            "Viewer": ["sg-epa-stakeholders"]
        }
    },
    {
        "name": "noaa",
        "display_name": "National Oceanic and Atmospheric Administration",
        "capacity_id": "cap-federal-prod",
        "environments": ["dev", "prod"],
        "datasets": ["weather_stations", "ocean_buoys", "climate_normals"],
        "compliance": ["FISMA", "FedRAMP"],
        "data_classification": "Public",
        "retention_years": 30,  # Climate data: long retention
        "role_assignments": {
            "Admin": ["sg-federal-platform-admins"],
            "Member": ["sg-noaa-data-engineers"],
            "Contributor": ["sg-noaa-meteorologists"],
            "Viewer": ["sg-noaa-researchers"]
        }
    }
]

# Monitor CU utilization per tenant workspace
def check_capacity_utilization(capacity_id: str) -> dict:
    """Get CU utilization breakdown by workspace."""
    # Use Fabric Capacity Metrics app or REST API
    metrics = get_capacity_metrics(capacity_id)

    utilization = {}
    for workspace in metrics["workspaces"]:
        utilization[workspace["name"]] = {
            "cu_seconds_24h": workspace["cuSeconds"],
            "pct_of_capacity": workspace["percentOfCapacity"],
            "throttled": workspace["throttled"],
            "top_operations": workspace["topOperations"][:5]
        }

    return utilization