@@ -1680,7 +1680,12 @@ def create_declarative_stream(
1680
1680
1681
1681
def _build_stream_slicer_from_partition_router (
1682
1682
self ,
1683
- model : Union [AsyncRetrieverModel , CustomRetrieverModel , SimpleRetrieverModel , StateDelegatingRetrieverModel ],
1683
+ model : Union [
1684
+ AsyncRetrieverModel ,
1685
+ CustomRetrieverModel ,
1686
+ SimpleRetrieverModel ,
1687
+ StateDelegatingRetrieverModel ,
1688
+ ],
1684
1689
config : Config ,
1685
1690
) -> Optional [PartitionRouter ]:
1686
1691
if (
@@ -1726,7 +1731,10 @@ def _build_incremental_cursor(
1726
1731
cursor_component = self ._create_component_from_model (
1727
1732
model = incremental_sync_model , config = config
1728
1733
)
1729
- is_global_cursor = hasattr (incremental_sync_model , "global_substream_cursor" ) and incremental_sync_model .global_substream_cursor
1734
+ is_global_cursor = (
1735
+ hasattr (incremental_sync_model , "global_substream_cursor" )
1736
+ and incremental_sync_model .global_substream_cursor
1737
+ )
1730
1738
1731
1739
if is_global_cursor :
1732
1740
return GlobalSubstreamCursor (
@@ -1756,7 +1764,12 @@ def _build_incremental_cursor(
1756
1764
1757
1765
def _build_resumable_cursor (
1758
1766
self ,
1759
- model : Union [AsyncRetrieverModel , CustomRetrieverModel , SimpleRetrieverModel , StateDelegatingRetrieverModel ],
1767
+ model : Union [
1768
+ AsyncRetrieverModel ,
1769
+ CustomRetrieverModel ,
1770
+ SimpleRetrieverModel ,
1771
+ StateDelegatingRetrieverModel ,
1772
+ ],
1760
1773
stream_slicer : Optional [PartitionRouter ],
1761
1774
) -> Optional [StreamSlicer ]:
1762
1775
if hasattr (model , "paginator" ) and model .paginator and not stream_slicer :
@@ -1778,12 +1791,17 @@ def _merge_stream_slicers(
1778
1791
if model .retriever .type == "StateDelegatingRetriever" and not model .incremental_sync :
1779
1792
raise ValueError ("StateDelegatingRetriever requires 'incremental_sync' to be enabled." )
1780
1793
1781
- if model .retriever .type == "AsyncRetriever" and model .incremental_sync .type != "DatetimeBasedCursor" :
1794
+ if (
1795
+ model .retriever .type == "AsyncRetriever"
1796
+ and model .incremental_sync .type != "DatetimeBasedCursor"
1797
+ ):
1782
1798
# We are currently in a transition to the Concurrent CDK and AsyncRetriever can only work with the
1783
1799
# support or unordered slices (for example, when we trigger reports for January and February, the report
1784
1800
# in February can be completed first). Once we have support for custom concurrent cursor or have a new
1785
1801
# implementation available in the CDK, we can enable more cursors here.
1786
- raise ValueError ("AsyncRetriever with cursor other than DatetimeBasedCursor is not supported yet." )
1802
+ raise ValueError (
1803
+ "AsyncRetriever with cursor other than DatetimeBasedCursor is not supported yet."
1804
+ )
1787
1805
1788
1806
if model .retriever .type == "AsyncRetriever" and model .retriever .partition_router :
1789
1807
# Note that this development is also done in parallel to the per partition development which once merged
@@ -2059,9 +2077,7 @@ def create_dynamic_schema_loader(
2059
2077
self , model : DynamicSchemaLoaderModel , config : Config , ** kwargs : Any
2060
2078
) -> DynamicSchemaLoader :
2061
2079
stream_slicer = self ._build_stream_slicer_from_partition_router (model .retriever , config )
2062
- combined_slicers = self ._build_resumable_cursor (
2063
- model .retriever , stream_slicer
2064
- )
2080
+ combined_slicers = self ._build_resumable_cursor (model .retriever , stream_slicer )
2065
2081
2066
2082
schema_transformations = []
2067
2083
if model .schema_transformations :
@@ -2916,9 +2932,7 @@ def create_http_components_resolver(
2916
2932
self , model : HttpComponentsResolverModel , config : Config
2917
2933
) -> Any :
2918
2934
stream_slicer = self ._build_stream_slicer_from_partition_router (model .retriever , config )
2919
- combined_slicers = self ._build_resumable_cursor (
2920
- model .retriever , stream_slicer
2921
- )
2935
+ combined_slicers = self ._build_resumable_cursor (model .retriever , stream_slicer )
2922
2936
2923
2937
retriever = self ._create_component_from_model (
2924
2938
model = model .retriever ,
0 commit comments