snowflake.core.database.DatabaseCollection

class snowflake.core.database.DatabaseCollection(root: Root)

Bases: AccountObjectCollectionParent[DatabaseResource]

Represents the collection operations on the Snowflake Database resource.

With this collection, you can create, iterate through, and search for Databases that you have access to in the current context.

Examples

Creating a database instance:

>>> databases = root.databases
>>> new_database = Database(
...     name="my_new_database", comment="this is my new database to prototype a new feature in"
... )
>>> databases.create(new_database)
Copy

Attributes

root

The Root object this collection belongs to.

Methods

create(database: DatabaseModel, *, clone: str | Clone | None = None, from_share: str | None = None, mode: CreateMode = CreateMode.error_if_exists) → DatabaseResource

Create a database in Snowflake.

Parameters:

  • database (Database) – The Database object, together with the Database’s properties: name; kind, comment, retention_time, budget, data_retention_time_in_days, default_ddl_collation, log_level, max_data_extension_time_in_days, suspend_task_after_num_failures, trace_level, user_task_managed_initial_warehouse_size, user_task_timeout_ms serverless_task_min_statement_size and serverless_task_max_statement_size are optional.

  • clone (str, or Clone, optional) – Whether to clone an existing database. An instance of Clone, or None if no cloning is necessary.

  • from_share (str, optional) – ID of the share from which to create the database, in the form "<provider_account>.<share_name>”.

  • mode (CreateMode, optional) –

    One of the following enum values.

    CreateMode.error_if_exists: Throw an snowflake.core.exceptions.ConflictError if the database already exists in Snowflake. Equivalent to SQL create database <name> ....

    CreateMode.or_replace: Replace if the database already exists in Snowflake. Equivalent to SQL create or replace database <name> ....

    CreateMode.if_not_exists: Do nothing if the database already exists in Snowflake. Equivalent to SQL create database <name> if not exists...

    Default is CreateMode.error_if_exists.

    Example

    Creating a new database if it does not already exist:

    >>> new_db_ref = root.databases.create(Database(name="my_new_database"), mode=CreateMode.if_not_exists)
>>> print(new_db_ref.fetch())
    Copy

create_async(database: DatabaseModel, *, clone: str | Clone | None = None, from_share: str | None = None, mode: CreateMode = CreateMode.error_if_exists) → PollingOperation[DatabaseResource]

An asynchronous version of create().

Refer to PollingOperation for more information on asynchronous execution and the return type.

items() → ItemsView[str, T]
iter(*, like: str | None = None, starts_with: str | None = None, limit: int | None = None, from_name: str | None = None) → Iterator[DatabaseModel]

Iterate through Database objects from Snowflake, filtering on any optional ‘like’ pattern.

Parameters:

  • like (str, optional) – A case-insensitive string functioning as a filter, with support for SQL wildcard characters (% and _).

  • starts_with (str, optional) – String used to filter the command output based on the string of characters that appear at the beginning of the object name. Uses case-sensitive pattern matching.

  • limit (int, optional) – Limit of the maximum number of rows returned by iter(). The default is None, which behaves equivalently to show_limit=10000. This value must be between 1 and 10000.

  • from_name (str, optional) – Fetch rows only following the first row whose object name matches the specified string. This is case-sensitive and does not have to be the full name.

Examples

Showing all databases that you have access to see:

>>> databases = root.databases.iter()
Copy

Showing information of the exact database you want to see:

>>> databases = root.databases.iter(like="your-database-name")
Copy

Showing databases starting with ‘your-database-name-‘:

>>> databases = root.databases.iter(like="your-database-name-%")
Copy

Using a for loop to retrieve information from iterator:

>>> for database in databases:
>>>     print(database.name, database.query)
Copy
iter_async(*, like: str | None = None, starts_with: str | None = None, limit: int | None = None, from_name: str | None = None) → PollingOperation[Iterator[DatabaseModel]]

An asynchronous version of iter().

Refer to PollingOperation for more information on asynchronous execution and the return type.

keys() → KeysView[str]
values() → ValuesView[T]
Langage: Français