Remove schema prefix from Hasura root fields, add integration test (#123)

Author: droserasprout

poetry.lock

@@ -52,6 +52,7 @@ pytest = "^3.0"
 pytest-cov = "^2.4"
 flake8 = "3.9.0"
 flakehell = "^0.9.0"
+testcontainers = "^3.4.1"
 
 [tool.poetry.extras]
 pytezos = ["pytezos"]

pyproject.toml

@@ -71,9 +71,9 @@ class PostgresDatabaseConfig:
 
     kind: Literal['postgres']
     host: str
-    port: int
-    user: str
-    database: str
+    user: str = 'postgres'
+    database: str = 'postgres'
+    port: int = 5432
     schema_name: str = 'public'
     password: str = ''
     immune_tables: Optional[List[str]] = None

src/dipdup/config.py

@@ -16,7 +16,7 @@
 from dipdup.config import HasuraConfig, HTTPConfig, PostgresDatabaseConfig
 from dipdup.exceptions import ConfigurationError
 from dipdup.http import HTTPGateway
-from dipdup.utils import iter_files, iter_models, pascal_to_snake
+from dipdup.utils import iter_files, iter_models, pascal_to_snake, remove_prefix
 
 _get_fields_query = '''
 query introspectionQuery($name: String!) {
@@ -47,7 +47,7 @@
 @dataclass
 class Field:
     name: str
-    type: Optional[str]
+    type: Optional[str] = None
 
     def camelize(self) -> 'Field':
         return Field(
@@ -57,8 +57,8 @@ def camelize(self) -> 'Field':
 
     @property
     def root(self) -> str:
-        # NOTE: Hasura omits default schema name in root field name
-        return humps.decamelize(self.name).lstrip('public_')
+        # NOTE: Hasura omits schema name prefix in root fields
+        return remove_prefix(humps.decamelize(self.name), 'public')
 
 
 class HasuraError(RuntimeError):
@@ -94,16 +94,15 @@ async def configure(self) -> None:
         metadata = await self._fetch_metadata()
 
         # NOTE: Hasura metadata updated in three steps.
-        # NOTE: Order matters because queries must be generated after applying camelcase to model names.
+        # NOTE: Order matters because queries must be generated after applying table customization to be valid.
         # NOTE: 1. Generate and apply tables metadata.
         source_tables_metadata = await self._generate_source_tables_metadata()
         metadata['sources'][0]['tables'] = source_tables_metadata
         await self._replace_metadata(metadata)
 
-        # NOTE: 2. Apply camelcase and refresh metadata
-        if self._hasura_config.camel_case:
-            await self._apply_camelcase()
-            metadata = await self._fetch_metadata()
+        # NOTE: 2. Apply table customization and refresh metadata
+        await self._apply_table_customization()
+        metadata = await self._fetch_metadata()
 
         # NOTE: 3. Generate and apply queries and rest endpoints
         query_collections_metadata = await self._generate_query_collections_metadata()
@@ -150,6 +149,17 @@ async def _healthcheck(self) -> None:
         else:
             raise HasuraError(f'Hasura instance not responding for {self._hasura_config.connection_timeout} seconds')
 
+        version_json = await (
+            await self._http._session.get(
+                f'{self._hasura_config.url}/v1/version',
+            )
+        ).json()
+        version = version_json['version']
+        if version.startswith('v1'):
+            raise HasuraError('Hasura v1 is not supported.')
+
+        self._logger.info('Connected to Hasura %s', version)
+
     async def _reset_metadata(self) -> None:
         self._logger.info('Resetting metadata')
         await self._hasura_request(
@@ -191,6 +201,13 @@ async def _get_views(self) -> List[str]:
         self._logger.info('Found %s regular and materialized views', len(views))
         return views
 
+    def _iterate_graphql_queries(self) -> Iterator[Tuple[str, str]]:
+        package = importlib.import_module(self._package)
+        package_path = dirname(package.__file__)
+        graphql_path = join(package_path, 'graphql')
+        for file in iter_files(graphql_path, '.graphql'):
+            yield file.name.split('/')[-1][:-8], file.read()
+
     async def _generate_source_tables_metadata(self) -> List[Dict[str, Any]]:
         """Generate source tables metadata based on project models and views.
 
@@ -235,13 +252,6 @@ async def _generate_source_tables_metadata(self) -> List[Dict[str, Any]]:
 
         return list(metadata_tables.values())
 
-    def _iterate_graphql_queries(self) -> Iterator[Tuple[str, str]]:
-        package = importlib.import_module(self._package)
-        package_path = dirname(package.__file__)
-        graphql_path = join(package_path, 'graphql')
-        for file in iter_files(graphql_path, '.graphql'):
-            yield file.name.split('/')[-1][:-8], file.read()
-
     async def _generate_query_collections_metadata(self) -> List[Dict[str, Any]]:
         queries = []
         for _, model in iter_models(self._package):
@@ -283,8 +293,7 @@ async def _get_fields_json(self, name: str) -> List[Dict[str, Any]]:
             raise HasuraError(f'Unknown table `{name}`') from e
 
     async def _get_fields(self, name: str = 'query_root') -> List[Field]:
-        # NOTE: Hasura omits default schema name
-        name = name.lstrip('public_')
+        name = Field(name).root
 
         try:
             fields_json = await self._get_fields_json(name)
@@ -320,17 +329,17 @@ async def _get_fields(self, name: str = 'query_root') -> List[Field]:
 
         return fields
 
-    async def _apply_camelcase(self) -> None:
+    async def _apply_table_customization(self) -> None:
         """Convert table and column names to camelCase.
 
         Based on https://github.com/m-rgba/hasura-snake-to-camel
         """
-        self._logger.info('Converting field names to camelCase')
 
         tables = await self._get_fields()
 
+        # TODO: Bulk request
         for table in tables:
-            custom_root_fields = self._format_custom_root_fields(table.root)
+            custom_root_fields = self._format_custom_root_fields(table)
             columns = await self._get_fields(table.root)
             custom_column_names = self._format_custom_column_names(columns)
             args: Dict[str, Any] = {
@@ -343,6 +352,7 @@ async def _apply_camelcase(self) -> None:
                 },
             }
 
+            self._logger.info('Applying `%s` table customization', table.root)
             await self._hasura_request(
                 endpoint='metadata',
                 json={
@@ -388,22 +398,32 @@ def _format_rest_endpoint(self, query_name: str) -> Dict[str, Any]:
             "comment": None,
         }
 
-    def _format_custom_root_fields(self, table: str) -> Dict[str, Any]:
+    def _format_custom_root_fields(self, table: Field) -> Dict[str, Any]:
+        table_name = remove_prefix(table.root, self._database_config.schema_name)
+
+        def _fmt(fmt: str) -> str:
+            if self._hasura_config.camel_case:
+                return humps.camelize(fmt.format(table_name))
+            return humps.decamelize(fmt.format(table_name))
+
         # NOTE: Do not change original Hasura format, REST endpoints generation will be broken otherwise
         return {
-            'select': humps.camelize(table),
-            'select_by_pk': humps.camelize(f'{table}_by_pk'),
-            'select_aggregate': humps.camelize(f'{table}_aggregate'),
-            'insert': humps.camelize(f'insert_{table}'),
-            'insert_one': humps.camelize(f'insert_{table}_one'),
-            'update': humps.camelize(f'update_{table}'),
-            'update_by_pk': humps.camelize(f'update_{table}_by_pk'),
-            'delete': humps.camelize(f'delete_{table}'),
-            'delete_by_pk': humps.camelize(f'delete_{table}_by_pk'),
+            'select': _fmt('{}'),
+            'select_by_pk': _fmt('{}_by_pk'),
+            'select_aggregate': _fmt('{}_aggregate'),
+            'insert': _fmt('insert_{}'),
+            'insert_one': _fmt('insert_{}_one'),
+            'update': _fmt('update_{}'),
+            'update_by_pk': _fmt('update_{}_by_pk'),
+            'delete': _fmt('delete_{}'),
+            'delete_by_pk': _fmt('delete_{}_by_pk'),
         }
 
     def _format_custom_column_names(self, fields: List[Field]) -> Dict[str, Any]:
-        return {humps.decamelize(f.name): humps.camelize(f.name) for f in fields}
+        if self._hasura_config.camel_case:
+            return {humps.decamelize(f.name): humps.camelize(f.name) for f in fields}
+        else:
+            return {humps.decamelize(f.name): humps.decamelize(f.name) for f in fields}
 
     def _format_table(self, name: str) -> Dict[str, Any]:
         return {
@@ -418,8 +438,7 @@ def _format_table(self, name: str) -> Dict[str, Any]:
     def _format_table_table(self, name: str) -> Dict[str, Any]:
         return {
             "schema": self._database_config.schema_name,
-            # NOTE: Remove schema prefix from table name
-            'name': name.replace(self._database_config.schema_name, '').strip('_'),
+            'name': remove_prefix(name, self._database_config.schema_name),
         }
 
     def _format_array_relationship(

src/dipdup/hasura.py

@@ -79,7 +79,7 @@ async def tortoise_wrapper(url: str, models: Optional[str] = None) -> AsyncItera
                     db_url=url,
                     modules=modules,  # type: ignore
                 )
-            except ConnectionRefusedError:
+            except (OSError, ConnectionRefusedError):
                 _logger.warning('Can\'t establish database connection, attempt %s/%s', attempt, attempts)
                 if attempt == attempts - 1:
                     raise
@@ -232,3 +232,10 @@ def import_from(module: str, obj: str) -> Any:
         return getattr(importlib.import_module(module), obj)
     except (ImportError, AttributeError) as e:
         raise HandlerImportError(module, obj) from e
+
+
+def remove_prefix(text: str, prefix: str) -> str:
+    """Remove prefix and strip underscores"""
+    if text.startswith(prefix):
+        text = text[len(prefix) :]
+    return text.strip('_')