75class SnowflakeCache(SnowflakeConfig, CacheBase):
76    """Configuration for the Snowflake cache."""
77
78    dedupe_mode: RecordDedupeMode = RecordDedupeMode.APPEND
79
80    _sql_processor_class: ClassVar[type[SqlProcessorBase]] = SnowflakeSqlProcessor
81
82    paired_destination_name: ClassVar[str | None] = "destination-bigquery"
83    paired_destination_config_class: ClassVar[type | None] = DestinationSnowflake
84
85    @property
86    def paired_destination_config(self) -> DestinationSnowflake:
87        """Return a dictionary of destination configuration values."""
88        return snowflake_cache_to_destination_configuration(cache=self)

 37class SnowflakeConfig(SqlConfig):
 38    """Configuration for the Snowflake cache."""
 39
 40    account: str
 41    username: str
 42    password: SecretString | None = None
 43
 44    private_key: SecretString | None = None
 45    private_key_path: str | None = None
 46    private_key_passphrase: SecretString | None = None
 47
 48    warehouse: str
 49    database: str
 50    role: str
 51    schema_name: str = Field(default=DEFAULT_CACHE_SCHEMA_NAME)
 52    data_retention_time_in_days: int | None = None
 53
 54    def _validate_authentication_config(self) -> None:
 55        """Validate that authentication configuration is correct."""
 56        auth_methods = {
 57            "password": self.password is not None,
 58            "private_key": self.private_key is not None,
 59            "private_key_path": self.private_key_path is not None,
 60        }
 61        has_passphrase = self.private_key_passphrase is not None
 62        primary_auth_count = sum(auth_methods.values())
 63
 64        if primary_auth_count == 0:
 65            if has_passphrase:
 66                raise ValueError(
 67                    "You have to provide a primary authentication method "
 68                    "if you want to use a private key passphrase."
 69                )
 70            return
 71
 72        if primary_auth_count > 1:
 73            provided_methods = [method for method, has_method in auth_methods.items() if has_method]
 74            raise ValueError(
 75                f"Multiple primary authentication methods provided: {', '.join(provided_methods)}. "
 76                "Please provide only one of: 'password', 'private_key', or 'private_key_path'."
 77            )
 78
 79        if has_passphrase and auth_methods["password"]:
 80            raise ValueError(
 81                "private_key_passphrase cannot be used with password authentication. "
 82                "It can only be used with 'private_key' or 'private_key_path'."
 83            )
 84
 85    def _get_private_key_content(self) -> bytes:
 86        """Get the private key content from either private_key or private_key_path."""
 87        if self.private_key:
 88            return str(self.private_key).encode("utf-8")
 89        if self.private_key_path:
 90            return Path(self.private_key_path).read_bytes()
 91        raise ValueError("No private key provided")
 92
 93    def _get_private_key_bytes(self) -> bytes:
 94        private_key_content = self._get_private_key_content()
 95
 96        passphrase = None
 97        if self.private_key_passphrase:
 98            passphrase = str(self.private_key_passphrase).encode("utf-8")
 99
100        private_key = serialization.load_pem_private_key(
101            private_key_content,
102            password=passphrase,
103            backend=default_backend(),
104        )
105
106        return private_key.private_bytes(
107            encoding=serialization.Encoding.DER,
108            format=serialization.PrivateFormat.PKCS8,
109            encryption_algorithm=serialization.NoEncryption(),
110        )
111
112    @overrides
113    def get_sql_alchemy_connect_args(self) -> dict[str, Any]:
114        """Return the SQL Alchemy connect_args."""
115        if self.private_key is None and self.private_key_path is None:
116            return {}
117        return {"private_key": self._get_private_key_bytes()}
118
119    @overrides
120    def get_create_table_extra_clauses(self) -> list[str]:
121        """Return a list of clauses to append on CREATE TABLE statements."""
122        clauses = []
123
124        if self.data_retention_time_in_days is not None:
125            clauses.append(f"DATA_RETENTION_TIME_IN_DAYS = {self.data_retention_time_in_days}")
126
127        return clauses
128
129    @overrides
130    def get_database_name(self) -> str:
131        """Return the name of the database."""
132        return self.database
133
134    @overrides
135    def get_sql_alchemy_url(self) -> SecretString:
136        """Return the SQLAlchemy URL to use."""
137        self._validate_authentication_config()
138
139        config = {
140            "account": self.account,
141            "user": self.username,
142            "database": self.database,
143            "warehouse": self.warehouse,
144            "schema": self.schema_name,
145            "role": self.role,
146        }
147        # password is absent when using key pair authentication
148        if self.password:
149            config["password"] = self.password
150        return SecretString(URL(**config))
151
152    def get_vendor_client(self) -> object:
153        """Return the Snowflake connection object."""
154        self._validate_authentication_config()
155
156        connection_config: dict[str, Any] = {
157            "user": self.username,
158            "account": self.account,
159            "warehouse": self.warehouse,
160            "database": self.database,
161            "schema": self.schema_name,
162            "role": self.role,
163        }
164
165        if self.password:
166            connection_config["password"] = self.password
167        elif self.private_key_path:
168            connection_config["private_key_file"] = self.private_key_path
169            if self.private_key_passphrase:
170                connection_config["private_key_file_pwd"] = self.private_key_passphrase
171            connection_config["authenticator"] = "SNOWFLAKE_JWT"
172        else:
173            connection_config["private_key"] = self._get_private_key_bytes()
174            connection_config["authenticator"] = "SNOWFLAKE_JWT"
175
176        return connector.connect(**connection_config)