use isoformat to parse str begin config

core/dbt/artifacts/resources/v1/config.py

@@ -1,6 +1,5 @@
 import re
 from dataclasses import dataclass, field
-from datetime import datetime
 from typing import Any, Dict, List, Optional, Union
 
 from mashumaro.jsonschema.annotations import Pattern
@@ -83,7 +82,7 @@ class NodeConfig(NodeAndTestConfig):
     incremental_strategy: Optional[str] = None
     batch_size: Any = None
     lookback: Any = 0
-    begin: Union[datetime, Any] = None
+    begin: Any = None
     persist_docs: Dict[str, Any] = field(default_factory=dict)
     post_hook: List[Hook] = field(
         default_factory=list,

core/dbt/parser/manifest.py

@@ -1380,6 +1380,18 @@ def check_valid_microbatch_config(self):
                         raise dbt.exceptions.ParsingError(
                             f"Microbatch model '{node.name}' must provide a 'begin' (datetime) config that indicates the earliest timestamp the microbatch model should be built from."
                         )
+
+                    # Try to cast begin to a datetime using same format as mashumaro for consistency with other yaml-provided datetimes
+                    # Mashumaro default: https://github.com/Fatal1ty/mashumaro/blob/4ac16fd060a6c651053475597b58b48f958e8c5c/README.md?plain=1#L1186
+                    if isinstance(begin, str):
+                        try:
+                            begin = datetime.datetime.fromisoformat(begin)
+                            node.config.begin = begin
+                        except Exception:
+                            raise dbt.exceptions.ParsingError(
+                                f"Microbatch model '{node.name}' must provide a 'begin' config of valid datetime (ISO format), but got: {begin}."
+                            )
+
                     if not isinstance(begin, datetime.datetime):
                         raise dbt.exceptions.ParsingError(
                             f"Microbatch model '{node.name}' must provide a 'begin' config of type datetime, but got: {type(begin)}."

tests/functional/microbatch/test_microbatch_config_validation.py

@@ -26,6 +26,17 @@
       begin: 2020-01-01
 """
 
+invalid_microbatch_model_config_yml = """
+models:
+  - name: microbatch
+    config:
+      materialized: incremental
+      incremental_strategy: microbatch
+      batch_size: day
+      event_time: event_time
+      begin: 2020-01-01 11 PM
+"""
+
 missing_event_time_microbatch_model_sql = """
 {{ config(materialized='incremental', incremental_strategy='microbatch', batch_size='day') }}
 select * from {{ ref('input_model') }}
@@ -118,7 +129,7 @@ def models(self):
         }
 
 
-class TestInvaliBeginMicrobatch(BaseMicrobatchTestParseError):
+class TestInvaliBeginTypeMicrobatch(BaseMicrobatchTestParseError):
     @pytest.fixture(scope="class")
     def models(self):
         return {
@@ -127,6 +138,16 @@ def models(self):
         }
 
 
+class TestInvaliBegiFormatMicrobatch(BaseMicrobatchTestParseError):
+    @pytest.fixture(scope="class")
+    def models(self):
+        return {
+            "input_model.sql": valid_input_model_sql,
+            "microbatch.sql": valid_microbatch_model_no_config_sql,
+            "microbatch.yml": invalid_microbatch_model_config_yml,
+        }
+
+
 class TestMissingBatchSizeMicrobatch(BaseMicrobatchTestParseError):
     @pytest.fixture(scope="class")
     def models(self):