feat(streamlit): resolve STREAMLIT grants at plan time (fetch_streamlit)

conftest.py

@@ -145,6 +145,21 @@ def dummy_cursor(request):
         yield request.getfixturevalue("cursor")
 
 
+@pytest.fixture(autouse=True)
+def reset_account_usage_caches_between_tests():
+    """
+    The ACCOUNT_USAGE caches in data_provider are keyed by id(session). CPython reuses the
+    addresses of freed objects, so a mock session in one test can land on the address of a
+    previous test's session and inherit its cached grants. Reset all four caches around every
+    test so cache state never leaks between them.
+    """
+    from snowcap.data_provider import reset_account_usage_caches
+
+    reset_account_usage_caches()
+    yield
+    reset_account_usage_caches()
+
+
 @pytest.fixture(autouse=True)
 def reset_cursor_context(dummy_cursor, test_db):
     """

docs/resources/grant.md

@@ -45,6 +45,11 @@ grants:
     on: warehouse some_warehouse
     to: some_role
 
+  # Apps: USAGE on a Streamlit app so a role can open and run it
+  - priv: USAGE
+    on: streamlit somedb.someschema.some_streamlit
+    to: app_viewer_role
+
   # AI: account-level privilege for Cortex AI SQL functions
   - priv: USE AI FUNCTIONS
     on: ACCOUNT

docs/resources/streamlit.md

@@ -69,4 +69,42 @@ streamlit_repo = Streamlit(
 * `query_warehouse` (string) - The name of the warehouse to use for queries in the app.
 * `comment` (string) - A comment or description for the Streamlit app.
 * `owner` (string or Role) - The role that owns the Streamlit app. Defaults to "SYSADMIN".
-* `tags` (dict) - A dictionary of tags to associate with the Streamlit app.
+* `tags` (dict) - A dictionary of tags to associate with the Streamlit app.
+
+## Granting access
+
+A Streamlit app is a schema-scoped object. Grant `USAGE` on the app to let a
+role open and run it — this is the whole access story for viewers when the app
+uses owner's rights (all app queries execute as the app owner, so viewers need
+no privileges on the underlying tables):
+
+```yaml
+grants:
+  # Let a viewer role open and run the app.
+  - priv: USAGE
+    on: streamlit my_db.my_schema.my_streamlit
+    to: app_viewer_role
+
+  # Schema-scope privilege to allow a role to create Streamlit apps.
+  - priv: CREATE STREAMLIT
+    on: schema my_db.my_schema
+    to: app_developer_role
+```
+
+```python
+# Grant USAGE so a role can open the app
+grant = Grant(
+    priv="USAGE",
+    on_streamlit="my_db.my_schema.my_streamlit",
+    to="app_viewer_role",
+)
+```
+
+| Privilege         | Purpose                                                                  |
+|-------------------|--------------------------------------------------------------------------|
+| `USAGE`           | Open, view, and run the Streamlit app (and `DESCRIBE` it).               |
+| `OWNERSHIP`       | Full control. Set at create/deploy time — Snowflake does not support transferring streamlit ownership via `GRANT OWNERSHIP`. |
+| `ALL`             | All privileges above.                                                    |
+
+The schema-scope privilege `CREATE STREAMLIT` lets a role create apps in that
+schema; creating an app with a `ROOT_LOCATION` stage also needs `CREATE STAGE`.

setup.py

@@ -55,7 +55,10 @@
             "pytest-xdist",
             "pytest>=6.0",
             "python-dotenv",
-            "ruff",
+            # ruff 0.16 expanded the default rule set (I001, UP, SIM, TRY, ...), which
+            # reports 723 errors on snowcap/ as it stands on main. Pin below it until
+            # those are addressed or the rule selection is made explicit in pyproject.
+            "ruff<0.16",
             "tabulate",
             "twine!=5.1.0",
             "types-pytz",

snowcap/data_provider.py

@@ -2773,6 +2773,37 @@ def fetch_stream(session: SnowflakeConnection, fqn: FQN):
         raise NotImplementedError(f"Unsupported stream source type {data['source_type']}")
 
 
+def fetch_streamlit(session: SnowflakeConnection, fqn: FQN):
+    streamlits = _show_resources(session, "STREAMLITS", fqn)
+    if len(streamlits) == 0:
+        return None
+    if len(streamlits) > 1:
+        raise Exception(f"Found multiple streamlits matching {fqn}")
+
+    data = streamlits[0]
+    desc_result = execute(session, f"DESC STREAMLIT {fqn}", cacheable=True)
+    properties = desc_result[0]
+    return {
+        "name": data["name"],
+        # from_ is omitted (like fetch_notebook): DESC STREAMLIT returns
+        # root_location as a fully-qualified, uppercased stage path, which
+        # never matches the declared from_ (e.g. "@my_stage") and would show
+        # perpetual drift. The spec marks from_ non-fetchable instead.
+        # version is only settable for repo-based apps and isn't returned by
+        # DESC STREAMLIT in a comparable form, so it's always None here —
+        # repo-based apps (from_="https://...", version="main") may drift on
+        # version. Known limitation.
+        "version": None,
+        # Snowflake's default main_file is streamlit_app.py; map it to None so
+        # an omitted field doesn't drift (same as fetch_notebook).
+        "main_file": None if properties.get("main_file") == "streamlit_app.py" else properties.get("main_file"),
+        "title": properties.get("title") or None,
+        "query_warehouse": data.get("query_warehouse") or None,
+        "comment": data.get("comment") or None,
+        "owner": _get_owner_identifier(data),
+    }
+
+
 def fetch_tag(session: SnowflakeConnection, fqn: FQN):
     try:
         show_result = execute(session, "SHOW TAGS IN ACCOUNT", cacheable=True)

snowcap/resources/streamlit.py

@@ -1,4 +1,4 @@
-from dataclasses import dataclass
+from dataclasses import dataclass, field
 from typing import Optional
 
 from ..enums import ResourceType
@@ -17,7 +17,10 @@ class _Streamlit(ResourceSpec):
     """
 
     name: ResourceName
-    from_: str
+    # DESC STREAMLIT returns root_location as a fully-qualified, uppercased
+    # stage path that never matches the declared from_ (e.g. "@my_stage"), so
+    # from_ is not fetchable — YAML is authoritative (same as Notebook).
+    from_: str = field(default=None, metadata={"fetchable": False})
     version: Optional[str] = None
     main_file: Optional[str] = None
     title: Optional[str] = None
@@ -27,7 +30,7 @@ class _Streamlit(ResourceSpec):
 
     def __post_init__(self):
         super().__post_init__()
-        if self.from_.startswith("@"):
+        if self.from_ is not None and self.from_.startswith("@"):
             if self.version is not None:
                 raise ValueError("Version should not be set when the source is a stage")
 
@@ -104,10 +107,10 @@ class Streamlit(NamedResource, TaggableResource, Resource):
     scope = SchemaScope()
     spec = _Streamlit
 
-    def init(
+    def __init__(
         self,
         name: str,
-        from_: str,
+        from_: str = None,
         version: Optional[str] = None,
         main_file: Optional[str] = None,
         title: Optional[str] = None,

tests/test_data_provider.py

@@ -853,15 +853,6 @@ def test_returns_region(self, mock_execute):
 class TestHasAccountUsageAccess:
     """Tests for _has_account_usage_access function."""
 
-    def setup_method(self):
-        """Clear caches before each test."""
-        from snowcap.data_provider import (
-            _ACCOUNT_USAGE_ACCESS_CACHE,
-            _ACCOUNT_USAGE_FALLBACK_CACHE,
-        )
-        _ACCOUNT_USAGE_ACCESS_CACHE.clear()
-        _ACCOUNT_USAGE_FALLBACK_CACHE.clear()
-
     @patch("snowcap.data_provider.execute")
     def test_returns_true_when_access_granted(self, mock_execute):
         """When ACCOUNT_USAGE query succeeds, function returns True."""
@@ -936,15 +927,6 @@ def test_different_sessions_have_independent_cache(self, mock_execute):
 class TestFetchGrantsFromAccountUsage:
     """Tests for _fetch_grants_from_account_usage function."""
 
-    def setup_method(self):
-        """Clear caches before each test."""
-        from snowcap.data_provider import (
-            _ACCOUNT_USAGE_ACCESS_CACHE,
-            _ACCOUNT_USAGE_FALLBACK_CACHE,
-        )
-        _ACCOUNT_USAGE_ACCESS_CACHE.clear()
-        _ACCOUNT_USAGE_FALLBACK_CACHE.clear()
-
     @patch("snowcap.data_provider.execute")
     def test_returns_normalized_grants(self, mock_execute):
         """Grants should be normalized to match SHOW GRANTS structure."""
@@ -1050,15 +1032,6 @@ def test_marks_fallback_on_error(self, mock_execute):
 class TestFetchRoleGrantsToUsersFromAccountUsage:
     """Tests for _fetch_role_grants_to_users_from_account_usage function."""
 
-    def setup_method(self):
-        """Clear caches before each test."""
-        from snowcap.data_provider import (
-            _ACCOUNT_USAGE_ACCESS_CACHE,
-            _ACCOUNT_USAGE_FALLBACK_CACHE,
-        )
-        _ACCOUNT_USAGE_ACCESS_CACHE.clear()
-        _ACCOUNT_USAGE_FALLBACK_CACHE.clear()
-
     @patch("snowcap.data_provider.execute")
     def test_returns_normalized_user_grants(self, mock_execute):
         """User grants should be normalized to match SHOW GRANTS OF ROLE structure."""
@@ -1102,15 +1075,6 @@ def test_returns_none_on_error(self, mock_execute):
 class TestShouldUseAccountUsage:
     """Tests for _should_use_account_usage helper function."""
 
-    def setup_method(self):
-        """Clear caches before each test."""
-        from snowcap.data_provider import (
-            _ACCOUNT_USAGE_ACCESS_CACHE,
-            _ACCOUNT_USAGE_FALLBACK_CACHE,
-        )
-        _ACCOUNT_USAGE_ACCESS_CACHE.clear()
-        _ACCOUNT_USAGE_FALLBACK_CACHE.clear()
-
     @patch("snowcap.data_provider._has_account_usage_access")
     def test_returns_false_when_config_disabled(self, mock_has_access):
         """Returns False when use_account_usage config is False."""
@@ -1166,11 +1130,6 @@ def test_returns_false_when_no_access(self, mock_has_access):
 class TestMarkAccountUsageFallback:
     """Tests for _mark_account_usage_fallback function."""
 
-    def setup_method(self):
-        """Clear caches before each test."""
-        from snowcap.data_provider import _ACCOUNT_USAGE_FALLBACK_CACHE
-        _ACCOUNT_USAGE_FALLBACK_CACHE.clear()
-
     def test_marks_session_for_fallback(self):
         """Should mark session ID in fallback cache."""
         from snowcap.data_provider import (
@@ -1187,15 +1146,6 @@ def test_marks_session_for_fallback(self):
 class TestFetchRolePrivilegesAccountUsage:
     """Tests for fetch_role_privileges with ACCOUNT_USAGE integration."""
 
-    def setup_method(self):
-        """Clear caches before each test."""
-        from snowcap.data_provider import (
-            _ACCOUNT_USAGE_ACCESS_CACHE,
-            _ACCOUNT_USAGE_FALLBACK_CACHE,
-        )
-        _ACCOUNT_USAGE_ACCESS_CACHE.clear()
-        _ACCOUNT_USAGE_FALLBACK_CACHE.clear()
-
     @patch("snowcap.data_provider._should_use_account_usage")
     @patch("snowcap.data_provider._fetch_grants_from_account_usage")
     @patch("snowcap.data_provider._show_grants_to_role")

tests/test_grant.py

@@ -174,6 +174,24 @@ def test_grant_on_cortex_search_service():
     assert "MONITOR ON CORTEX SEARCH SERVICE" in monitor_grant.create_sql()
 
 
+def test_grant_on_streamlit():
+    """USAGE on a STREAMLIT parses and renders correctly.
+
+    Streamlit is a schema-scoped app object. Granting USAGE lets a viewer
+    role open and run the app:
+        GRANT USAGE ON STREAMLIT <db>.<schema>.<app> TO ROLE r
+    OWNERSHIP is established at create/deploy time, not transferred via GRANT.
+    """
+    grant = res.Grant(
+        priv="USAGE",
+        on_streamlit="somedb.someschema.someapp",
+        to="somerole",
+    )
+    assert grant._data.on == "SOMEDB.SOMESCHEMA.SOMEAPP"
+    assert grant._data.on_type == ResourceType.STREAMLIT
+    assert "USAGE ON STREAMLIT" in grant.create_sql()
+
+
 def test_grant_database_role_to_database_role():
     database = res.Database(name="somedb")
     parent = res.DatabaseRole(name="parent", database=database)

tests/test_privs.py

@@ -347,6 +347,25 @@ def test_ownership_privilege(self):
         assert CortexSearchServicePriv.OWNERSHIP.value == "OWNERSHIP"
 
 
+#############################################################################
+# StreamlitPriv Tests
+#############################################################################
+
+
+class TestStreamlitPriv:
+    """Tests for StreamlitPriv enum values (Snowflake Streamlit apps)."""
+
+    def test_all_privilege(self):
+        assert StreamlitPriv.ALL.value == "ALL"
+
+    def test_usage_privilege(self):
+        """USAGE is the priv needed to open and run a Streamlit app."""
+        assert StreamlitPriv.USAGE.value == "USAGE"
+
+    def test_ownership_privilege(self):
+        assert StreamlitPriv.OWNERSHIP.value == "OWNERSHIP"
+
+
 #############################################################################
 # is_ownership_priv() Tests
 #############################################################################