Snowman: Pydantic Model Generator and Query Builder for Snowflake

Snowman is a Python library that provides a way to extract Snowflake table information in a type-safe.

Snowman provides two main features:

• Automatically generate Pydantic models from Snowflake information schemas
• Query builder to generate SQL queries in a type-safe

Generated Pydantic Model

Source: Snowflake

CREATE TABLE DATABASE.SCHEMA.USER (
    ID INTEGER NOT NULL COMMENT 'User ID',
    NAME TEXT NOT NULL COMMENT 'User Name',
    AGE INTEGER DEFAULT NULL COMMENT 'User Age',
    CREATED_AT TIMESTAMP_TZ NOT NULL DEFAULT CURRENT_TIMESTAMP() COMMENT 'Created At'
) COMMENT 'User Table';

Output: Python Code

#
# Code generated by snowman; DO NOT EDIT.
#
# For more information about snowman,
# please refer to https://github.com/yassun7010/snowman-py .
#

import datetime
import typing

import pydantic
import snowman

if typing.TYPE_CHECKING:
    from . import _schema as _schema


# TABLE: DATABASE.SCHEMA.USER
@snowman.table("DATABASE", "SCHEMA", "USER")
class User(
    snowman.Table[
        "User",
        "_schema._UserColumnAccessor",
        "_schema._UserOrderItemAccessor",
        "_schema._UserInsertTypedDict",
        "_schema._UserUpdateTypedDict",
    ]
):
    """User Table"""

    id: typing.Annotated[
        snowman.datatype.INTEGER, pydantic.Field(title="User ID", alias="ID")
    ]
    """User ID"""

    name: typing.Annotated[
        snowman.datatype.TEXT, pydantic.Field(title="User Name", alias="NAME")
    ]
    """User Name"""

    age: typing.Annotated[
        snowman.datatype.INTEGER | None, pydantic.Field(title="User Age", alias="AGE")
    ] = None
    """User Age"""

    created_at: typing.Annotated[
        snowman.datatype.TIMESTAMP_TZ,
        pydantic.Field(title="Created At", alias="CREATED_AT"),
    ] = snowman.pydantic.DefaultFactory(datetime.datetime.now)
    """Created At"""

Query Builder

SELECTINSERTUPDATEDELETETRUNCATE

🚧 This is a draft. 🚧

It does not support complex queries that include only specific fields or table joins.

Example

from snowman.query import select
from your.database.schema import User
from your.snowflake import snowflake_conn

with snowflake_conn.cursor() as cursor:
    user: User | None = (
        select()
        .from_(User)
        .where(
            lambda c: (c.self.id == 1)
            .and_(c.self.name == "John Doe")
            .and_(c.self.age >= 18),
        )
        .order.by(
            lambda c: [
                c.self.id.desc,
                c.self.name.asc.nulls.last,
            ],
        )
        .limit(1)
        .offset(0)
        .execute(cursor)
        .fetchone()
    )

Example

from snowman.query import insert
from your.database.schema import User
from your.snowflake import snowflake_conn

with snowflake_conn.cursor() as cursor:
    insert.into(User).values(
        User(
            id=1,
            name="John Doe",
        ),
    ).execute(cursor)

Example

import datetime

from snowman.query import update
from your.database.schema import User
from your.snowflake import snowflake_conn

with snowflake_conn.cursor() as cursor:
    update(
        User,
    ).set(
        {"name": "Jane Doe"},
    ).where(
        lambda c: (c.self.name.in_(["Jane", "Doe"]))
        .and_(c.self.age > 18)
        .and_(c.self.created_at >= datetime.datetime(2001, 1, 1)),
    ).execute(cursor)

Example

from snowman.query import delete
from your.database.schema import User
from your.snowflake import snowflake_conn

with snowflake_conn.cursor() as cursor:
    delete.from_(
        User,
    ).where(
        lambda c: c.self.id == 1,
    ).execute(cursor)

Example

from snowman.query import truncate
from your.database.schema import User
from your.snowflake import snowflake_conn

with snowflake_conn.cursor() as cursor:
    truncate.table.if_.exists(User).execute(cursor)