diff --git a/dbt_meshify/cli.py b/dbt_meshify/cli.py
index cde14c3..46b40b7 100644
--- a/dbt_meshify/cli.py
+++ b/dbt_meshify/cli.py
@@ -68,6 +68,14 @@
     help="Additional properties to assign to a group Owner.",
 )
 
+read_catalog = click.option(
+    "--read-catalog",
+    "-r",
+    is_flag=True,
+    envvar="DBT_MESHIFY_READ_CATALOG",
+    help="Skips the dbt docs generate step and reads the local catalog.json file.",
+)
+
 
 def owner(func):
     """Add click options and argument validation for creating Owner objects."""
diff --git a/dbt_meshify/dbt_projects.py b/dbt_meshify/dbt_projects.py
index cf92c13..6a98fb1 100644
--- a/dbt_meshify/dbt_projects.py
+++ b/dbt_meshify/dbt_projects.py
@@ -1,7 +1,6 @@
 import copy
 import hashlib
 import json
-import logging
 import os
 from pathlib import Path
 from typing import Any, Dict, MutableMapping, Optional, Set, Union
@@ -22,6 +21,7 @@
 from dbt.contracts.results import CatalogArtifact, CatalogTable
 from dbt.graph import Graph
 from dbt.node_types import NodeType
+from loguru import logger
 
 from dbt_meshify.dbt import Dbt
 from dbt_meshify.storage.file_content_editors import (
@@ -30,8 +30,6 @@
 )
 from dbt_meshify.storage.file_manager import DbtFileManager
 
-logger = logging.getLogger()
-
 
 class BaseDbtProject:
     """A base-level representation of a dbt project."""
@@ -190,15 +188,29 @@ def _load_project(path) -> Project:
         return Project.from_dict(project_dict)
 
     @classmethod
-    def from_directory(cls, directory: os.PathLike) -> "DbtProject":
+    def from_directory(cls, directory: os.PathLike, read_catalog: bool) -> "DbtProject":
         """Create a new DbtProject using a dbt project directory"""
 
         dbt = Dbt()
+        project = cls._load_project(directory)
+
+        def get_catalog(directory: os.PathLike, read_catalog: bool) -> CatalogArtifact:
+            catalog_path = Path(directory) / (project.target_path or "target") / "catalog.json"
+            if read_catalog:
+                logger.info(f"Reading catalog from {catalog_path}")
+                try:
+                    catalog_dict = json.loads(catalog_path.read_text())
+                    return CatalogArtifact.from_dict(catalog_dict)
+                except FileNotFoundError:
+                    logger.info(f"Catalog not found at {catalog_path}, running dbt docs generate")
+                    return dbt.docs_generate(directory)
+            else:
+                return dbt.docs_generate(directory)
 
         return DbtProject(
             manifest=dbt.parse(directory),
-            project=cls._load_project(directory),
-            catalog=dbt.docs_generate(directory),
+            project=project,
+            catalog=get_catalog(directory, read_catalog),
             dbt=dbt,
             path=Path(directory),
         )
diff --git a/dbt_meshify/main.py b/dbt_meshify/main.py
index fad4204..c25ada3 100644
--- a/dbt_meshify/main.py
+++ b/dbt_meshify/main.py
@@ -19,6 +19,7 @@
     owner_name,
     owner_properties,
     project_path,
+    read_catalog,
     select,
     selector,
 )
@@ -71,7 +72,7 @@ def connect(projects_dir):
             break
 
         path = Path(path_string).expanduser().resolve()
-        project = DbtProject.from_directory(path)
+        project = DbtProject.from_directory(path, read_catalog)
         holder.register_project(project)
 
     print(holder.project_map())
@@ -82,16 +83,16 @@ def connect(projects_dir):
 @click.argument("project_name")
 @exclude
 @project_path
+@read_catalog
 @select
 @selector
-def split(project_name, select, exclude, project_path, selector, create_path):
+def split(project_name, select, exclude, project_path, selector, create_path, read_catalog):
     """
     Splits out a new subproject from a dbt project by adding all necessary dbt Mesh constructs to the resources based on the selected resources.
 
     """
-
     path = Path(project_path).expanduser().resolve()
-    project = DbtProject.from_directory(path)
+    project = DbtProject.from_directory(path, read_catalog)
 
     subproject = project.split(
         project_name=project_name, select=select, exclude=exclude, selector=selector
@@ -112,15 +113,16 @@ def split(project_name, select, exclude, project_path, selector, create_path):
 @operation.command(name="add-contract")
 @exclude
 @project_path
+@read_catalog
 @select
 @selector
-def add_contract(select, exclude, project_path, selector, public_only=False):
+def add_contract(select, exclude, project_path, selector, read_catalog, public_only=False):
     """
     Adds a contract to all selected models.
     """
     path = Path(project_path).expanduser().resolve()
     logger.info(f"Reading dbt project at {path}")
-    project = DbtProject.from_directory(path)
+    project = DbtProject.from_directory(path, read_catalog)
     resources = list(
         project.select_resources(
             select=select, exclude=exclude, selector=selector, output_key="unique_id"
@@ -149,17 +151,19 @@ def add_contract(select, exclude, project_path, selector, public_only=False):
 @operation.command(name="add-version")
 @exclude
 @project_path
+@read_catalog
 @select
 @selector
 @click.option("--prerelease", "--pre", default=False, is_flag=True)
 @click.option("--defined-in", default=None)
-def add_version(select, exclude, project_path, selector, prerelease, defined_in):
+def add_version(select, exclude, project_path, selector, prerelease, defined_in, read_catalog):
     """
     Adds/increments model versions for all selected models.
     """
     path = Path(project_path).expanduser().resolve()
+
     logger.info(f"Reading dbt project at {path}")
-    project = DbtProject.from_directory(path)
+    project = DbtProject.from_directory(path, read_catalog)
     resources = list(
         project.select_resources(
             select=select, exclude=exclude, selector=selector, output_key="unique_id"
@@ -182,21 +186,23 @@ def add_version(select, exclude, project_path, selector, prerelease, defined_in)
 
 
 @operation.command(name="create-group")
+@click.argument("name")
 @exclude
+@group_yml_path
+@owner
+@owner_email
+@owner_name
+@owner_properties
 @project_path
+@read_catalog
 @select
 @selector
-@click.argument("name")
-@owner_name
-@owner_email
-@owner_properties
-@owner
-@group_yml_path
 def create_group(
     name,
     project_path: os.PathLike,
     group_yml_path: os.PathLike,
     select: str,
+    read_catalog: bool,
     owner_name: Optional[str] = None,
     owner_email: Optional[str] = None,
     owner_properties: Optional[str] = None,
@@ -210,7 +216,7 @@ def create_group(
 
     path = Path(project_path).expanduser().resolve()
     logger.info(f"Reading dbt project at {path}")
-    project = DbtProject.from_directory(path)
+    project = DbtProject.from_directory(path, read_catalog)
 
     if group_yml_path is None:
         group_yml_path = (path / Path("models/_groups.yml")).resolve()
@@ -243,16 +249,17 @@ def create_group(
 
 
 @cli.command(name="group")
+@click.argument("name")
 @exclude
+@group_yml_path
+@owner
+@owner_email
+@owner_name
+@owner_properties
 @project_path
+@read_catalog
 @select
 @selector
-@click.argument("name")
-@owner_name
-@owner_email
-@owner_properties
-@owner
-@group_yml_path
 @click.pass_context
 def group(
     ctx,
@@ -260,6 +267,7 @@ def group(
     project_path: os.PathLike,
     group_yml_path: os.PathLike,
     select: str,
+    read_catalog: bool,
     owner_name: Optional[str] = None,
     owner_email: Optional[str] = None,
     owner_properties: Optional[str] = None,
diff --git a/test-projects/split/split_proj/models/staging/__models.yml b/test-projects/split/split_proj/models/staging/__models.yml
index 2e35e94..2f0cf02 100644
--- a/test-projects/split/split_proj/models/staging/__models.yml
+++ b/test-projects/split/split_proj/models/staging/__models.yml
@@ -37,6 +37,17 @@ models:
           - not_null
           - unique
 
+        data_type: varchar
+      - name: location_id
+        data_type: varchar
+      - name: customer_id
+        data_type: varchar
+      - name: order_total
+        data_type: float
+      - name: tax_paid
+        data_type: float
+      - name: ordered_at
+        data_type: timestamp
   - name: stg_products
     description: Product (food and drink items that can be ordered) data with basic cleaning and transformation applied, one row per product.
     columns:
diff --git a/tests/conftest.py b/tests/conftest.py
new file mode 100644
index 0000000..419be87
--- /dev/null
+++ b/tests/conftest.py
@@ -0,0 +1,20 @@
+import pytest
+from _pytest.logging import LogCaptureFixture
+from loguru import logger
+
+
+@pytest.fixture
+def caplog(caplog: LogCaptureFixture):
+    """
+    code taken from loguru docs site
+    https://loguru.readthedocs.io/en/stable/resources/migration.html#replacing-caplog-fixture-from-pytest-library
+    """
+    handler_id = logger.add(
+        caplog.handler,
+        format="{message}",
+        level=0,
+        filter=lambda record: record["level"].no >= caplog.handler.level,
+        enqueue=False,  # Set to 'True' if your test is spawning child processes.
+    )
+    yield caplog
+    logger.remove(handler_id)
diff --git a/tests/integration/test_contract_command.py b/tests/integration/test_contract_command.py
index d988e93..cf39a1a 100644
--- a/tests/integration/test_contract_command.py
+++ b/tests/integration/test_contract_command.py
@@ -58,3 +58,43 @@ def test_add_contract_to_yml(start_yml, end_yml):
         actual = yaml.safe_load(f)
     yml_file.unlink()
     assert actual == yaml.safe_load(end_yml)
+
+
+@pytest.mark.parametrize(
+    "start_yml,end_yml,read_catalog",
+    [
+        (model_yml_no_col_no_version, expected_contract_yml_no_col, True),
+        (model_yml_no_col_no_version, expected_contract_yml_no_col, False),
+    ],
+    ids=["1", "2"],
+)
+def test_add_contract_read_catalog(start_yml, end_yml, read_catalog, caplog):
+    yml_file = proj_path / "models" / "_models.yml"
+    yml_file.parent.mkdir(parents=True, exist_ok=True)
+    runner = CliRunner()
+    # do a dbt run to create the duckdb
+    # and enable the docs generate command to work
+    dbt.invoke(directory=proj_path, runner_args=["run"])
+    # only create file if start_yml is not None
+    # in situations where models don't have a patch path, there isn't a yml file to read from
+    if start_yml:
+        yml_file.touch()
+        start_yml_content = yaml.safe_load(start_yml)
+        with open(yml_file, "w+") as f:
+            yaml.safe_dump(start_yml_content, f, sort_keys=False)
+    args = ["--select", "shared_model", "--project-path", proj_path_string]
+    if read_catalog:
+        args.append("--read-catalog")
+    result = runner.invoke(add_contract, args)
+    assert result.exit_code == 0
+    # reset the read path to the default in the logic
+    with open(yml_file, "r") as f:
+        actual = yaml.safe_load(f)
+    yml_file.unlink()
+    assert actual == yaml.safe_load(end_yml)
+    if read_catalog:
+        assert "Reading catalog from" in caplog.text
+        assert "Generating catalog with dbt docs generate..." not in caplog.text
+    else:
+        assert "Reading catalog from" not in caplog.text
+        assert "Generating catalog with dbt docs generate..." in caplog.text
diff --git a/tests/integration/test_dependency_detection.py b/tests/integration/test_dependency_detection.py
index 8acf00b..1aa749e 100644
--- a/tests/integration/test_dependency_detection.py
+++ b/tests/integration/test_dependency_detection.py
@@ -13,12 +13,16 @@ class TestLinkerSourceDependencies:
     @pytest.fixture
     def src_proj_a(self) -> DbtProject:
         """Load the `src_proj_a` project."""
-        return DbtProject.from_directory(Path("test-projects/source-hack/src_proj_a/").resolve())
+        return DbtProject.from_directory(
+            Path("test-projects/source-hack/src_proj_a/").resolve(), read_catalog=False
+        )
 
     @pytest.fixture
     def src_proj_b(self) -> DbtProject:
         """Load the `src_proj_b` project."""
-        return DbtProject.from_directory(Path("test-projects/source-hack/src_proj_b/").resolve())
+        return DbtProject.from_directory(
+            Path("test-projects/source-hack/src_proj_b/").resolve(), read_catalog=False
+        )
 
     @pytest.fixture
     def dest_proj_a(self) -> DbtProject:
@@ -29,12 +33,14 @@ def dest_proj_a(self) -> DbtProject:
         dbt = Dbt()
         dbt.invoke(path, ["deps"])
 
-        return DbtProject.from_directory(path)
+        return DbtProject.from_directory(path, read_catalog=False)
 
     @pytest.fixture
     def dest_proj_b(self) -> DbtProject:
         """Load the `dest_proj_b` project."""
-        return DbtProject.from_directory(Path("test-projects/source-hack/dest_proj_b/").resolve())
+        return DbtProject.from_directory(
+            Path("test-projects/source-hack/dest_proj_b/").resolve(), read_catalog=False
+        )
 
     def test_linker_detects_source_dependencies(self, src_proj_a, src_proj_b):
         """Verify that Linker detects source-hacked projects."""
diff --git a/tests/integration/test_group_command.py b/tests/integration/test_group_command.py
index dcefb1c..b5b2e6f 100644
--- a/tests/integration/test_group_command.py
+++ b/tests/integration/test_group_command.py
@@ -44,7 +44,7 @@ def test_group_command(select, expected_public_contracted_models):
         ],
     )
     assert result.exit_code == 0
-    project = DbtProject.from_directory(proj_path)
+    project = DbtProject.from_directory(proj_path, read_catalog=False)
     # ensure that the correct set of public models is created
     public_contracted_models = [
         model.name
diff --git a/tests/integration/test_subproject_creator.py b/tests/integration/test_subproject_creator.py
index 80a576d..22fc3a1 100644
--- a/tests/integration/test_subproject_creator.py
+++ b/tests/integration/test_subproject_creator.py
@@ -44,7 +44,7 @@ def setup_new_project(write_packages_yml: bool = False):
 
 
 def split_project(select: str = "my_first_dbt_model"):
-    project = DbtProject.from_directory(Path("test"))
+    project = DbtProject.from_directory(Path("test"), read_catalog=False)
     subproject = project.split(project_name='subdir', select=select)
     return subproject