LITE-29340: Basic project's folder structure and dependencies

connect_bi_reporter/connect_services/devops.py

@@ -0,0 +1,6 @@
+def create_schedule_task(client, context, payload):
+    ...
+
+
+def get_schedule_tasks(client, context):
+    ...

connect_bi_reporter/connect_services/reports.py

@@ -0,0 +1,6 @@
+def get_reporting_schedules(client, filters):
+    pass
+
+
+def get_reporting_reports(client, filters):
+    pass

connect_bi_reporter/db.py

@@ -0,0 +1,182 @@
+from contextlib import contextmanager
+from functools import reduce
+import os
+import random
+
+from connect.eaas.core.inject.common import get_config
+from fastapi import Depends
+from sqlalchemy import create_engine
+from sqlalchemy.engine.base import Engine
+from sqlalchemy.ext.declarative import declarative_base
+from sqlalchemy.orm import Session, sessionmaker
+from sqlalchemy.exc import DontWrapMixin
+
+
+_MAX_RETRIES = 1000
+_ENGINE = None
+
+
+def _get_numeric_string(size):
+    return str(random.randint(1 * 10 ** (size - 1), 1 * 10 ** size - 1))
+
+
+def _generate_verbose_id(prefix):
+    return (
+        f'{prefix}-{_get_numeric_string(3)}'
+        f'-{_get_numeric_string(3)}-{_get_numeric_string(3)}'
+    )
+
+
+class VerboseSessionError(Exception, DontWrapMixin):
+    '''
+    Exception class to handle errors through custom methods of `VerboseBaseSession`.
+    '''
+
+
+class VerboseBaseSession(Session):
+    '''
+    Custom `sqlalchemy.orm.Session` class to generate and add verbose_id to a model.
+    '''
+    def _set_verbose(self, instance):
+        instance_class = instance.__class__
+        for _ in range(1, _MAX_RETRIES + 1):
+            verbose_id = _generate_verbose_id(instance_class.PREFIX)
+            if not (
+                self.query(
+                    self.query(instance_class).filter(instance_class.id == verbose_id).exists(),
+                ).scalar()
+            ):
+                instance.id = verbose_id
+                return instance
+
+        raise VerboseSessionError(
+            f'Could not generate {instance_class.__name__} verbose ID'
+            f' after {_MAX_RETRIES} attempts.',
+        )
+
+    def _set_verbose_all(self, instances):
+        checked_instances = []
+        if instances:
+            count = len(instances)
+            all_same_class_check = reduce(lambda x, y: x == y, [ins.__class__ for ins in instances])
+            assert all_same_class_check, 'All instances must be of the same class.'
+
+            instance_class = instances[0].__class__
+            for _ in range(1, _MAX_RETRIES + 1):
+                ids = [_generate_verbose_id(instance_class.PREFIX) for _ in range(count)]
+                if not (
+                    self.query(
+                        self.query(instance_class).filter(instance_class.id.in_(ids)).exists(),
+                    ).scalar()
+                ):
+                    break
+                ids = []
+            if not ids:
+                raise VerboseSessionError(
+                    f'Could not generate a group of {count} {instance_class.__name__} verbose ID'
+                    f' after {_MAX_RETRIES} attempts.',
+                )
+
+            for instance, verbose_id in zip(instances, ids):
+                instance.id = verbose_id
+                checked_instances.append(instance)
+        return checked_instances
+
+    def add_with_verbose(self, instance):
+        instance = self._set_verbose(instance)
+        return self.add(instance)
+
+    def add_all_with_verbose(self, instances):
+        instances = self._set_verbose_all(instances)
+        return self.add_all(instances)
+
+    def add_next_with_verbose(self, instance, related_id_field):
+        instance_class = instance.__class__
+        new_suffix = 0
+        related_id_value = getattr(instance, related_id_field)
+        if (
+            self.query(self.query(instance_class).filter(
+                instance_class.__dict__[related_id_field] == related_id_value).exists(),
+            ).scalar()
+        ):
+            last_obj = self.query(instance_class).order_by(
+                instance_class.id.desc(),
+            ).first()
+            _instance_id, suffix = last_obj.id.rsplit('-', 1)
+            new_suffix = int(suffix) + 1
+        else:
+            id_body = related_id_value.split('-', 1)[-1]
+            _instance_id = f"{instance_class.PREFIX}-{id_body}"
+
+        instance.id = '{0}-{1}'.format(_instance_id, '{0:03d}'.format(new_suffix))
+        return self.add(instance)
+
+    def add_all_with_next_verbose(self, instances, related_id_field):
+        first_item = instances[0]
+        instance_class = first_item.__class__
+        new_suffix = 0
+        related_id_value = getattr(first_item, related_id_field)
+
+        if (
+            self.query(self.query(instance_class).filter(
+                instance_class.__dict__[related_id_field] == related_id_value).exists(),
+            ).scalar()
+        ):
+
+            last_obj = self.query(instance_class).order_by(
+                instance_class.id.desc(),
+            ).first()
+            _instance_id, suffix = last_obj.id.rsplit('-', 1)
+            new_suffix = int(suffix) + 1
+        else:
+            id_body = related_id_value.split('-', 1)[-1]
+            _instance_id = f"{instance_class.PREFIX}-{id_body}"
+
+        for instance in instances:
+            instance.id = '{0}-{1}'.format(_instance_id, '{0:03d}'.format(new_suffix))
+            new_suffix += 1
+
+        return self.add_all(instances)
+
+
+SessionLocal = sessionmaker(autocommit=False, autoflush=False, class_=VerboseBaseSession)
+Model = declarative_base()
+
+
+def get_engine(config: dict = Depends(get_config)):
+    global _ENGINE
+    if not _ENGINE:
+        _ENGINE = create_engine(
+            config.get(
+                'DATABASE_URL',
+                os.getenv('DATABASE_URL', 'postgresql+psycopg2://postgres:1q2w3e@db/bi_reporter'),
+            ),
+            pool_pre_ping=True,
+            pool_recycle=300,
+        )
+
+    return _ENGINE
+
+
+def get_db(engine: Engine = Depends(get_engine)):
+    db: VerboseBaseSession = SessionLocal(bind=engine)
+    try:
+        yield db
+    finally:
+        db.close()
+
+
+def create_db(config: dict = Depends(get_config)):
+    engine = get_engine(config)
+    Model.metadata.create_all(bind=engine)
+    return engine
+
+
+@contextmanager
+def get_db_ctx_manager(config):
+    engine: Engine = get_engine(config)
+    db: VerboseBaseSession = SessionLocal(bind=engine)
+    try:
+        yield db
+    finally:
+        db.close()

connect_bi_reporter/errors.py

@@ -0,0 +1,71 @@
+from typing import Any, Dict, Union
+
+from connect.client import ClientError
+
+
+class Error:
+
+    STATUS_CODE = 400
+
+    def __init__(self, message, error_code):
+        self.message = message
+        self.error_code = error_code
+
+    def __call__(self, **kwds: Dict[str, Any]) -> ClientError:
+        format_kwargs = kwds.get('format_kwargs', {})
+
+        message = self.message.format(**format_kwargs)
+        errors = kwds.get('errors')
+
+        if not errors:
+            errors = [message or 'Unexpected error.']
+        if not isinstance(errors, list):
+            errors = [errors]
+
+        return ClientError(
+            message=message,
+            status_code=self.STATUS_CODE,
+            error_code=self.error_code,
+            errors=errors,
+        )
+
+
+class ExtensionErrorMeta(type):
+    PREFIX = 'EXT'
+    ERRORS = {}
+
+    def __getattr__(cls, __name: str) -> Union[Error, AttributeError]:
+        valid_dict = {cls.PREFIX: cls.ERRORS}
+        try:
+            prefix, code = __name.split('_')
+            error = valid_dict[prefix][int(code)]
+        except (KeyError, ValueError):
+            raise AttributeError(f"type object '{cls.__name__}' has no attribute '{__name}'")
+        return Error(message=error, error_code=__name)
+
+
+class ExtensionErrorBase(metaclass=ExtensionErrorMeta):
+    '''
+    Base Error class to group a set of errors base on a prefix.
+    By default the `PREFIX` value is `EXT`, but it can be overwritten.
+    Also `status_code` and/or list of `errors` can be provided.
+
+    Usage:
+
+    ```
+    # Define a custom error class
+    class MyError(ExtensionErrorBase)
+        PREFIX = "BIR"
+        ERRORS = {
+            1: "Some error",
+            2: "Some {template} error.",
+            3: "Not found",
+
+        }
+
+    # raise the error
+    raise MyError.BIR_001()
+    raise MyError.BIR_002(format_kwargs={"template": "foo"})
+    raise MyError.BIR_003(status_code=404)
+    ```
+    '''

connect_bi_reporter/events.py

@@ -3,16 +3,11 @@
 # Copyright (c) 2024, Ingram Micro
 # All rights reserved.
 #
-from connect.eaas.core.decorators import (
-    event,
-    schedulable,
-    variables,
-)
+from connect.eaas.core.decorators import event, variables
 from connect.eaas.core.extension import EventsApplicationBase
-from connect.eaas.core.responses import (
-    BackgroundResponse,
-    ScheduledExecutionResponse,
-)
+from connect.eaas.core.responses import BackgroundResponse
+
+from connect_bi_reporter.uploads.tasks import UploadTaskApplicationMixin
 
 
 @variables([
@@ -27,7 +22,10 @@
         'secure': True,
     },
 ])
-class ConnectBiReporterEventsApplication(EventsApplicationBase):
+class ConnectBiReporterEventsApplication(
+    EventsApplicationBase,
+    UploadTaskApplicationMixin,
+):
     @event(
         'installation_status_change',
         statuses=[
@@ -37,10 +35,3 @@ class ConnectBiReporterEventsApplication(EventsApplicationBase):
     def handle_installation_status_change(self, request):
         self.logger.info(f"Obtained request with id {request['id']}")
         return BackgroundResponse.done()
-
-    @schedulable('Schedulable method', 'It can be used to test DevOps scheduler.')
-    def execute_scheduled_processing(self, schedule):
-        self.logger.info(
-            f"Received event for schedule  {schedule['id']}",
-        )
-        return ScheduledExecutionResponse.done()

connect_bi_reporter/uploads/tasks.py

@@ -0,0 +1,20 @@
+from connect.eaas.core.decorators import schedulable
+from connect.eaas.core.responses import (
+    ScheduledExecutionResponse,
+)
+
+
+class UploadTaskApplicationMixin:
+    @schedulable(
+        'Create Uploads',
+        'DESCRIPTION',
+    )
+    def create_uploads(self, schedule):
+        return ScheduledExecutionResponse.done()
+
+    @schedulable(
+        'Process Uploads',
+        'DESCRIPTION',
+    )
+    def process_uploads(self, schedule):
+        return ScheduledExecutionResponse.done()

connect_bi_reporter/webapp.py

@@ -3,13 +3,13 @@
 # Copyright (c) 2024, Ingram Micro
 # All rights reserved.
 #
-from connect.eaas.core.decorators import (
-    router,
-    web_app,
-)
+from connect.eaas.core.decorators import router, web_app
 from connect.eaas.core.extension import WebApplicationBase
 
 
 @web_app(router)
 class ConnectBiReporterWebApplication(WebApplicationBase):
-    ...
+
+    @classmethod
+    def on_startup(cls, logger, config):
+        ...

docker-compose.yml

@@ -14,6 +14,8 @@ services:
       - node_modules:/extension/node_modules
     env_file:
       - .connect_bi_reporter_dev.env
+    depends_on:
+      - db
 
   connect_bi_reporter_bash:
     container_name: connect_bi_reporter_bash
@@ -40,3 +42,28 @@ services:
       - node_modules:/extension/node_modules
     env_file:
       - .connect_bi_reporter_dev.env
+    depends_on:
+      - db
+    links:
+      - "db_ram:db"
+
+  db:
+    image: postgres:9.6-alpine
+    restart: always
+    ports:
+      - '5433:5432'
+    expose:
+      - '5432'
+    volumes:
+      - ./pg_data:/var/lib/postgresql/data
+    environment:
+      POSTGRES_PASSWORD: 1q2w3e
+      POSTGRES_DB: bi_reporter
+
+  db_ram:
+    image: postgres:9.6-alpine
+    tmpfs:
+      - /var/lib/postgresql/data
+    environment:
+      POSTGRES_PASSWORD: 1q2w3e
+      POSTGRES_DB: bi_reporter