Validate configurations before creation

.github/workflows/docker-hub-image-build.yml

@@ -4,6 +4,12 @@ on:
   release:
       types:
           - "released"
+  workflow_dispatch:
+      inputs:
+          versionTag:
+              description: "Version Tag"
+              required: true
+              default: ''
 
 jobs:
   main:
@@ -20,6 +26,7 @@ jobs:
 
       - name: Get the version
         id: get_version
+        if: github.event_name == 'push'
         run: echo ::set-output name=VERSION::${GITHUB_REF/refs\/tags\//}
 
       - name: Login to DockerHub
@@ -36,13 +43,13 @@ jobs:
           file: ./Dockerfile.prod
           platforms: linux/amd64
           build-args: |
-            release_version=${{ steps.get_version.outputs.VERSION }}
+            release_version=${{ github.event.inputs.versionTag || steps.get_version.outputs.VERSION }}
           push: true
           cache-from: type=local,src=/tmp/.buildx-cache
           cache-to: type=local,dest=/tmp/.buildx-cache
           tags: |
             onaio/duva:latest
-            onaio/duva:${{ steps.get_version.outputs.VERSION }}
+            onaio/duva:${{ github.event.inputs.versionTag || steps.get_version.outputs.VERSION }}
 
       - name: Image digest
         run: echo ${{ steps.docker_build.outputs.digest }}

app/common_tags.py

@@ -10,3 +10,4 @@
 
 SYNC_FAILURES_METADATA = "sync-failures"
 JOB_ID_METADATA = "job-id"
+FAILURE_REASON_METADATA = "failure-reason"

app/libs/tableau/client.py

@@ -5,6 +5,10 @@
 from app.models import Configuration
 
 
+class InvalidConfiguration(Exception):
+    pass
+
+
 class TableauClient:
     def __init__(self, configuration: Configuration):
         self.project_name = configuration.project_name
@@ -13,6 +17,25 @@ def __init__(self, configuration: Configuration):
         self.site_name = configuration.site_name
         self.server_address = configuration.server_address
 
+    @staticmethod
+    def validate_configuration(configuration):
+        if isinstance(configuration, Configuration):
+            access_token = Configuration.decrypt_value(configuration.token_value)
+        else:
+            access_token = configuration.token_value
+
+        tableau_auth = TSC.PersonalAccessTokenAuth(
+            token_name=configuration.token_name,
+            personal_access_token=access_token,
+            site_id=configuration.site_name,
+        )
+        try:
+            server = TSC.Server(configuration.server_address, use_server_version=True)
+            server.auth.sign_in(tableau_auth)
+            server.auth.sign_out()
+        except Exception as e:
+            raise InvalidConfiguration(f"Failed to validate configuration: {e}")
+
     def publish_hyper(self, hyper_name):
         """
         Signs in and publishes an extract directly to Tableau Online/Server

app/routers/configuration.py

@@ -13,6 +13,7 @@
 from app.models import Configuration, User
 from app.utils.auth_utils import IsAuthenticatedUser
 from app.utils.utils import get_db
+from app.libs.tableau.client import TableauClient, InvalidConfiguration
 
 
 router = APIRouter()
@@ -79,10 +80,13 @@ def create_configuration(
     """
     config_data = schemas.ConfigurationCreate(user=user.id, **config_data.dict())
     try:
+        TableauClient.validate_configuration(config_data)
         config = Configuration.create(db, config_data)
         return config
     except (UniqueViolation, IntegrityError):
         raise HTTPException(status_code=400, detail="Configuration already exists")
+    except InvalidConfiguration as e:
+        raise HTTPException(status_code=400, detail=str(e))
 
 
 @router.patch(
@@ -108,11 +112,14 @@ def patch_configuration(
                     if key == "token_value":
                         value = Configuration.encrypt_value(value)
                     setattr(config, key, value)
+            TableauClient.validate_configuration(config)
             db.commit()
             db.refresh(config)
             return config
         except (UniqueViolation, IntegrityError):
             raise HTTPException(status_code=400, detail="Configuration already exists")
+        except InvalidConfiguration as e:
+            raise HTTPException(status_code=400, detail=str(e))
     else:
         raise HTTPException(404, detail="Tableau Configuration not found.")
 

app/routers/file.py

@@ -17,6 +17,7 @@
 from app import schemas
 from app.common_tags import HYPER_PROCESS_CACHE_KEY
 from app.libs.s3.client import S3Client
+from app.libs.tableau.client import TableauClient, InvalidConfiguration
 from app.models import HyperFile, Configuration, User
 from app.settings import settings
 from app.utils.auth_utils import IsAuthenticatedUser
@@ -157,6 +158,9 @@ def list_hyper_files(
                 form_id=hyperfile.form_id,
                 filename=hyperfile.filename,
                 file_status=hyperfile.file_status,
+                last_updated=hyperfile.last_updated,
+                last_synced=hyperfile.last_synced,
+                meta_data=hyperfile.meta_data,
             )
         )
     return response
@@ -291,6 +295,16 @@ def trigger_hyper_file_sync(
 
     if not hyper_file:
         raise HTTPException(404, "File not found.")
+
+    if hyper_file.configuration:
+        try:
+            TableauClient.validate_configuration(hyper_file.configuration)
+        except InvalidConfiguration as e:
+            raise HTTPException(
+                400,
+                detail=f"Invalid configuration ID {hyper_file.configuration.id}: {e}",
+            )
+
     if hyper_file.user == user.id:
         status_code = 200
         if hyper_file.file_status not in [

app/schemas.py

@@ -56,6 +56,9 @@ class FileListItem(BaseModel):
     form_id: int
     filename: str
     file_status: FileStatusEnum = FileStatusEnum.file_unavailable.value
+    last_updated: Optional[datetime] = None
+    last_synced: Optional[datetime] = None
+    meta_data: Optional[dict] = None
 
 
 class FileCreate(FileBase):
@@ -151,6 +154,7 @@ class FileResponseBody(FileBase):
     download_url: Optional[str]
     download_url_valid_till: Optional[str]
     configuration_url: Optional[str]
+    meta_data: Optional[dict] = None
 
     class Config:
         orm_mode = True

app/tests/routes/test_configuration.py

@@ -1,10 +1,15 @@
+from unittest.mock import patch
+
 from app.models import Configuration
 from app import schemas
 from app.tests.test_base import TestBase
 
 
 class TestConfiguration(TestBase):
-    def _create_configuration(self, auth_credentials: dict, config_data: dict = None):
+    @patch("app.routers.configuration.TableauClient")
+    def _create_configuration(
+        self, auth_credentials: dict, mock_client, config_data: dict = None
+    ):
         config_data = (
             config_data
             or schemas.ConfigurationCreateRequest(
@@ -15,6 +20,7 @@ def _create_configuration(self, auth_credentials: dict, config_data: dict = None
                 project_name="default",
             ).dict()
         )
+        mock_client.validate_configuration.return_value = True
         response = self.client.post(
             "/api/v1/configurations", json=config_data, headers=auth_credentials
         )
@@ -72,7 +78,8 @@ def test_delete_config(self, create_user_and_login):
         assert response.status_code == 204
         assert len(Configuration.get_all(self.db)) == current_count - 1
 
-    def test_patch_config(self, create_user_and_login):
+    @patch("app.routers.configuration.TableauClient")
+    def test_patch_config(self, mock_client, create_user_and_login):
         _, jwt = create_user_and_login
         jwt = jwt.decode("utf-8")
         auth_credentials = {"Authorization": f"Bearer {jwt}"}
@@ -91,6 +98,7 @@ def test_patch_config(self, create_user_and_login):
             id=config_id,
             export_settings=schemas.ExportConfigurationSettings(),
         ).dict()
+        mock_client.validate_configuration.return_value = True
 
         # Able to patch Tableau Configuration
         response = self.client.patch(

app/tests/routes/test_file.py

@@ -224,6 +224,7 @@ def test_file_with_config(self, mock_presigned_create, create_user_and_login):
             "last_synced": None,
             "last_updated": None,
             "configuration_url": f"http://testserver/api/v1/configurations/{config.id}",
+            "meta_data": {"job-id": "", "sync-failures": 0},
         }
 
         assert response.status_code == 201
@@ -273,6 +274,9 @@ def test_file_list(self, create_user_and_login):
             id=hyperfile.id,
             form_id=hyperfile.form_id,
             filename=hyperfile.filename,
+            last_updated=hyperfile.last_updated,
+            last_synced=hyperfile.last_synced,
+            meta_data=hyperfile.meta_data,
         ).dict()
         expected_data.update({"file_status": schemas.FileStatusEnum.queued.value})
         assert response.json()[0] == expected_data
@@ -343,6 +347,7 @@ def test_file_get(self, mock_presigned_create, create_user_and_login):
             "download_url",
             "download_url_valid_till",
             "configuration_url",
+            "meta_data",
         ]
         assert response.status_code == 200
         assert list(response.json().keys()) == expected_keys

app/utils/hyper_utils.py

@@ -24,7 +24,11 @@
 from app.database import SessionLocal
 from app.schemas import FileStatusEnum
 from app.settings import settings
-from app.common_tags import JOB_ID_METADATA, SYNC_FAILURES_METADATA
+from app.common_tags import (
+    JOB_ID_METADATA,
+    SYNC_FAILURES_METADATA,
+    FAILURE_REASON_METADATA,
+)
 from app.jobs.scheduler import schedule_cron_job, cancel_job
 from app.libs.s3.client import S3Client
 from app.libs.tableau.client import TableauClient
@@ -153,6 +157,7 @@ def handle_csv_import(
 
         if configuration:
             tableau_client = TableauClient(configuration=configuration)
+            tableau_client.validate_configuration(configuration)
             tableau_client.publish_hyper(file_path)
 
         return import_count
@@ -216,6 +221,7 @@ def handle_hyper_file_job_completion(
     file_status: str = FileStatusEnum.file_available.value,
     job_id_meta_tag: str = JOB_ID_METADATA,
     job_failure_counter_meta_tag: str = SYNC_FAILURES_METADATA,
+    failure_reason: str = None,
 ):
     """
     Handles updating a HyperFile according to the outcome of a running Job; Updates
@@ -228,13 +234,17 @@ def handle_hyper_file_job_completion(
         if object_updated:
             hf.last_updated = datetime.now()
         metadata[job_failure_counter_meta_tag] = 0
+        metadata.pop(FAILURE_REASON_METADATA, None)
     else:
         failure_count = metadata.get(job_failure_counter_meta_tag)
         if isinstance(failure_count, int):
             metadata[job_failure_counter_meta_tag] = failure_count + 1
         else:
             metadata[job_failure_counter_meta_tag] = failure_count = 0
 
+        if failure_reason:
+            metadata[FAILURE_REASON_METADATA] = failure_reason
+
         if failure_count >= settings.job_failure_limit and hf.is_active:
             cancel_hyper_file_job(
                 hyperfile_id,

app/utils/onadata_utils.py

@@ -5,7 +5,6 @@
 from typing import Optional
 
 import httpx
-import sentry_sdk
 from redis import Redis
 from redis.exceptions import LockError
 from fastapi_cache import caches
@@ -191,15 +190,14 @@ def start_csv_import_to_hyper(
         host=settings.redis_host, port=settings.redis_port, db=settings.redis_db
     )
     hyperfile: HyperFile = HyperFile.get(db, object_id=hyperfile_id)
-    job_status: str = schemas.FileStatusEnum.file_available.value
-    err: Exception = None
+    user = User.get(db, hyperfile.user)
+    job_status = schemas.FileStatusEnum.latest_sync_failed.value
+    err = None
 
-    if hyperfile:
+    if hyperfile and user and user.server:
         try:
             with redis_client.lock(f"{HYPERFILE_SYNC_LOCK_PREFIX}{hyperfile.id}"):
-                user = User.get(db, hyperfile.user)
                 server = user.server
-
                 hyperfile.file_status = schemas.FileStatusEnum.syncing.value
                 db.commit()
                 db.refresh(hyperfile)
@@ -214,6 +212,7 @@ def start_csv_import_to_hyper(
 
                     if export:
                         handle_csv_import_to_hyperfile(hyperfile, export, process, db)
+                        job_status = schemas.FileStatusEnum.file_available.value
 
                         if schedule_cron and not hyperfile.meta_data.get(
                             JOB_ID_METADATA
@@ -224,25 +223,27 @@ def start_csv_import_to_hyper(
                     else:
                         job_status = schemas.FileStatusEnum.file_unavailable.value
                 except (CSVExportFailure, ConnectionRequestError, Exception) as exc:
-                    err = exc
+                    err = str(exc)
                     job_status = schemas.FileStatusEnum.latest_sync_failed.value
 
-                successful_import = (
-                    job_status == schemas.FileStatusEnum.file_available.value
-                )
-                handle_hyper_file_job_completion(
-                    hyperfile.id,
-                    db,
-                    job_succeeded=successful_import,
-                    object_updated=successful_import,
-                    file_status=job_status,
-                )
-                db.close()
-                if err:
-                    FAILED_IMPORTS.inc()
-                    sentry_sdk.capture_exception(err)
+            successful_import = (
+                job_status == schemas.FileStatusEnum.file_available.value
+            )
+            if successful_import:
                 SUCCESSFUL_IMPORTS.inc()
-                return successful_import
+            else:
+                FAILED_IMPORTS.inc()
+
+            handle_hyper_file_job_completion(
+                hyperfile.id,
+                db,
+                job_succeeded=successful_import,
+                object_updated=successful_import,
+                file_status=job_status,
+                failure_reason=err,
+            )
+            db.close()
+            return successful_import
         except LockError:
             pass
 

docker-compose.yml

@@ -3,8 +3,6 @@ version: "3"
 services:
   cache:
     image: redis:6-alpine
-    ports:
-      - 6379:6379
   db:
     image: postgres:13
     volumes:

requirements.in

@@ -8,6 +8,7 @@ python-dateutil
 pandas
 boto3
 tableauserverclient
+tableauhyperapi==0.0.14265
 sqlalchemy
 cryptography
 rq

requirements.pip

@@ -181,7 +181,7 @@ starlette==0.13.6
     #   starlette-exporter
 starlette-exporter==0.9.0
     # via -r requirements.in
-tableauhyperapi==0.0.11556
+tableauhyperapi==0.0.14265
     # via -r requirements.in
 tableauserverclient==0.13
     # via -r requirements.in