snowflake.core.schema.SchemaCollection¶

class snowflake.core.schema.SchemaCollection(database: DatabaseResource, root: Root)¶

Bases: ObjectCollection[SchemaResource]

Represents the collection operations on the Snowflake schema resource.

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

Examples

Creating a schema instance:

>>> schemas = root.databases["my_db"].schemas
>>> new_schema = Schema("my_schema")
>>> schemas.create(new_schema)

Copy

Attributes

database¶

The DatabaseResource this schema belongs to.

root¶

The Root object this schema belongs to.

Methods

create(schema: ModelSchemaModel, *, clone: str | Clone | None = None, mode: CreateMode = CreateMode.error_if_exists) → SchemaResource¶

Create a schema in Snowflake.

Parameters:

• schema (SchemaResource) – The Schema object, together with the Schema’s properties: name; kind, comment, managed_access, retention_time, budget, data_retention_time_in_days, default_ddl_colaltion, log_level, pipe_execution_paused, max_data_extension_time_in_days, suspend_task_after_num_failures, trace_level, user_task_managed_initial_warehouse_size and user_task_timeout_ms are optional.

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

• mode (CreateMode, optional) –

  One of the following enum values.

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

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

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

  Default is CreateMode.error_if_exists.

Examples

Creating a new schema called new_schema in my_db:

>>> schemas = root.databases["my_db"].schemas
>>> new_schema_ref = schemas.create(Schema("new_schema"))

Copy

Creating a new schema called new_schema in my_db by cloning an existing schema:

>>> schemas = root.databases["my_db"].schemas
>>> new_schema_ref = schemas.create(
...     "new_schema",
...     clone = Clone(source="original_schema", point_of_time=PointOfTimeOffset(reference="at", when="-5")),
...     mode = CreateMode.or_replace
... )

Copy

Creating a new schema called new_schema in my_db by cloning an existing schema in another database:

>>> schemas = root.databases["my_db"].schemas
>>> new_schema_ref = schemas.create(
...     "new_schema",
...     clone = Clone(
...         source="another_database.original_schema",
...         point_of_time=PointOfTimeOffset(reference="at", when="-5")
...     ),
...     mode = CreateMode.or_replace
... )

Copy

items() → ItemsView[str, T]¶

iter(*, like: str | None = None, starts_with: str | None = None, limit: int | None = None, from_name: str | None = None) → Iterator[ModelSchemaModel]¶

Iterate through Schema 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 schemas that you have access to see:

>>> schemas = db_ref.schemas.iter()

Copy

Showing information of the exact schema you want to see:

>>> schemas = db_ref.schemas.iter(like="your-schema-name")

Copy

Showing schemas starting with ‘your-schema-name-‘:

>>> schemas = db_ref.schemas.iter(like="your-schema-name-%")

Copy

Using a for loop to retrieve information from iterator:

>>> for schema in schemas:
>>>     print(schema.name, schema.query)

Copy

keys() → KeysView[str]¶

values() → ValuesView[T]¶