Heartbeat REST endpoint (#211)

* Fix processing modified storage models

* --force flag for hasura configure

* Fix schema init, apply builtin scripts

* comment

* Add built-in rest endpoint

* Renamings, filter head status by name

* Changelog

* Fix view condition

Author: droserasprout

CHANGELOG.md

@@ -7,7 +7,9 @@ Please use [this](https://docs.gitlab.com/ee/development/changelog.html) documen
 ### Added
 
 * cli: Added `schema init` command to initialize database schema.
+* cli: Added `--force` flag to `hasura configure` command.
 * codegen: Added support for subpackages inside callback directories.
+* hasura: Added `dipdup_head_status` view and REST endpoint.
 * index: Added an ability to skip historical data while synchronizing `big_map` indexes.
 * metadata: Added `metadata` datasource.
 * tzkt: Added `get_big_map` and `get_contract_big_maps` datasource methods.

src/dipdup/cli.py

@@ -49,6 +49,7 @@
 from dipdup.models import Index
 from dipdup.models import Schema
 from dipdup.utils import iter_files
+from dipdup.utils.database import execute_sql_scripts
 from dipdup.utils.database import set_decimal_context
 from dipdup.utils.database import tortoise_wrapper
 from dipdup.utils.database import wipe_schema
@@ -304,9 +305,10 @@ async def hasura(ctx):
 
 
 @hasura.command(name='configure', help='Configure Hasura GraphQL Engine')
+@click.option('--force', is_flag=True, help='Proceed even if Hasura is already configured')
 @click.pass_context
 @cli_wrapper
-async def hasura_configure(ctx):
+async def hasura_configure(ctx, force: bool):
     config: DipDupConfig = ctx.obj.config
     url = config.database.connection_string
     models = f'{config.package}.models'
@@ -320,7 +322,7 @@ async def hasura_configure(ctx):
 
     async with tortoise_wrapper(url, models):
         async with hasura_gateway:
-            await hasura_gateway.configure()
+            await hasura_gateway.configure(force)
 
 
 @cli.group(help='Manage database schema')
@@ -403,8 +405,14 @@ async def schema_init(ctx):
     async with AsyncExitStack() as stack:
         await dipdup._set_up_database(stack)
         await dipdup._set_up_hooks()
+        await dipdup._create_datasources()
         await dipdup._initialize_schema()
 
+        # NOTE: It's not necessary a reindex, but it's safe to execute built-in scripts to (re)create views.
+        conn = get_connection(None)
+        sql_path = join(dirname(__file__), 'sql', 'on_reindex')
+        await execute_sql_scripts(conn, sql_path)
+
     _logger.info('Schema initialized')
 
 

src/dipdup/context.py

@@ -16,7 +16,6 @@
 from typing import Union
 from typing import cast
 
-import sqlparse  # type: ignore
 from tortoise import Tortoise
 from tortoise.exceptions import OperationalError
 from tortoise.transactions import get_connection
@@ -50,7 +49,7 @@
 from dipdup.models import ReindexingReason
 from dipdup.models import Schema
 from dipdup.utils import FormattedLogger
-from dipdup.utils import iter_files
+from dipdup.utils.database import execute_sql_scripts
 from dipdup.utils.database import wipe_schema
 
 pending_indexes = deque()  # type: ignore
@@ -333,14 +332,7 @@ async def execute_sql(self, ctx: 'DipDupContext', name: str) -> None:
 
         # NOTE: SQL hooks are executed on default connection
         connection = get_connection(None)
-
-        for file in iter_files(sql_path, '.sql'):
-            ctx.logger.info('Executing `%s`', file.name)
-            sql = file.read()
-            for statement in sqlparse.split(sql):
-                # NOTE: Ignore empty statements
-                with suppress(AttributeError):
-                    await connection.execute_script(statement)
+        await execute_sql_scripts(connection, sql_path)
 
     @contextmanager
     def _wrapper(self, kind: str, name: str) -> Iterator[None]:

src/dipdup/hasura.py

@@ -102,7 +102,7 @@ def __init__(
         self._hasura_config = hasura_config
         self._database_config = database_config
 
-    async def configure(self) -> None:
+    async def configure(self, force: bool = False) -> None:
         """Generate Hasura metadata and apply to instance with credentials from `hasura` config section."""
 
         if self._database_config.schema_name != 'public':
@@ -116,7 +116,7 @@ async def configure(self) -> None:
         metadata = await self._fetch_metadata()
         metadata_hash = self._hash_metadata(metadata)
 
-        if hasura_schema.hash == metadata_hash:
+        if not force and hasura_schema.hash == metadata_hash:
             self._logger.info('Metadata is up to date, no action required')
             return
 
@@ -342,6 +342,9 @@ async def _generate_query_collections_metadata(self) -> List[Dict[str, Any]]:
         for query_name, query in self._iterate_graphql_queries():
             queries.append({'name': query_name, 'query': query})
 
+        # NOTE: This is the only view we add by ourselves and thus know all params. Won't work for any view.
+        queries.append(self._format_rest_head_status_query())
+
         return queries
 
     async def _generate_rest_endpoints_metadata(self, query_names: List[str]) -> List[Dict[str, Any]]:
@@ -455,6 +458,16 @@ def _format_rest_query(self, name: str, table: str, filter: str, fields: Iterabl
             'query': 'query ' + name + ' (' + query_arg + ') {' + table + '(' + query_filter + ') {' + query_fields + '}}',
         }
 
+    def _format_rest_head_status_query(self) -> Dict[str, Any]:
+        name = 'dipdup_head_status'
+        if self._hasura_config.camel_case:
+            name = humps.camelize(name)
+
+        return {
+            'name': name,
+            'query': 'query ' + name + ' ($name: String!) {' + name + '(where: {name: {_eq: $name}}) {status}}',
+        }
+
     def _format_rest_endpoint(self, query_name: str) -> Dict[str, Any]:
         return {
             "definition": {

src/dipdup/sql/on_reindex/dipdup_head_status.sql

@@ -0,0 +1,10 @@
+CREATE
+OR REPLACE VIEW dipdup_head_status AS
+SELECT
+    name,
+    CASE
+        WHEN timestamp < NOW() - interval '2 minutes' THEN 'OUTDATED'
+        ELSE 'OK'
+    END AS status
+FROM
+    dipdup_head;

src/dipdup/utils/database.py

@@ -4,6 +4,7 @@
 import importlib
 import logging
 from contextlib import asynccontextmanager
+from contextlib import suppress
 from enum import Enum
 from os.path import dirname
 from os.path import join
@@ -16,6 +17,7 @@
 from typing import Type
 from typing import Union
 
+import sqlparse  # type: ignore
 from tortoise import Model
 from tortoise import Tortoise
 from tortoise.backends.asyncpg.client import AsyncpgDBClient
@@ -30,6 +32,7 @@
 
 from dipdup.enums import ReversedEnum
 from dipdup.exceptions import DatabaseConfigurationError
+from dipdup.utils import iter_files
 from dipdup.utils import pascal_to_snake
 
 _logger = logging.getLogger('dipdup.database')
@@ -144,13 +147,27 @@ async def create_schema(conn: BaseDBAsyncClient, name: str) -> None:
     await conn.execute_script(_truncate_schema_sql)
 
 
+async def execute_sql_scripts(conn: BaseDBAsyncClient, path: str) -> None:
+    for file in iter_files(path, '.sql'):
+        _logger.info('Executing `%s`', file.name)
+        sql = file.read()
+        for statement in sqlparse.split(sql):
+            # NOTE: Ignore empty statements
+            with suppress(AttributeError):
+                await conn.execute_script(statement)
+
+
 async def generate_schema(conn: BaseDBAsyncClient, name: str) -> None:
     if isinstance(conn, SqliteClient):
         await Tortoise.generate_schemas()
     elif isinstance(conn, AsyncpgDBClient):
         await create_schema(conn, name)
         await set_schema(conn, name)
         await Tortoise.generate_schemas()
+
+        # NOTE: Apply built-in scripts before project ones
+        sql_path = join(dirname(__file__), '..', 'sql', 'on_reindex')
+        await execute_sql_scripts(conn, sql_path)
     else:
         raise NotImplementedError