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