How do I setup everything under a fixed snowflake Schema?

[jasonsross]

In my org we have to work within fixed schemas. We might have 2 schemas, one for prod and one for dev tables, but we do not have permissions to create new schemas. How can I setup sqlmesh to do it's magic all within a single schema that already exists, ie snowflake_db.some_schema.[all_sqlmesh_tables/views_here]?

[hanseltime]

Hey! I just ran into this since giving "create schema" permissions is a pretty large surface area that we don't want to manage.

The solution I have come up with in the interim is to override the adapter's create_schema method in a config.py

from sqlmesh.core.config import Config, GatewayConfig, SnowflakeConnectionConfig, PostgresConnectionConfig
from sqlmesh.core.engine_adapter.base import EngineAdapter
from sqlmesh.core.engine_adapter.snowflake import SnowflakeEngineAdapter
from sqlmesh.core.engine_adapter.postgres import PostgresEngineAdapter
if t.TYPE_CHECKING:
    from sqlmesh.core._typing import SchemaName
else:
    SchemaName = t.Any
from sqlmesh.core.config import ModelDefaultsConfig, LinterConfig
from sqlglot import exp
from os import environ


class NoopCreateSchemaSnowflakeAdapter(SnowflakeEngineAdapter):
    """Basic Schema creation makes permissioning a manual nightmare long term.
    We override the "always try to create" behavior in our adapters

    Args:
        SnowflakeEngineAdapter (_type_): _description_
    """
    def create_schema(self,                       
        schema_name: SchemaName,
        ignore_if_exists: bool = True,
        warn_on_error: bool = True,
        properties: t.Optional[t.List[exp.Expression]] = None,) -> None:  # keep signature compatible
        # Completely ignore schema creation
        logging.getLogger(__name__).info("Skipping create_schema(%s)", schema_name)
        return None


class PatchedSnowflakeConnectionConfig(SnowflakeConnectionConfig):
    @property
    def _engine_adapter(self) -> t.Type[EngineAdapter]:
        # Use our wrapper instead of the built-in adapter
        return NoopCreateSchemaSnowflakeAdapter
    
class NoopCreateSchemaPostgresAdapter(PostgresEngineAdapter):
    """Basic Schema creation makes permissioning a manual nightmare long term.
    We override the "always try to create" behavior in our adapters

    Args:
        PostgresEngineAdapter (_type_): _description_
    """
    def create_schema(self,                       
        schema_name: SchemaName,
        ignore_if_exists: bool = True,
        warn_on_error: bool = True,
        properties: t.Optional[t.List[exp.Expression]] = None,) -> None:  # keep signature compatible
        # Completely ignore schema creation
        logging.getLogger(__name__).info("Skipping create_schema(%s)", schema_name)
        return None


class PatchedPostgresConnectionConfig(PostgresConnectionConfig):
    @property
    def _engine_adapter(self) -> t.Type[EngineAdapter]:
        # Use our custom adapter instead of the default PostgresEngineAdapter
        return NoopCreateSchemaPostgresAdapter


config = Config(
    gateways={
        # "prod": GatewayConfig(
        #     connection=PatchedSnowflakeConnectionConfig(
        #         type="snowflake",
        #         account="YOUR_ACCOUNT",
        #         user="YOUR_USER",
        #         password="YOUR_PASSWORD",
        #         warehouse="YOUR_WAREHOUSE",
        #         database="YOUR_DB",
        #         role="YOUR_ROLE",
        #     ),
        #     # optionally have a separate state_connection too, same pattern if desired
        # ),
        "local": GatewayConfig(
            connection=PatchedPostgresConnectionConfig(
                host=environ["ELT_DB_HOST"],
                user=environ["ELT_DB_USER"],
                password=environ["ELT_DB_PASSWORD"],
                port=int(environ["ELT_DB_PORT"]),
                database=environ["ELT_DB_NAME"]
            )
        )
    },
)

I've also opened an issue requesting this as a first class feature: #5623