Chore: Remove optional snapshot_is from get_snapshots() API

Author: VaggelisD

sqlmesh/core/state_sync/base.py

@@ -86,14 +86,12 @@ class StateReader(abc.ABC):
 
     @abc.abstractmethod
     def get_snapshots(
-        self,
-        snapshot_ids: t.Optional[t.Iterable[SnapshotIdLike]],
+        self, snapshot_ids: t.Iterable[SnapshotIdLike]
     ) -> t.Dict[SnapshotId, Snapshot]:
         """Bulk fetch snapshots given the corresponding snapshot ids.
 
         Args:
-            snapshot_ids: Iterable of snapshot ids to get. If not provided all
-                available snapshots will be returned.
+            snapshot_ids: Iterable of snapshot ids to get.
 
         Returns:
             A dictionary of snapshot ids to snapshots for ones that could be found.

sqlmesh/core/state_sync/cache.py

@@ -46,11 +46,8 @@ def _from_cache(
         return snapshot
 
     def get_snapshots(
-        self, snapshot_ids: t.Optional[t.Iterable[SnapshotIdLike]]
+        self, snapshot_ids: t.Iterable[SnapshotIdLike]
     ) -> t.Dict[SnapshotId, Snapshot]:
-        if snapshot_ids is None:
-            return self.state_sync.get_snapshots(snapshot_ids)
-
         existing = {}
         missing = set()
         now = now_timestamp()

sqlmesh/core/state_sync/db/facade.py

@@ -365,7 +365,7 @@ def get_environments_summary(self) -> t.List[EnvironmentSummary]:
 
     def get_snapshots(
         self,
-        snapshot_ids: t.Optional[t.Iterable[SnapshotIdLike]],
+        snapshot_ids: t.Iterable[SnapshotIdLike],
     ) -> t.Dict[SnapshotId, Snapshot]:
         """Fetches snapshots from the state.
 

sqlmesh/core/state_sync/db/snapshot.py

@@ -35,7 +35,6 @@
 )
 from sqlmesh.utils.migration import index_text_type, blob_text_type
 from sqlmesh.utils.date import now_timestamp, TimeLike, now, to_timestamp
-from sqlmesh.utils.errors import SQLMeshError
 from sqlmesh.utils.pydantic import PydanticModel
 from sqlmesh.utils import unique
 
@@ -309,7 +308,7 @@ def touch_snapshots(self, snapshot_ids: t.Iterable[SnapshotIdLike]) -> None:
 
     def get_snapshots(
         self,
-        snapshot_ids: t.Optional[t.Iterable[SnapshotIdLike]],
+        snapshot_ids: t.Iterable[SnapshotIdLike],
     ) -> t.Dict[SnapshotId, Snapshot]:
         """Fetches snapshots.
 
@@ -319,8 +318,6 @@ def get_snapshots(
         Returns:
             A dictionary of snapshot IDs to snapshots.
         """
-        if snapshot_ids is None:
-            raise SQLMeshError("Must provide snapshot IDs to fetch snapshots.")
         return self._get_snapshots(snapshot_ids)
 
     def snapshots_exist(self, snapshot_ids: t.Iterable[SnapshotIdLike]) -> t.Set[SnapshotId]: