Managing Snowflake databases, schemas, tables, and views with Python¶

Creating a database¶

You can create a database by calling the DatabaseCollection.create method and passing a Database object that represents the database you want to create. To create a database, first create a Database object that specifies the database name.

Code in the following example creates a Database object representing a database named my_db and then creates the database by passing the Database object to the DatabaseCollection.create method:

from snowflake.core.database import Database

my_db = Database(name="my_db")
root.databases.create(my_db)

Getting database details¶

You can get information about a database by calling the DatabaseResource.fetch method, which returns a Database object.

Code in the following example gets information about a database named my_db:

my_db = root.databases["my_db"].fetch()
print(my_db.to_dict())

Listing databases¶

You can list databases using the iter method, which returns a PagedIter iterator.

Code in the following example lists databases whose name begins with my:

databases = root.databases.iter(like="my%")
for database in databases:
  print(database.name)

Dropping a database¶

You can drop a database using the DatabaseResource.drop method.

Code in the following example drops the my_db database:

my_db_res = root.databases["my_db"]
my_db_res.drop()

Creating a schema¶

To create a schema, first create a Schema object that specifies the schema name.

Code in the following example creates a Schema object representing a schema named my_schema:

from snowflake.core.schema import Schema

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

The code then creates the schema in the my_db database by passing the Schema object to the SchemaCollection.create method.

Getting schema details¶

You can get information about a schema by calling the SchemaResource.fetch method, which returns a Schema object.

Code in the following example gets a Schema object that represents the my_schema schema:

my_schema = root.databases["my_db"].schemas["my_schema"].fetch()
print(my_schema.to_dict())

Listing schemas¶

You can list the schemas in a specified database using the iter method. The method returns a PagedIter iterator of Schema objects.

Code in the following example lists schema names in the my_db database:

schema_list = root.databases["my_db"].schemas.iter()
for schema_obj in schema_list:
  print(schema_obj.name)

Dropping a schema¶

You can drop a schema using the SchemaResource.drop method.

Code in the following example drops the my_schema schema:

my_schema_res = root.databases["my_db"].schemas["my_schema"]
my_schema_res.drop()

Creating a table¶

To create a table, first create a Table object that specifies the table name, column names, and column data types.

Code in the following example creates a Table object representing a table named my_table with the specified columns:

from snowflake.core.table import Table, TableColumn

my_table = Table(
  name="my_table",
  columns=[TableColumn(name="c1", datatype="int", nullable=False),
           TableColumn(name="c2", datatype="string")]
)
root.databases["my_db"].schemas["my_schema"].tables.create(my_table)

The code then creates the table in the my_db database and my_schema schema by passing the Table object to the TableCollection.create method.

Getting table details¶

You can get information about a table by calling the TableResource.fetch method, which returns a Table object.

Code in the following example gets information about a table named my_table:

my_table = root.databases["my_db"].schemas["my_schema"].tables["my_table"].fetch()
print(my_table.to_dict())

Creating or altering a table¶

You can set properties of a Table object and pass it to the TableResource.create_or_alter method to create a table if it doesn’t exist, or alter it according to the table definition if it does exist. The behavior of create_or_alter is intended to be idempotent, which means that the resulting table object will be the same regardless of whether the table exists before you run the method.

Code in the following example appends a new column named c3 of datatype int to the my_table table, and then alters the table in Snowflake:

from snowflake.core.table import PrimaryKey, TableColumn

my_table = root.databases["my_db"].schemas["my_schema"].tables["my_table"].fetch()
my_table.columns.append(TableColumn(name="c3", datatype="int", nullable=False, constraints=[PrimaryKey()]))

my_table_res = root.databases["my_db"].schemas["my_schema"].tables["my_table"]
my_table_res.create_or_alter(my_table)

Listing tables¶

You can list the tables in a specified schema using the iter method, which returns a PagedIter iterator of Table objects.

Code in the following example lists tables whose name begins with my:

tables = root.databases["my_db"].schemas["my_schema"].tables.iter(like="my%")
for table_obj in tables:
  print(table_obj.name)

Dropping a table¶

You can drop a table using the TableResource.drop method.

Code in the following example drops the my_table table:

my_table_res = root.databases["my_db"].schemas["my_schema"].tables["my_table"]
my_table_res.drop()

Creating an event table¶

To create an event table, first create a EventTable object, and then create a EventTableCollection object from the API Root object. Using EventTableCollection.create, add the new event table to Snowflake.

Code in the following example creates a EventTable object that represents an event table named my_event_table with the specified parameters:

from snowflake.core.event_table import EventTable

event_table = EventTable(
  name="my_event_table",
  data_retention_time_in_days = 3,
  max_data_extension_time_in_days = 5,
  change_tracking = True,
  default_ddl_collation = 'EN-CI',
  comment = 'CREATE EVENT TABLE'
)

event_tables = root.databases["my_db"].schemas["my_schema"].event_tables
event_tables.create(my_event_table)

The code creates a EventTableCollection variable event_tables and uses EventTableCollection.create to create a new event table in Snowflake.

Getting event table details¶

You can get information about an event table by calling the EventTableResource.fetch method, which returns a EventTable object.

Code in the following example gets information about an event table named my_event_table:

my_event_table = root.databases["my_db"].schemas["my_schema"].event_tables["my_event_table"].fetch()
print(my_event_table.to_dict())

Listing event tables¶

You can list event tables using the EventTableCollection.iter method, which returns a PagedIter iterator of EventTable objects.

Code in the following example lists event tables whose name starts with my in the my_db database and my_schema schema, and prints the name of each:

from snowflake.core.event_table import EventTableCollection

event_tables: EventTableCollection = root.databases["my_db"].schemas["my_schema"].event_tables
event_table_iter = event_tables.iter(like="my%")  # returns a PagedIter[EventTable]
for event_table_obj in event_table_iter:
  print(event_table_obj.name)

Code in the following example also lists event tables whose name begins with my, but it uses the starts_with parameter instead of like. This example also sets the optional parameter show_limit=10 to limit the number of results to 10:

event_tables: EventTableCollection = root.databases["my_db"].schemas["my_schema"].event_tables
event_table_iter = event_tables.iter(starts_with="my", show_limit=10)
for event_table_obj in event_table_iter:
  print(event_table_obj.name)

Performing event table operations¶

You can perform common event table operations—such as renaming an event table and dropping an event table—with a EventTableResource object.

To demonstrate operations you can do with an event table resource, code in the following example does the following:

1. Fetches the my_event_table event table resource object.

2. Renames the event table.

3. Drops the event table.

my_event_table_res = root.databases["my_db"].schemas["my_schema"].event_tables["my_event_table"]

my_event_table_res.rename("my_other_event_table")
my_event_table_res.drop()

Creating a view¶

To create a view, first create a View object that specifies the view name, columns, and SQL query statement.

Code in the following example creates a View object representing a view named my_view with the specified columns and SQL query:

from snowflake.core.view import View, ViewColumn

my_view = View(
  name="my_view",
  columns=[
      ViewColumn(name="c1"), ViewColumn(name="c2"), ViewColumn(name="c3"),
  ],
  query="SELECT * FROM my_table",
)

root.databases["my_db"].schemas["my_schema"].views.create(my_view)

The code then creates the view in the my_db database and my_schema schema by passing the View object to the ViewCollection.create method.

Getting view details¶

You can get information about a view by calling the ViewResource.fetch method, which returns a View object.

Code in the following example gets a View object that represents the my_view view:

my_view = root.databases["my_db"].schemas["my_schema"].views["my_view"].fetch()
print(my_view.to_dict())

Listing views¶

You can list the views in a specified database using the iter method. The method returns a PagedIter iterator of View objects.

Code in the following example lists views whose name begins with my in the my_db database and my_schema schema:

view_list = root.databases["my_db"].schemas["my_schema"].views.iter(like="my%")
for view_obj in view_list:
  print(view_obj.name)

Code in the following example also lists views whose name begins with my, but it uses the starts_with parameter instead of like. This example also sets the optional parameter show_limit=10 to limit the number of results to 10:

view_list = root.databases["my_db"].schemas["my_schema"].views.iter(starts_with="my", show_limit=10)
for view_obj in view_list:
  print(view_obj.name)

Dropping a view¶

You can drop a view using the ViewResource.drop method.

Code in the following example drops the my_view view:

my_view_res = root.databases["my_db"].schemas["my_schema"].views["my_view"]
my_view_res.drop()