Fix processing empty SQL and GraphQL files, refactor codegen (#108)

Author: droserasprout

src/demo_hic_et_nunc/dipdup-local.yml

@@ -1,7 +1,7 @@
 database:
   kind: postgres
   host: 127.0.0.1
-  port: 6423
+  port: 5432
   user: ${POSTGRES_USER:-dipdup}
   password: ${POSTGRES_PASSWORD:-changeme}
   database: ${POSTGRES_DB:-dipdup}

src/dipdup/codegen.py

@@ -109,7 +109,23 @@ def merge(self, other: Optional['HTTPConfig']) -> None:
 
 
 @dataclass
-class ContractConfig:
+class NameMixin:
+    def __post_init_post_parse__(self) -> None:
+        self._name: Optional[str] = None
+
+    @property
+    def name(self) -> str:
+        if self._name is None:
+            raise RuntimeError('Config is not pre-initialized')
+        return self._name
+
+    @name.setter
+    def name(self, name: str) -> None:
+        self._name = name
+
+
+@dataclass
+class ContractConfig(NameMixin):
     """Contract config
 
     :param network: Corresponding network alias, only for sanity checks
@@ -135,22 +151,6 @@ def valid_address(cls, v):
         return v
 
 
-@dataclass
-class NameMixin:
-    def __post_init_post_parse__(self) -> None:
-        self._name: Optional[str] = None
-
-    @property
-    def name(self) -> str:
-        if self._name is None:
-            raise RuntimeError('Config is not pre-initialized')
-        return self._name
-
-    @name.setter
-    def name(self, name: str) -> None:
-        self._name = name
-
-
 @dataclass
 class TzktDatasourceConfig(NameMixin):
     """TzKT datasource config
@@ -625,13 +625,13 @@ class BlockIndexConfig(IndexConfig):
 
 
 @dataclass
-class StaticTemplateConfig:
+class IndexTemplateConfig:
     kind = 'template'
     template: str
     values: Dict[str, str]
 
 
-IndexConfigT = Union[OperationIndexConfig, BigMapIndexConfig, BlockIndexConfig, StaticTemplateConfig]
+IndexConfigT = Union[OperationIndexConfig, BigMapIndexConfig, BlockIndexConfig, IndexTemplateConfig]
 IndexConfigTemplateT = Union[OperationIndexConfig, BigMapIndexConfig, BlockIndexConfig]
 HandlerPatternConfigT = Union[OperationHandlerOriginationPatternConfig, OperationHandlerTransactionPatternConfig]
 
@@ -708,7 +708,7 @@ class DipDupConfig:
     datasources: Dict[str, DatasourceConfigT]
     contracts: Dict[str, ContractConfig] = Field(default_factory=dict)
     indexes: Dict[str, IndexConfigT] = Field(default_factory=dict)
-    templates: Optional[Dict[str, IndexConfigTemplateT]] = None
+    templates: Dict[str, IndexConfigTemplateT] = Field(default_factory=dict)
     database: Union[SqliteDatabaseConfig, PostgresDatabaseConfig] = SqliteDatabaseConfig(kind='sqlite')
     hasura: Optional[HasuraConfig] = None
     jobs: Optional[Dict[str, JobConfig]] = None
@@ -774,10 +774,10 @@ def get_configure_fn(self) -> Type:
         except (ModuleNotFoundError, AttributeError) as e:
             raise HandlerImportError(module=module_name, obj=CONFIGURE_HANDLER) from e
 
-    def resolve_static_templates(self) -> None:
+    def resolve_index_templates(self) -> None:
         _logger.info('Substituting index templates')
         for index_name, index_config in self.indexes.items():
-            if isinstance(index_config, StaticTemplateConfig):
+            if isinstance(index_config, IndexTemplateConfig):
                 template = self.get_template(index_config.template)
                 raw_template = json.dumps(template, default=pydantic_encoder)
                 for key, value in index_config.values.items():
@@ -841,10 +841,12 @@ def _pre_initialize_index(self, index_name: str, index_config: IndexConfigT) ->
         self._pre_initialized.append(index_name)
 
     def pre_initialize(self) -> None:
-        for name, config in self.datasources.items():
-            config.name = name
+        for name, contract_config in self.contracts.items():
+            contract_config.name = name
+        for name, datasource_config in self.datasources.items():
+            datasource_config.name = name
 
-        self.resolve_static_templates()
+        self.resolve_index_templates()
         for index_name, index_config in self.indexes.items():
             self._pre_initialize_index(index_name, index_config)
 
@@ -942,7 +944,7 @@ def _initialize_index(self, index_name: str, index_config: IndexConfigT) -> None
         if index_name in self._initialized:
             return
 
-        if isinstance(index_config, StaticTemplateConfig):
+        if isinstance(index_config, IndexTemplateConfig):
             raise RuntimeError('Config is not pre-initialized')
 
         if isinstance(index_config, OperationIndexConfig):

src/dipdup/config.py

@@ -6,7 +6,7 @@
 from tortoise import Tortoise
 from tortoise.transactions import in_transaction
 
-from dipdup.config import ContractConfig, DipDupConfig, PostgresDatabaseConfig, StaticTemplateConfig
+from dipdup.config import ContractConfig, DipDupConfig, IndexTemplateConfig, PostgresDatabaseConfig
 from dipdup.datasources import DatasourceT
 from dipdup.exceptions import ContractAlreadyExistsError, IndexAlreadyExistsError
 from dipdup.utils import FormattedLogger
@@ -98,7 +98,7 @@ def add_index(self, name: str, template: str, values: Dict[str, Any]) -> None:
         if name in self.config.indexes:
             raise IndexAlreadyExistsError(self, name)
         self.config.get_template(template)
-        self.config.indexes[name] = StaticTemplateConfig(
+        self.config.indexes[name] = IndexTemplateConfig(
             template=template,
             values=values,
         )

src/dipdup/context.py

@@ -1,7 +1,7 @@
 import asyncio
 import hashlib
 import logging
-from contextlib import AsyncExitStack, asynccontextmanager
+from contextlib import AsyncExitStack, asynccontextmanager, suppress
 from os import listdir
 from os.path import join
 from typing import Dict, List, Optional, cast
@@ -21,9 +21,9 @@
     DatasourceConfigT,
     DipDupConfig,
     IndexConfigTemplateT,
+    IndexTemplateConfig,
     OperationIndexConfig,
     PostgresDatabaseConfig,
-    StaticTemplateConfig,
     TzktDatasourceConfig,
 )
 from dipdup.context import DipDupContext, RollbackHandlerContext
@@ -36,7 +36,7 @@
 from dipdup.hasura import HasuraGateway
 from dipdup.index import BigMapIndex, Index, OperationIndex
 from dipdup.models import BigMapData, HeadBlockData, IndexType, OperationData, State
-from dipdup.utils import FormattedLogger, slowdown, tortoise_wrapper
+from dipdup.utils import FormattedLogger, iter_files, slowdown, tortoise_wrapper
 
 INDEX_DISPATCHER_INTERVAL = 1.0
 from dipdup.scheduler import add_job, create_scheduler
@@ -83,7 +83,7 @@ async def reload_config(self) -> None:
         self._ctx.config.initialize()
 
         for index_config in self._ctx.config.indexes.values():
-            if isinstance(index_config, StaticTemplateConfig):
+            if isinstance(index_config, IndexTemplateConfig):
                 raise RuntimeError
             await self.add_index(index_config)
 
@@ -336,15 +336,11 @@ async def _execute_sql_scripts(self, reindex: bool) -> None:
         if not exists(sql_path):
             return
         self._logger.info('Executing SQL scripts from `%s`', sql_path)
-        for filename in sorted(listdir(sql_path)):
-            if not filename.endswith('.sql'):
-                continue
-
-            with open(join(sql_path, filename)) as file:
-                sql = file.read()
-
-            self._logger.info('Executing `%s`', filename)
-            await get_connection(None).execute_script(sql)
+        for file in iter_files(sql_path, '.sql'):
+            self._logger.info('Executing `%s`', file.name)
+            sql = file.read()
+            with suppress(AttributeError):
+                await get_connection(None).execute_script(sql)
 
     def _finish_migration(self, version: str) -> None:
         self._logger.warning('==================== WARNING =====================')

src/dipdup/dipdup.py

@@ -2,21 +2,19 @@
 import importlib
 import logging
 from contextlib import suppress
-from os import listdir
 from os.path import dirname, join
 from typing import Any, Dict, Iterator, List, Optional, Tuple
 
 import humps  # type: ignore
 from aiohttp import ClientConnectorError, ClientOSError
-from genericpath import exists
 from pydantic.dataclasses import dataclass
 from tortoise import fields
 from tortoise.transactions import get_connection
 
 from dipdup.config import HasuraConfig, HTTPConfig, PostgresDatabaseConfig, pascal_to_snake
 from dipdup.exceptions import ConfigurationError
 from dipdup.http import HTTPGateway
-from dipdup.utils import iter_models
+from dipdup.utils import iter_files, iter_models
 
 
 @dataclass
@@ -230,14 +228,8 @@ 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')
-        if not exists(graphql_path):
-            return
-        for filename in sorted(listdir(graphql_path)):
-            if not filename.endswith('.graphql'):
-                continue
-
-            with open(join(graphql_path, filename)) as file:
-                yield filename[:-8], file.read()
+        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 = []

src/dipdup/hasura.py

@@ -1,13 +1,16 @@
 import asyncio
 import decimal
+import errno
 import importlib
 import logging
 import pkgutil
 import time
 import types
 from contextlib import asynccontextmanager
 from logging import Logger
-from typing import Any, AsyncIterator, Dict, Iterator, List, Optional, Tuple, Type
+from os import listdir, makedirs
+from os.path import dirname, exists, getsize, join
+from typing import Any, AsyncIterator, Dict, Iterator, List, Optional, TextIO, Tuple, Type
 
 import humps  # type: ignore
 from tortoise import Tortoise
@@ -159,3 +162,47 @@ def _log(self, level, msg, args, exc_info=None, extra=None, stack_info=False, st
         if self.fmt:
             msg = self.fmt.format(msg)
         super()._log(level, msg, args, exc_info, extra, stack_info, stacklevel)
+
+
+def iter_files(path: str, ext: Optional[str] = None) -> Iterator[TextIO]:
+    """Iterate over files in a directory. Sort alphabetically, filter by extension, skip empty files."""
+    if not exists(path):
+        raise StopIteration
+    for filename in sorted(listdir(path)):
+        filepath = join(path, filename)
+        if ext and not filename.endswith(ext):
+            continue
+        if not getsize(filepath):
+            continue
+
+        with open(filepath) as file:
+            yield file
+
+
+def mkdir_p(path: str) -> None:
+    """Create directory tree, ignore if already exists"""
+    try:
+        makedirs(path)
+    except OSError as e:
+        if e.errno != errno.EEXIST:
+            raise
+
+
+def touch(path: str) -> None:
+    """Create empty file, ignore if already exists"""
+    mkdir_p(dirname(path))
+    try:
+        open(path, 'a').close()
+    except IOError as e:
+        if e.errno != errno.EEXIST:
+            raise
+
+
+def write(path: str, content: str, overwrite: bool = False) -> bool:
+    """Write content to file, create directory tree if necessary"""
+    mkdir_p(dirname(path))
+    if exists(path) and not overwrite:
+        return False
+    with open(path, 'w') as file:
+        file.write(content)
+    return True