Merge pull request #206 from developmentseed/feature/collection-base-class

add Collection Abstract BaseClass

Author: vincentsarago

CHANGES.md

@@ -8,6 +8,11 @@ Note: Minor version `0.X.0` update might break the API, It's recommended to pin
 
 ## [unreleased]
 
+## [0.11.0] - TBD
+
+* rename `tipg.collections.Collection -> tipg.collections.PgCollection`
+* add `tipg.collections.Collection` abstract base class
+
 ## [0.10.0] - 2025-02-20
 
 * convert tile bbox into collection's CRS for `MVT` where selection (author @callsumzg, https://github.com/developmentseed/tipg/pull/205)

tipg/collections.py

@@ -1,5 +1,6 @@
 """tipg.dbmodel: database events."""
 
+import abc
 import datetime
 import re
 from functools import lru_cache
@@ -158,13 +159,12 @@ class Parameter(Column):
     default: Optional[str] = None
 
 
-class Collection(BaseModel):
-    """Model for DB Table and Function."""
+class Collection(BaseModel, metaclass=abc.ABCMeta):
+    """Collection Base Class."""
 
     type: str
     id: str
     table: str
-    dbschema: str = Field(alias="schema")
     title: Optional[str] = None
     description: Optional[str] = None
     table_columns: List[Column] = []
@@ -298,6 +298,57 @@ def get_column(self, property_name: str) -> Optional[Column]:
 
         return None
 
+    @property
+    def queryables(self) -> Dict:
+        """Return the queryables."""
+        if self.geometry_columns:
+            geoms = {
+                col.name: {"$ref": geojson_schema.get(col.geometry_type.upper(), "")}
+                for col in self.geometry_columns
+            }
+        else:
+            geoms = {}
+
+        props = {
+            col.name: {"name": col.name, "type": col.json_type}
+            for col in self.properties
+            if col.name not in geoms
+        }
+
+        return {**geoms, **props}
+
+    @abc.abstractmethod
+    async def features(self, *args, **kwargs) -> ItemList:
+        """Get Items."""
+        ...
+
+    @abc.abstractmethod
+    async def get_tile(self, *args, **kwargs) -> bytes:
+        """Get MVT Tile."""
+        ...
+
+
+class CollectionList(TypedDict):
+    """Collections."""
+
+    collections: List[Collection]
+    matched: Optional[int]
+    next: Optional[int]
+    prev: Optional[int]
+
+
+class Catalog(TypedDict):
+    """Internal Collection Catalog."""
+
+    collections: Dict[str, Collection]
+    last_updated: datetime.datetime
+
+
+class PgCollection(Collection):
+    """Model for DB Table and Function."""
+
+    dbschema: str = Field(alias="schema")
+
     def _select_no_geo(self, properties: Optional[List[str]], addid: bool = True):
         nocomma = False
         columns = self.columns(properties)
@@ -858,42 +909,9 @@ async def get_tile(
         debug_query(q, *p)
 
         async with pool.acquire() as conn:
-            return await conn.fetchval(q, *p)
-
-    @property
-    def queryables(self) -> Dict:
-        """Return the queryables."""
-        if self.geometry_columns:
-            geoms = {
-                col.name: {"$ref": geojson_schema.get(col.geometry_type.upper(), "")}
-                for col in self.geometry_columns
-            }
-        else:
-            geoms = {}
+            tile = await conn.fetchval(q, *p)
 
-        props = {
-            col.name: {"name": col.name, "type": col.json_type}
-            for col in self.properties
-            if col.name not in geoms
-        }
-
-        return {**geoms, **props}
-
-
-class CollectionList(TypedDict):
-    """Collections."""
-
-    collections: List[Collection]
-    matched: Optional[int]
-    next: Optional[int]
-    prev: Optional[int]
-
-
-class Catalog(TypedDict):
-    """Internal Collection Catalog."""
-
-    collections: Dict[str, Collection]
-    last_updated: datetime.datetime
+        return bytes(tile)
 
 
 async def get_collection_index(  # noqa: C901
@@ -990,7 +1008,7 @@ async def get_collection_index(  # noqa: C901
                     if table_conf.geomcol == c["name"] or geometry_column is None:
                         geometry_column = c
 
-            catalog[table_id] = Collection(
+            catalog[table_id] = PgCollection(
                 type=table["entity"],
                 id=table_id,
                 table=table["name"],

tipg/factory.py

@@ -1606,7 +1606,7 @@ async def collection_get_tile(
                 dt=datetime_column,
             )
 
-            return Response(bytes(tile), media_type=MediaType.mvt.value)
+            return Response(tile, media_type=MediaType.mvt.value)
 
     def _tilejson_routes(self):
         ############################################################################