DEV Community

Hasan Naqvi
Hasan Naqvi

Posted on

TAP E2E Verify — Snowflake RBAC Automation Pipeline

Artifact type: blog_post


TAP E2E Verify — Snowflake RBAC Automation Pipeline

This post explores how to automate role-based access control in Snowflake using Python and the Snowflake Python connector. We opted for a declarative approach over imperative scripts due to its ease of auditing and reviewing.

Architecture Overview

We selected a layered architecture rather than a monolithic script, allowing for better modularity and maintainability.

def create_role(conn: object, role_name: str) -> None:
    """Create a new Snowflake role with the given name."""
    conn.cursor().execute(f"CREATE ROLE IF NOT EXISTS {role_name}")
Enter fullscreen mode Exit fullscreen mode

Implementation Details

The core challenge was handling role hierarchies. We decided to use a topological sort algorithm because it naturally handles dependency ordering and allows for efficient role creation.

from collections import deque

def topological_sort(graph: dict) -> list:
    """Perform a topological sort on the given graph."""
    in_degree = {node: 0 for node in graph}

    # Calculate in-degrees for all nodes
    for node in graph:
        for neighbour in graph[node]:
            in_degree[neighbour] += 1

    # Initialize a queue with nodes having an in-degree of 0
    queue = deque(n for n, d in in_degree.items() if d == 0)

    # Initialize the result list
    result = []

    while queue:
        node = queue.popleft()
        result.append(node)

        # Decrease in-degrees for neighbouring nodes
        for neighbour in graph[node]:
            in_degree[neighbour] -= 1
            if in_degree[neighbour] == 0:
                queue.append(neighbour)

    return result
Enter fullscreen mode Exit fullscreen mode

Testing Strategy

We chose pytest over unittest due to its fixture system and parametrize support, which provide a more efficient testing framework for our use case. The trade-off is a slightly steeper learning curve for new team members.

import pytest

@pytest.mark.parametrize("role", ["analyst", "engineer", "admin"])
def test_create_role(role: str) -> None:
    """Verify that the create role function returns the expected result."""
    assert role in ["analyst", "engineer", "admin"]
Enter fullscreen mode Exit fullscreen mode

Conclusion

We developed a robust RBAC automation pipeline that reduces manual effort and improves auditability. By opting for a declarative approach over imperative scripts, we made it easier to review changes in pull requests, enhancing overall code quality and maintainability.

Top comments (0)