Skip to main content

Module airbyte.caches.snowflake

A Snowflake implementation of the PyAirbyte cache.

Usage Example​

Password connection:

from airbyte as ab
from airbyte.caches import SnowflakeCache

cache = SnowflakeCache(
    account="myaccount",
    username="myusername",
    password=ab.get_secret("SNOWFLAKE_PASSWORD"), # optional
    warehouse="mywarehouse",
    database="mydatabase",
    role="myrole",
    schema_name="myschema",
)

Private key connection:

from airbyte as ab
from airbyte.caches import SnowflakeCache

cache = SnowflakeCache(
    account="myaccount",
    username="myusername",
    private_key=ab.get_secret("SNOWFLAKE_PRIVATE_KEY"),
    private_key_passphrase=ab.get_secret("SNOWFLAKE_PRIVATE_KEY_PASSPHRASE"), # optional
    warehouse="mywarehouse",
    database="mydatabase",
    role="myrole",
    schema_name="myschema",
)

Private key path connection:

from airbyte as ab
from airbyte.caches import SnowflakeCache

cache = SnowflakeCache(
    account="myaccount",
    username="myusername",
    private_key_path="path/to/my/private_key.pem",
    private_key_passphrase=ab.get_secret("SNOWFLAKE_PRIVATE_KEY_PASSPHRASE"), # optional
    warehouse="mywarehouse",
    database="mydatabase",
    role="myrole",
    schema_name="myschema",
)

Classes​

SnowflakeCache(**data: Any) : Configuration for the Snowflake cache.

Initialize the cache and backends.

Ancestors (in MRO)​

  • airbyte._processors.sql.snowflake.SnowflakeConfig
  • airbyte.caches.base.CacheBase
  • airbyte.shared.sql_processor.SqlConfig
  • pydantic.main.BaseModel
  • airbyte._writers.base.AirbyteWriterInterface
  • abc.ABC

Class variables​

dedupe_mode: RecordDedupeMode :

model_config :

paired_destination_config_class: ClassVar[type | None] : DestinationSnowflake(database: 'str', host: 'str', role: 'str', schema: 'str', username: 'str', warehouse: 'str', credentials: 'Optional[AuthorizationMethod]' = None, DESTINATION_TYPE: 'Final[Snowflake]' = <Snowflake.SNOWFLAKE: 'snowflake'>, disable_type_dedupe: 'Optional[bool]' = False, jdbc_url_params: 'Optional[str]' = None, raw_data_schema: 'Optional[str]' = None, retention_period_days: 'Optional[int]' = 1, use_merge_for_upsert: 'Optional[bool]' = False)

paired_destination_name: ClassVar[str | None] :

SnowflakeConfig(**data: Any) : Configuration for the Snowflake cache.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Ancestors (in MRO)​

  • airbyte.shared.sql_processor.SqlConfig
  • pydantic.main.BaseModel
  • abc.ABC

Descendants​

  • airbyte.caches.snowflake.SnowflakeCache

Class variables​

account: str :

data_retention_time_in_days: int | None :

database: str :

model_config :

password: SecretString | None :

private_key: SecretString | None :

private_key_passphrase: SecretString | None :

private_key_path: str | None :

role: str :

username: str :

warehouse: str :

Methods​

get_sql_alchemy_url(self) ‑> airbyte.secrets.base.SecretString : Return the SQLAlchemy URL to use.

get_vendor_client(self) ‑> object : Return the Snowflake connection object.

Was this page helpful?