Great question. Short answer: Prefect doesn’t “template” SQL natively, but it works well with .sql files and safe parameterization via integrations. Use parameterized queries through the Prefect integrations for your database; only use Python string interpolation for identifiers you’ve validated (e.g., table names), not for values. Recommended patterns (Prefect 3.x): - Store SQL in .sql files in your repo and read them at runtime (keeps flows clean and versionable) - Parameterize values using your DB integration (avoid f-strings for values to prevent injection) - If you need templating, use Jinja for structure (optional) but still pass values as parameters - Use Prefect Blocks for credentials and connections Links: - prefect-sqlalchemy - prefect-snowflake - prefect-gcp (BigQuery) - prefect-dbt (if you want heavy templating/macros) - Storing secrets with Blocks Code examples 1) Generic SQL (Postgres/MySQL/etc.) with SQLAlchemy - Install: pip install "prefect[sqlalchemy]" - Keep your query in sql/select_user.sql:

SELECT id, email
FROM users
WHERE id = :id;

- Flow:

from pathlib import Path
from prefect import flow
from prefect_sqlalchemy.database import SqlAlchemyConnector

@flow
def run_query(user_id: int):
    sql = Path("sql/select_user.sql").read_text()
    with SqlAlchemyConnector.load("my-postgres") as conn:
        # Safe parameterization with named placeholders
        rows = conn.fetch_many(sql, parameters={"id": user_id}, size=1000)
        for row in rows:
            print(row)

if __name__ == "__main__":
    run_query(42)

2) Snowflake - Install: pip install "prefect[snowflake]" - SQL with named placeholders:

SELECT *
FROM ORDERS
WHERE REGION = :region;

- Flow:

from pathlib import Path
from prefect import flow
from prefect_snowflake.database import SnowflakeConnector, snowflake_query

@flow
def run_snowflake(region: str):
    sql = Path("sql/orders_by_region.sql").read_text()
    connector = SnowflakeConnector.load("my-snowflake")
    rows = snowflake_query(sql, connector, params={"region": region})
    for row in rows:
        print(row)

3) BigQuery - Install: pip install "prefect[gcp]" - SQL using BigQuery parameter syntax (e.g., @country):

SELECT customer_id, country
FROM `my_project.my_dataset.customers`
WHERE country = @country;

- Flow (no params shown here; see docs link above for exact

query_params

shape):

from pathlib import Path
from prefect import flow
from prefect_gcp.credentials import GcpCredentials
from prefect_gcp.bigquery import bigquery_query

@flow
def run_bigquery():
    sql = Path("sql/customers_by_country.sql").read_text()
    gcp_creds = GcpCredentials.load("my-gcp")
    df = bigquery_query(sql, gcp_credentials=gcp_creds, to_dataframe=True)
    print(df.head())

Note:

bigquery_query

supports

query_params

for parameterized queries—see the docs for the accepted tuple structure. Optional templating with Jinja (for structure, not values) - Useful for dynamically building pieces of SQL (CTEs, column lists, feature flags) - Still pass values as parameters to your connector/task

from pathlib import Path
from jinja2 import Environment, BaseLoader

template_text = Path("sql/report.sql.j2").read_text()
template = Environment(loader=BaseLoader()).from_string(template_text)
sql = template.render(snapshot_date="2025-10-31", include_debug=False)
# Then pass `sql` to your database task and parameterize any runtime values

When to use dbt

- If you need extensive templating/macros, environment-aware SQL, lineage, and testing, use dbt and orchestrate it with Prefect via prefect-dbt. dbt handles Jinja templating natively, and Prefect handles scheduling, retries, and orchestration. Best practices - Prefer parameterized queries over Python string interpolation for values - For identifiers (e.g., table names), use a strict allowlist and insert them after validation; drivers don’t parameterize identifiers - Store credentials in Prefect Blocks and load them in flows - For large reads with SQLAlchemy, use

fetch_many(..., size=...)

to avoid loading everything in memory - Organize SQL in a

sql/

directory; consider

.sql.j2

naming for templated files If you share which database(s) you’re targeting and whether you need templating or just variable substitution, I can tailor examples for your exact setup (including BigQuery

query_params

usage if that’s your target).