sqlmesh.cli.project_init

  1import typing as t
  2from enum import Enum
  3from pathlib import Path
  4from dataclasses import dataclass
  5from rich.prompt import Prompt
  6from rich.console import Console
  7from sqlmesh.integrations.dlt import generate_dlt_models_and_settings
  8from sqlmesh.utils.date import yesterday_ds
  9from sqlmesh.utils.errors import SQLMeshError
 10from sqlmesh.core.config.common import VirtualEnvironmentMode
 11
 12from sqlmesh.core.config.common import DBT_PROJECT_FILENAME
 13from sqlmesh.core.config.connection import (
 14    CONNECTION_CONFIG_TO_TYPE,
 15    DIALECT_TO_TYPE,
 16    INIT_DISPLAY_INFO_TO_TYPE,
 17)
 18
 19
 20PRIMITIVES = (str, int, bool, float)
 21
 22
 23class ProjectTemplate(Enum):
 24    DEFAULT = "default"
 25    DBT = "dbt"
 26    EMPTY = "empty"
 27    DLT = "dlt"
 28
 29
 30class InitCliMode(Enum):
 31    DEFAULT = "default"
 32    FLOW = "flow"
 33
 34
 35def _gen_config(
 36    engine_type: t.Optional[str],
 37    settings: t.Optional[str],
 38    start: t.Optional[str],
 39    template: ProjectTemplate,
 40    cli_mode: InitCliMode,
 41    dialect: t.Optional[str] = None,
 42) -> str:
 43    project_dialect = dialect or DIALECT_TO_TYPE.get(engine_type)
 44
 45    connection_settings = (
 46        settings
 47        or """      type: duckdb
 48      database: db.db"""
 49    )
 50
 51    if not settings and template != ProjectTemplate.DBT:
 52        doc_link = "https://sqlmesh.readthedocs.io/en/stable/integrations/engines{engine_link}"
 53        engine_link = ""
 54
 55        if engine_type in CONNECTION_CONFIG_TO_TYPE:
 56            required_fields = []
 57            non_required_fields = []
 58
 59            for name, field in CONNECTION_CONFIG_TO_TYPE[engine_type].model_fields.items():
 60                field_name = field.alias or name
 61
 62                default_value = field.get_default()
 63
 64                if isinstance(default_value, Enum):
 65                    default_value = default_value.value
 66                elif not isinstance(default_value, PRIMITIVES):
 67                    default_value = ""
 68
 69                required = field.is_required() or field_name == "type"
 70                option_str = f"      {'# ' if not required else ''}{field_name}: {default_value}\n"
 71
 72                # specify the DuckDB database field so quickstart runs out of the box
 73                if engine_type == "duckdb" and field_name == "database":
 74                    option_str = "      database: db.db\n"
 75                    required = True
 76
 77                if required:
 78                    required_fields.append(option_str)
 79                else:
 80                    non_required_fields.append(option_str)
 81
 82            connection_settings = "".join(required_fields + non_required_fields)
 83
 84            engine_link = f"/{engine_type}/#connection-options"
 85
 86        connection_settings = (
 87            "      # For more information on configuring the connection to your execution engine, visit:\n"
 88            "      # https://sqlmesh.readthedocs.io/en/stable/reference/configuration/#connection\n"
 89            f"      # {doc_link.format(engine_link=engine_link)}\n{connection_settings}"
 90        )
 91
 92    default_configs = {
 93        ProjectTemplate.DEFAULT: f"""# --- Gateway Connection ---
 94gateways:
 95  {engine_type}:
 96    connection:
 97{connection_settings}
 98default_gateway: {engine_type}
 99
100# --- Model Defaults ---
101# https://sqlmesh.readthedocs.io/en/stable/reference/model_configuration/#model-defaults
102
103model_defaults:
104  dialect: {project_dialect}
105  start: {start or yesterday_ds()} # Start date for backfill history
106  cron: '@daily'    # Run models daily at 12am UTC (can override per model)
107
108# --- Linting Rules ---
109# Enforce standards for your team
110# https://sqlmesh.readthedocs.io/en/stable/guides/linter/
111
112linter:
113  enabled: true
114  rules:
115    - ambiguousorinvalidcolumn
116    - invalidselectstarexpansion
117    - noambiguousprojections
118""",
119        ProjectTemplate.DBT: f"""# --- DBT-specific options ---
120dbt:
121  # This configuration ensures that each dbt target gets its own isolated state.
122  # The inferred state schemas are named "sqlmesh_state_<profile name>_<target schema>", eg "sqlmesh_state_jaffle_shop_dev"
123  # If this is undesirable, you may manually configure the gateway to use a specific state schema name
124  # https://sqlmesh.readthedocs.io/en/stable/integrations/dbt/#selecting-a-different-state-connection
125  infer_state_schema_name: True
126
127# --- Virtual Data Environment Mode ---
128# Enable Virtual Data Environments (VDE) for *development* environments.
129# Note that the production environment in dbt projects is not virtual by default to maintain compatibility with existing tooling.
130# https://sqlmesh.readthedocs.io/en/stable/guides/configuration/#virtual-data-environment-modes
131virtual_environment_mode: {VirtualEnvironmentMode.DEV_ONLY.lower()}
132
133# --- Plan Defaults ---
134# https://sqlmesh.readthedocs.io/en/stable/reference/configuration/#plan
135plan:
136  # For Virtual Data Environments, this ensures that any changes are always considered against prod,
137  # rather than the previous state of that environment
138  always_recreate_environment: True
139
140# --- Model Defaults ---
141# https://sqlmesh.readthedocs.io/en/stable/reference/model_configuration/#model-defaults
142model_defaults:
143  start: {start or yesterday_ds()}
144""",
145    }
146
147    default_configs[ProjectTemplate.EMPTY] = default_configs[ProjectTemplate.DEFAULT]
148    default_configs[ProjectTemplate.DLT] = default_configs[ProjectTemplate.DEFAULT]
149
150    flow_cli_mode = """
151# FLOW: Minimal prompts, automatic changes, summary output
152# https://sqlmesh.readthedocs.io/en/stable/reference/configuration/#plan
153
154plan:
155  no_diff: true             # Hide detailed text differences for changed models
156  no_prompts: true          # No interactive prompts
157  auto_apply: true          # Apply changes automatically
158
159# --- Optional: Set a default target environment ---
160# This is intended for local development to prevent users from accidentally applying plans to the prod environment.
161# It is a development only config and should NOT be committed to your git repo.
162# https://sqlmesh.readthedocs.io/en/stable/guides/configuration/#default-target-environment
163
164# Uncomment the following line to use a default target environment derived from the logged in user's name.
165# default_target_environment: dev_{{ user() }}
166
167# Example usage:
168# sqlmesh plan            # Automatically resolves to: sqlmesh plan dev_yourname
169# sqlmesh plan prod       # Specify `prod` to apply changes to production
170"""
171
172    return default_configs[template] + (flow_cli_mode if cli_mode == InitCliMode.FLOW else "")
173
174
175@dataclass
176class ExampleObjects:
177    sql_models: t.Dict[str, str]
178    python_models: t.Dict[str, str]
179    seeds: t.Dict[str, str]
180    audits: t.Dict[str, str]
181    tests: t.Dict[str, str]
182    sql_macros: t.Dict[str, str]
183    python_macros: t.Dict[str, str]
184
185
186def _gen_example_objects(schema_name: str) -> ExampleObjects:
187    sql_models: t.Dict[str, str] = {}
188    python_models: t.Dict[str, str] = {}
189    seeds: t.Dict[str, str] = {}
190    audits: t.Dict[str, str] = {}
191    tests: t.Dict[str, str] = {}
192    sql_macros: t.Dict[str, str] = {}
193    python_macros: t.Dict[str, str] = {"__init__": ""}
194
195    full_model_name = f"{schema_name}.full_model"
196    incremental_model_name = f"{schema_name}.incremental_model"
197    seed_model_name = f"{schema_name}.seed_model"
198
199    sql_models[full_model_name] = f"""MODEL (
200  name {full_model_name},
201  kind FULL,
202  cron '@daily',
203  grain item_id,
204  audits (assert_positive_order_ids),
205);
206
207SELECT
208  item_id,
209  COUNT(DISTINCT id) AS num_orders,
210FROM
211  {incremental_model_name}
212GROUP BY item_id
213  """
214
215    sql_models[incremental_model_name] = f"""MODEL (
216  name {incremental_model_name},
217  kind INCREMENTAL_BY_TIME_RANGE (
218    time_column event_date
219  ),
220  start '2020-01-01',
221  cron '@daily',
222  grain (id, event_date)
223);
224
225SELECT
226  id,
227  item_id,
228  event_date,
229FROM
230  {seed_model_name}
231WHERE
232  event_date BETWEEN @start_date AND @end_date
233  """
234
235    sql_models[seed_model_name] = f"""MODEL (
236  name {seed_model_name},
237  kind SEED (
238    path '../seeds/seed_data.csv'
239  ),
240  columns (
241    id INTEGER,
242    item_id INTEGER,
243    event_date DATE
244  ),
245  grain (id, event_date)
246);
247  """
248
249    seeds["seed_data"] = """id,item_id,event_date
2501,2,2020-01-01
2512,1,2020-01-01
2523,3,2020-01-03
2534,1,2020-01-04
2545,1,2020-01-05
2556,1,2020-01-06
2567,1,2020-01-07
257"""
258
259    audits["assert_positive_order_ids"] = """AUDIT (
260  name assert_positive_order_ids,
261);
262
263SELECT *
264FROM @this_model
265WHERE
266  item_id < 0
267  """
268
269    tests["test_full_model"] = f"""test_example_full_model:
270  model: {full_model_name}
271  inputs:
272    {incremental_model_name}:
273      rows:
274      - id: 1
275        item_id: 1
276      - id: 2
277        item_id: 1
278      - id: 3
279        item_id: 2
280  outputs:
281    query:
282      rows:
283      - item_id: 1
284        num_orders: 2
285      - item_id: 2
286        num_orders: 1
287  """
288
289    return ExampleObjects(
290        sql_models=sql_models,
291        python_models=python_models,
292        seeds=seeds,
293        audits=audits,
294        tests=tests,
295        python_macros=python_macros,
296        sql_macros=sql_macros,
297    )
298
299
300def init_example_project(
301    path: t.Union[str, Path],
302    engine_type: t.Optional[str],
303    dialect: t.Optional[str] = None,
304    template: ProjectTemplate = ProjectTemplate.DEFAULT,
305    pipeline: t.Optional[str] = None,
306    dlt_path: t.Optional[str] = None,
307    schema_name: str = "sqlmesh_example",
308    cli_mode: InitCliMode = InitCliMode.DEFAULT,
309    start: t.Optional[str] = None,
310) -> Path:
311    root_path = Path(path)
312
313    config_path = root_path / "config.yaml"
314    if template == ProjectTemplate.DBT:
315        # name the config file `sqlmesh.yaml` to make it clear that within the context of all
316        # the existing yaml files DBT project, this one specifically relates to configuring the sqlmesh engine
317        config_path = root_path / "sqlmesh.yaml"
318
319    audits_path = root_path / "audits"
320    macros_path = root_path / "macros"
321    models_path = root_path / "models"
322    seeds_path = root_path / "seeds"
323    tests_path = root_path / "tests"
324
325    if config_path.exists():
326        raise SQLMeshError(
327            f"Found an existing config file '{config_path}'.\n\nPlease change to another directory or remove the existing file."
328        )
329
330    if template == ProjectTemplate.DBT and not Path(root_path, DBT_PROJECT_FILENAME).exists():
331        raise SQLMeshError(
332            "Required dbt project file 'dbt_project.yml' not found in the current directory.\n\nPlease add it or change directories before running `sqlmesh init` to set up your project."
333        )
334
335    engine_types = "', '".join(CONNECTION_CONFIG_TO_TYPE)
336    if engine_type is None and template != ProjectTemplate.DBT:
337        raise SQLMeshError(
338            f"Missing `engine` argument to `sqlmesh init` - please specify a SQL engine for your project. Options: '{engine_types}'."
339        )
340
341    if engine_type and engine_type not in CONNECTION_CONFIG_TO_TYPE:
342        raise SQLMeshError(
343            f"Invalid engine '{engine_type}'. Please specify one of '{engine_types}'."
344        )
345
346    models: t.Set[t.Tuple[str, str]] = set()
347    settings = None
348    if engine_type and template == ProjectTemplate.DLT:
349        project_dialect = dialect or DIALECT_TO_TYPE.get(engine_type)
350        if pipeline and project_dialect:
351            dlt_models, settings, start = generate_dlt_models_and_settings(
352                pipeline_name=pipeline, dialect=project_dialect, dlt_path=dlt_path
353            )
354        else:
355            raise SQLMeshError(
356                "Please provide a DLT pipeline with the `--dlt-pipeline` flag to generate a SQLMesh project from DLT."
357            )
358
359    _create_config(config_path, engine_type, dialect, settings, start, template, cli_mode)
360    if template == ProjectTemplate.DBT:
361        return config_path
362
363    _create_folders([audits_path, macros_path, models_path, seeds_path, tests_path])
364
365    if template == ProjectTemplate.DLT:
366        _create_object_files(
367            models_path, {model[0].split(".")[-1]: model[1] for model in dlt_models}, "sql"
368        )
369        return config_path
370
371    example_objects = _gen_example_objects(schema_name=schema_name)
372
373    if template != ProjectTemplate.EMPTY:
374        _create_object_files(models_path, example_objects.sql_models, "sql")
375        _create_object_files(models_path, example_objects.python_models, "py")
376        _create_object_files(seeds_path, example_objects.seeds, "csv")
377        _create_object_files(audits_path, example_objects.audits, "sql")
378        _create_object_files(tests_path, example_objects.tests, "yaml")
379        _create_object_files(macros_path, example_objects.python_macros, "py")
380        _create_object_files(macros_path, example_objects.sql_macros, "sql")
381
382    return config_path
383
384
385def _create_folders(target_folders: t.Sequence[Path]) -> None:
386    for folder_path in target_folders:
387        folder_path.mkdir(exist_ok=True)
388        (folder_path / ".gitkeep").touch()
389
390
391def _create_config(
392    config_path: Path,
393    engine_type: t.Optional[str],
394    dialect: t.Optional[str],
395    settings: t.Optional[str],
396    start: t.Optional[str],
397    template: ProjectTemplate,
398    cli_mode: InitCliMode,
399) -> None:
400    project_config = _gen_config(engine_type, settings, start, template, cli_mode, dialect)
401
402    _write_file(
403        config_path,
404        project_config,
405    )
406
407
408def _create_object_files(path: Path, object_dict: t.Dict[str, str], file_extension: str) -> None:
409    for object_name, object_def in object_dict.items():
410        # file name is table component of catalog.schema.table
411        _write_file(path / f"{object_name.split('.')[-1]}.{file_extension}", object_def)
412
413
414def _write_file(path: Path, payload: str) -> None:
415    with open(path, "w", encoding="utf-8") as fd:
416        fd.write(payload)
417
418
419def interactive_init(
420    path: Path,
421    console: Console,
422    project_template: t.Optional[ProjectTemplate] = None,
423) -> t.Tuple[ProjectTemplate, t.Optional[str], t.Optional[InitCliMode]]:
424    console.print("──────────────────────────────")
425    console.print("Welcome to SQLMesh!")
426
427    project_template = _init_template_prompt(console) if not project_template else project_template
428
429    if project_template == ProjectTemplate.DBT:
430        return (project_template, None, None)
431
432    engine_type = _init_engine_prompt(console)
433    cli_mode = _init_cli_mode_prompt(console)
434
435    return (project_template, engine_type, cli_mode)
436
437
438def _init_integer_prompt(
439    console: Console, err_msg_entity: str, num_options: int, retry_func: t.Callable[[t.Any], t.Any]
440) -> int:
441    err_msg = "\nERROR: '{option_str}' is not a valid {err_msg_entity} number - please enter a number between 1 and {num_options} or exit with control+c\n"
442    while True:
443        option_str = Prompt.ask("Enter a number", console=console)
444
445        value_error = False
446        try:
447            option_num = int(option_str)
448        except ValueError:
449            value_error = True
450
451        if value_error or option_num < 1 or option_num > num_options:
452            console.print(
453                err_msg.format(
454                    option_str=option_str, err_msg_entity=err_msg_entity, num_options=num_options
455                ),
456                style="red",
457            )
458            continue
459        console.print("")
460        return option_num
461
462
463def _init_display_choices(values_dict: t.Dict[str, str], console: Console) -> t.Dict[int, str]:
464    display_num_to_value = {}
465    for i, value_str in enumerate(values_dict.keys()):
466        console.print(f"    \\[{i + 1}] {' ' if i < 9 else ''}{value_str} {values_dict[value_str]}")
467        display_num_to_value[i + 1] = value_str
468    console.print("")
469    return display_num_to_value
470
471
472def _init_template_prompt(console: Console) -> ProjectTemplate:
473    console.print("──────────────────────────────\n")
474    console.print("What type of project do you want to set up?\n")
475
476    # These are ordered for user display - do not reorder
477    template_descriptions = {
478        ProjectTemplate.DEFAULT.name: "- Create SQLMesh example project models and files",
479        ProjectTemplate.DBT.value: "    - You have an existing dbt project and want to run it with SQLMesh",
480        ProjectTemplate.EMPTY.name: "  - Create a SQLMesh configuration file and project directories only",
481    }
482
483    display_num_to_template = _init_display_choices(template_descriptions, console)
484
485    template_num = _init_integer_prompt(
486        console, "project type", len(template_descriptions), _init_template_prompt
487    )
488
489    return ProjectTemplate(display_num_to_template[template_num].lower())
490
491
492def _init_engine_prompt(console: Console) -> str:
493    console.print("──────────────────────────────\n")
494    console.print("Choose your SQL engine:\n")
495
496    # INIT_DISPLAY_INFO_TO_TYPE is a dict of {engine_type: (display_order, display_name)}
497    DISPLAY_NAME_TO_TYPE = {v[1]: k for k, v in INIT_DISPLAY_INFO_TO_TYPE.items()}
498    ordered_engine_display_names = {
499        info[1]: "" for info in sorted(INIT_DISPLAY_INFO_TO_TYPE.values(), key=lambda x: x[0])
500    }
501    display_num_to_display_name = _init_display_choices(ordered_engine_display_names, console)
502
503    engine_num = _init_integer_prompt(
504        console, "engine", len(ordered_engine_display_names), _init_engine_prompt
505    )
506
507    return DISPLAY_NAME_TO_TYPE[display_num_to_display_name[engine_num]]
508
509
510def _init_cli_mode_prompt(console: Console) -> InitCliMode:
511    console.print("──────────────────────────────\n")
512    console.print("Choose your SQLMesh CLI experience:\n")
513
514    cli_mode_descriptions = {
515        InitCliMode.DEFAULT.name: "- See and control every detail",
516        InitCliMode.FLOW.name: "   - Automatically run changes and show summary output",
517    }
518
519    display_num_to_cli_mode = _init_display_choices(cli_mode_descriptions, console)
520
521    cli_mode_num = _init_integer_prompt(
522        console, "config", len(cli_mode_descriptions), _init_cli_mode_prompt
523    )
524
525    return InitCliMode(display_num_to_cli_mode[cli_mode_num].lower())