import os
import snowflake.connector
from titan import Blueprint
from titan.resources import Database, Grant, Role, RoleGrant, Warehouse
connection_params = {
"account": os.environ["SNOWFLAKE_ACCOUNT"],
"user": os.environ["SNOWFLAKE_USER"],
"password": os.environ["SNOWFLAKE_PASSWORD"],
}
DEFAULT_WAREHOUSE_OPTIONS = {
"warehouse_size": "XSMALL",
"auto_suspend": 60,
"min_cluster_count": 1,
"max_cluster_count": 1,
}
def dbt():
prod_db = Database(name="ANALYTICS")
transforming_wh = Warehouse(
name="TRANSFORMING_DPG_WH",
comment="Warehouse used for transformation workloads.",
**DEFAULT_WAREHOUSE_OPTIONS,
)
transformer_role = Role(
name="TRANSFORMER_ROLE",
comment=f"Able to read from and write to {prod_db.name} database.",
)
grants = [
RoleGrant(role=transformer_role, to_role="SYSADMIN"),
Grant(priv="USAGE", on=prod_db, to=transformer_role),
]
return (
prod_db,
transforming_wh,
transformer_role,
*grants,
)
if __name__ == "__main__":
bp = Blueprint(name="setup-dbt", account=os.environ["SNOWFLAKE_ACCOUNT"])
bp.add(*dbt())
session = snowflake.connector.connect(**connection_params)
plan = bp.plan(session)
# Update snowflake to match blueprint
bp.apply(session, plan)
When I run this script, I get back the following error:
File "/Users/dpg/personal/snowflake-dbt-demo-new/.venv/lib/python3.9/site-packages/titan/identifiers.py", line 38, in <listcomp>
params = "?" + "&".join([f"{k.lower()}={v}" for k, v in self.params.items()]) if self.params else ""
TypeError: __format__ must return a str, not FQN
When I remove the lone Grant resource within my grants variable, I don't have any issues. I may also just be doing something wrong, so apologize in advance if that's the case. Appreciate any help!
I have the following code:
When I run this script, I get back the following error:
When I remove the lone
Grant
resource within mygrants
variable, I don't have any issues. I may also just be doing something wrong, so apologize in advance if that's the case. Appreciate any help!