Query Builder¶
Snowman provides a query builder that maintains the syntax of SQL.
You can generate INSERT / UPDATE / DELETE / TRUNCATE queries, but
it does not currently official support complex query syntax like SELECT.
Tip
The query builder is based on simple rules.
- SQL reserved words can be written in a way that can be accessed like properties with a dot
.in Python. - If a SQL reserved word matches a Python reserved word, an underscore
_is added to the end of the sql reserved word.
If you know SQL, you can build Snowman queries with a fluent python coding 🚀
Note
The paramstyle used by Snowman is format.
The reason for not supporting pyformat is to avoid key duplication
when complex query patterns occur.
In other words, it uses %s to embed parameters, not %(name)s.
Builder Examples¶
Select Query¶
🚧 This is a draft. This interface is subject to disruptive changes. 🚧
With the current Python features, it is not possible to write complex queries that include only specific fields or table joins in a type-safe manner.
SELECT is an experimental feature, and may be implemented in the future, but it is believed that Python Type Hint evolution is necessary for that.
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 select
from snowman.query.minify import minify
from your.database.schema import User
query, params = (
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)
).build()
expected = minify(
"""
SELECT
*
FROM
DATABASE.SCHEMA.USER
WHERE
ID = %s
AND NAME = %s
AND AGE >= %s
ORDER BY
ID DESC,
NAME ASC NULLS LAST
LIMIT %s
OFFSET %s
""",
)
assert query == expected
assert params == (1, "John Doe", 18, 1, 0)
Insert Query¶
Example
import datetime
from snowman.query import insert
from snowman.query.minify import minify
from your.database.schema import User
now = datetime.datetime.now()
query, params = (
insert.into(
User,
).values(
User(
id=1,
name="John Doe",
created_at=now,
),
)
).build()
expected = minify(
"""
INSERT INTO
DATABASE.SCHEMA.USER
(
ID,
NAME,
AGE,
CREATED_AT
)
VALUES (
%s,
%s,
%s,
%s
)
""",
)
assert query == expected
assert params == (1, "John Doe", None, now)
Update Query¶
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
import datetime
from snowman.query import update
from snowman.query.minify import minify
from your.database.schema import User
query, params = (
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)),
)
).build()
expected = minify(
"""
UPDATE
DATABASE.SCHEMA.USER
SET
NAME = %s
WHERE
NAME IN (%s)
AND AGE > %s
AND CREATED_AT >= %s
""",
)
assert query == expected
assert params == ("Jane Doe", ["Jane", "Doe"], 18, datetime.datetime(2001, 1, 1))
Delete Query¶
Example
from snowman.query import delete
from snowman.query.minify import minify
from your.database.schema import User
query, params = (
delete.from_(
User,
).where(
lambda c: c.self.id == 1,
)
).build()
expected = minify(
"""
DELETE FROM
DATABASE.SCHEMA.USER
WHERE
ID = %s
""",
)
assert query == expected
assert params == (1,)
Truncate Query¶
Where Clause¶
There are several ways to generate the WHERE clause.
For simple conditions, you can complete everything in Python, and benefit from static analysis.
The method of generating the WHERE clause using function like lambda is the easiest way to write type safe queries without importing other modules.
The argument name of the lambda function is recommended to use c (the first letter of Context).
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.group(
(c.self.age.is_.not_.null)
.and_(c.self.age >= 18)
.and_(c.self.age <= 100),
),
)
.and_(c.self.created_at >= datetime.datetime(2001, 1, 1)),
).execute(cursor)
Tip
c.self is an abbreviation for c(User).
This is useful when writing repeatedly, and can be easily written with dot chain.
If you explicitly state that it is a column of User, it will be as follows.
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(User).name.in_(["Jane", "Doe"]))
.and_(
c.group(
(c(User).age.is_.not_.null)
.and_(c(User).age >= 18)
.and_(c(User).age <= 100),
),
)
.and_(c(User).created_at >= datetime.datetime(2001, 1, 1)),
).execute(cursor)
The method using lambda expressions is easy to write expressions,
but if you write complex conditions, errors may be difficult to read.
By using Condition, you can write conditions more strictly and make it easier to track errors.
Example
import datetime
from snowman.query import update
from snowman.query.expression import column as c
from snowman.query.expression import group
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(
(c(User).name.in_(["Jane", "Doe"]))
.and_(
group(
(c(User).age.is_.not_.null)
.and_(c(User).age >= 18)
.and_(c(User).age <= 100),
),
)
.and_(c(User).created_at >= datetime.datetime(2001, 1, 1)),
).execute(cursor)
If you want to write a more complex condition, you can use string and params.
This method is useful when you want to write a condition that is difficult to write with Condition.
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(
"""
NAME IN (%s)
AND (
AGE IS NOT NULL
AND AGE >= %s
AND AGE <= %s
)
AND CREATED_AT >= %s
""",
(["Jane", "Doe"], 18, 100, datetime.datetime(2001, 1, 1)),
).execute(cursor)