I believe there is a bug located within the following script, but am unsure. The arguments and the get values seem to be out of sync.
class SnowflakeCredentials(Block): """ Block used to manage authentication with Snowflake. Args: account (str): The snowflake account name. user (str): The user name used to authenticate. password (SecretStr): The password used to authenticate. private_key (SecretStr): The PEM used to authenticate. authenticator (str): The type of authenticator to use for initializing connection (oauth, externalbrowser, etc); refer to [Snowflake documentation](https://docs.snowflake.com/en/user-guide/python-connector-api.html#connect) for details, and note that `externalbrowser` will only work in an environment where a browser is available. token (SecretStr): The OAuth or JWT Token to provide when authenticator is set to OAuth. okta_endpoint (str): The Okta endpoint to use when authenticator is set to `okta_endpoint`, e.g. `https://<okta_account_name>.okta.com`. role (str): The name of the default role to use. autocommit (bool): Whether to automatically commit. Example: Load stored Snowflake credentials: ```python from prefect_snowflake import SnowflakeCredentials snowflake_credentials_block = SnowflakeCredentials.load("BLOCK_NAME") ``` authenticator: Literal[ "snowflake", "externalbrowser", "okta_endpoint", "oauth", "username_password_mfa", ] = Field( # noqa default="snowflake", description=("The type of authenticator to use for initializing connection"), ), ) endpoint: Optional[str] = Field( default=None, description=( "The Okta endpoint to use when authenticator is set to `okta_endpoint`" ), )
Which then violates the following method below as the requested value doesn’t exist.
@root_validator(pre=True) def _validate_okta_kwargs(cls, values): """ Ensure an authorization value has been provided by the user. """ authenticator = values.get("authenticator") okta_endpoint = values.get("okta_endpoint") if authenticator == "okta_endpoint" and not okta_endpoint: raise ValueError( "If authenticator is set to `okta_endpoint`, " "`okta_endpoint` must be provided" ) return values