History kafka feat

README.md

@@ -36,6 +36,7 @@ It helps you focus on processsing the files needed and creating reports.
 - [`Report` declaration](#report-declaration)
   - [Creating the `Report component`](#creating-the-report-component)
   - [Creating the `Report`](#creating-the-report)
+- [History](#history)
 - [Custom namespaces](#custom-namespaces)
 - [Endpoints from simple functions](#endpoints-from-simple-functions)
 - [The `main` file](#the-main-file)
@@ -1260,6 +1261,97 @@ Reports api will be handled by the `ifmp_app` instance.
 
 
 
+<a name="history"></a>
+# History
+
+History is useful for provinding a live log of the processing step
+
+Recap in the worker function:
+
+```python
+from settings import config
+from licenseware import States, log
+
+from .rv_tools_data_worker import ProcessRVToolsEvent
+
+
+def rv_tools_worker(event: dict):
+
+    log.info("Starting working")
+    log.debug(event)
+
+    notify_upload_status(event, status=states.RUNNING)
+    try:
+        ProcessRVToolsEvent(event).run_processing_pipeline()
+    finally:
+        log.info("Finished working")
+        notify_upload_status(event, status=states.IDLE)
+
+```
+
+Let's see how to apply history on `ProcessRVToolsEvent`:
+
+```python
+from licenseware import history
+from licenseware.constants.worker_event_type import WorkerEvent
+from licenseware.pubsub.types import EventType
+from licenseware.repository import MongoRepository
+from app.dependencies import producer, mongodb_connection
+
+
+class ProcessRVToolsEvent:
+    def __init__(self, event: dict) -> None:
+        self.event = WorkerEvent(**event)
+        self.filepath = None
+        self.filename = None
+        self.producer = producer
+        self.event_type = EventType.PROCESSING_DETAILS
+        self.repo = MongoRepository(
+            mongodb_connection, 
+            collection="SomeCollection", 
+            data_validator="ignore"
+        )
+
+    @history.log # decorate the function you need to save in history
+    def get_data(self, fp: str):
+        # here we are pushing some entities to stream
+        history.publish_entities(
+            self.producer,
+            self.event,
+            entities=["some-entiry-here"],
+        )
+        return "ok"
+
+    @history.log
+    def save_data(self, data: dict):
+        return self.repo.insert_one(data)
+
+    # this one is the main function which triggers the processing pipe
+    def run_pipeline(self):
+
+        for fp in self.event.filepaths:
+            # Update the filepath/filename to self so th l
+            self.filepath = fp
+            self.filename = os.path.basename(fp)
+            # Data processing pipeline:
+            data = self.get_data(fp)
+            save_data(data)
+
+
+event = {
+    "tenant_id": str(uuid.uuid4()),
+    "authorization": str(uuid.uuid4()),
+    "uploader_id": "rv_tools",
+    "uploader_name": "RvTOOLS",
+    "event_id": str(uuid.uuid4()),
+    "app_id": "ifmp-service",
+    "app_name": "Infra App",
+    "filepaths": ["tests/test_data/comma.csv"],
+}
+
+```
+
+If you can't use the `history.log` decorator for some reason you can directly call `history.publish` and provide the required information need to publish to kafka.
 
 
 <a name="custom-namespaces"></a>

licenseware/app_builder/app_builder.py

@@ -29,6 +29,7 @@
     get_features_namespace,
 )
 from licenseware.app_builder.features_route import add_features_route
+from licenseware.app_builder.health_route import add_health_route
 from licenseware.app_builder.refresh_registration_route import (
     add_refresh_registration_route,
 )
@@ -96,6 +97,7 @@ class base_paths:
     features_path: str = "/features"
     data_sync_path: str = "/data-sync"
     reprocess_data_path: str = "/reprocess-data"
+    health_path: str = "/healthz"
 
 
 class AppBuilder:
@@ -151,6 +153,7 @@ def __init__(
         self.features_path = base_paths.features_path
         self.data_sync_path = base_paths.data_sync_path
         self.reprocess_data_path = base_paths.reprocess_data_path
+        self.health_path = base_paths.health_path
 
         self.app_activation_url = envs.BASE_URL + self.app_activation_path
         self.refresh_registration_url = envs.BASE_URL + self.refresh_registration_path
@@ -161,6 +164,7 @@ def __init__(
         self.features_url = envs.BASE_URL + self.features_path
         self.data_sync_url = envs.BASE_URL + self.data_sync_path
         self.reproces_data_url = envs.BASE_URL + self.reprocess_data_path
+        self.health_url = envs.BASE_URL + self.health_path
 
         self.authorizations = doc_authorizations
         self.decorators = [] if api_decorators is None else api_decorators
@@ -259,6 +263,7 @@ def init_routes(self):
         # Here we are adding the routes available for each app
         # Api must be passed from route function back to this context
         api_funcs = [
+            add_health_route,
             add_refresh_registration_route,
             add_editable_tables_route,
             add_tenant_registration_route,
@@ -344,6 +349,13 @@ def add_features_routes(self):
         for func in ns_funcs:
             self.register_namespace(func(ns=features_namespace, features=self.features))
 
+    def add_health_routes(self):
+
+        ns_funcs = [get_features_namespace]
+
+        for func in ns_funcs:
+            self.register_namespace(func(ns=features_namespace, features=self.features))
+
     def add_data_sync_routes(self):
 
         if self.data_sync_schema is None:

licenseware/app_builder/health_route.py

@@ -0,0 +1,22 @@
+"""
+
+When this endpoint is called the generated metadata from marshmallow schemas provided in `AppBuilder` at `editable_tables_schemas` parameter will be returned.
+
+See `editable_table` package for more information.
+
+"""
+
+from flask_restx import Api, Resource
+
+from licenseware.decorators import failsafe
+
+
+def add_health_route(api: Api, appvars: dict):
+    @api.route(appvars["health_path"])
+    class HealthRoute(Resource):
+        @failsafe(fail_code=500)
+        @api.doc(description="Yep still up")
+        def get(self):
+            return {"status": "success"}
+
+    return api

licenseware/app_builder/uploads_namespace/filestream_validation_namespace.py

@@ -6,6 +6,7 @@
 """
 
 
+import uuid
 from typing import List
 
 from flask import request
@@ -19,6 +20,7 @@
 from licenseware.tenants import clear_tenant_data
 from licenseware.uploader_builder import UploaderBuilder
 from licenseware.utils.common import trigger_broker_funcs
+from licenseware.utils.miscellaneous import get_flask_request_dict
 
 
 class FileUploadDetailedValidationSchema(Schema):
@@ -57,13 +59,21 @@ class FileStreamValidate(Resource):
         @authorization_check
         def post(self):
 
+            tenant_id = request.headers.get("Tenantid")
+
             clear_data = request.args.get("clear_data", "false")
             if "true" in clear_data.lower():
-                clear_tenant_data(
-                    request.headers.get("Tenantid"), uploader.collections_list
-                )
+                clear_tenant_data(tenant_id, uploader.collections_list)
+
+            event_id = request.args.get("event_id") or str(uuid.uuid4())
+
+            upload_response = uploader.upload_files(
+                request,
+                tenant_id,
+                event_id,
+                **get_flask_request_dict(request),
+            )
 
-            upload_response = uploader.upload_files(request)
             if uploader.broker_funcs:
                 trigger_broker_funcs(
                     request, uploader.broker_funcs, upload_response=upload_response[0]

licenseware/auth/auth.py

@@ -102,5 +102,6 @@ def _login(self):
         if response.status_code == 200:
             return response.json(), 200
 
+        log.warning(response.content)
         log.error(f"Could not login with {self.machine_name}")
         exit(1)

licenseware/common/serializers/event_schema.py

@@ -6,7 +6,10 @@
 class EventSchema(Schema):
     tenant_id = fields.Str(required=True, validate=validate_uuid4)
     event_id = fields.Str(required=True, validate=validate_uuid4)
+    app_id = fields.Str(required=True)
     uploader_id = fields.Str(required=True, validate=validate.Length(min=3))
     filepaths = fields.List(fields.Str, required=True)
     flask_request = fields.Dict(required=False, allow_none=True)
     validation_response = fields.Dict(required=False, allow_none=True)
+    app_name = fields.Str(required=True)
+    uploader_name = fields.Str(required=True)

licenseware/config/__init__.py

@@ -0,0 +1 @@
+from .config import *

licenseware/config/config.py

@@ -0,0 +1,129 @@
+from pydantic import BaseSettings
+
+from licenseware.constants.base_enum import BaseEnum
+
+
+class Collections(BaseEnum):
+    DATA = "Data"
+    QUOTA = "Quota"
+    HISTORY = "ProcessingHistory"
+    REPORT_SNAPSHOTS = "ReportSnapshots"
+    FEATURE = "Features"
+    TOKEN = "Tokens"
+    # Outdated
+    MONGO_COLLECTION_ANALYSIS_NAME = "History"
+
+
+class LogLevel(BaseEnum):
+    DEBUG = "DEBUG"
+    INFO = "INFO"
+    WARNING = "WARNING"
+    ERROR = "ERROR"
+    CRITICAL = "CRITICAL"
+
+
+class Environment(BaseEnum):
+    TEST = "TEST"
+    DEV = "DEV"
+    PROD = "PROD"
+    DESKTOP = "DESKTOP"
+
+
+class CeleryBrokerType(BaseEnum):
+    REDIS = "REDIS"
+    RABBITMQ = "RABBITMQ"
+
+
+class WebAppFramework(BaseEnum):
+    FASTAPI = "FASTAPI"
+    FLASK = "FLASK"
+
+
+class Config(BaseSettings):  # pragma no cover
+    APP_ID: str = None
+    APP_SECRET: str = None
+    FILE_UPLOAD_PATH: str = "/tmp/lware"
+    CURRENT_ENVIRONMENT: Environment = Environment.DEV
+    ENVIRONMENTS: Environment = Environment
+    LOG_LEVEL: LogLevel = LogLevel.INFO
+    PORT: int = 8000
+    DASHBOARD_WORKERS_HOST: str = None
+    DASHBOARD_WORKERS_PORT: int = None
+
+    BASE_URL: str = None
+
+    @property
+    def APP_URL(self):
+        return self.BASE_URL + "/" + self.APP_ID
+
+    PUBLIC_TOKEN_REPORT_URL: str = None
+    FRONTEND_URL: str = None
+
+    MACHINE_NAME: str = None
+    MACHINE_PASSWORD: str = None
+
+    AUTH_SERVICE_URL: str = None
+    AUTH_MACHINE_LOGIN_URL: str = None
+    AUTH_USER_LOGIN_URL: str = None
+    AUTH_USER_INFO_URL: str = None
+    AUTH_MACHINE_CHECK_URL: str = None
+    AUTH_USER_CHECK_URL: str = None
+
+    REGISTRY_SERVICE_URL: str = None
+    REGISTRY_SERVICE_APPS_URL: str = None
+    REGISTRY_SERVICE_UPLOADERS_URL: str = None
+    REGISTRY_SERVICE_REPORTS_URL: str = None
+    REGISTRY_SERVICE_COMPONENTS_URL: str = None
+
+    # mongodb://lware:lware-secret@localhost:27017
+    MONGO_HOST: str = "mongodb"
+    MONGO_DBNAME: str = "db"
+    MONGO_PORT: int = 27017
+    MONGO_USER: str = "lware"
+    MONGO_PASSWORD: str = "lwaredev"
+    MONGO_COLLECTION: Collections = Collections
+
+    REDIS_HOST: str = "redis"
+    REDIS_PORT: int = 6379
+    REDIS_DB: int = 0
+    REDIS_RESULT_CACHE_DB: int = 1
+    REDIS_PASSWORD: str = None
+    EXPIRE_REGISTRATION: int = 900  # 15 mins
+    EXPIRE_UPLOADER_STATUS: int = 7200  # 2 hours
+    EXPIRE_USER_CHECK: int = 60  # 1 minute
+    EXPIRE_MACHINE_CHECK: int = 60  # 1 minute
+    EXPIRE_NOTIFICATION: int = 259_200  # 3 days
+
+    KAFKA_BROKER_URL: str = "PLAINTEXT://kafka:9092"
+    KAFKA_CONSUMER_POLL: float = 1.0
+    KAFKA_SECURITY_PROTOCOL: str = "PLAINTEXT"
+
+    JAEGER_MODE: str = "grpc"
+    JAEGER_COLLECTOR_ENDPOINT_GRPC_ENDPOINT: str = "jaeger-collector:14250"
+    JAEGER_COLLECTOR_THRIFT_URL: str = "http://jaeger-collector:14268"
+    JAEGER_AGENT_HOST_NAME: str = "jaeger-agent"
+    JAEGER_AGENT_PORT: int = 6831
+    OPEN_TELEMETRY_HOST: str = "127.0.0.1"
+    OPEN_TELEMETRY_PORT: int = 6831
+
+    CELERY_BROKER_REDIS_DB: int = 1
+    CELERY_BACKEND_REDIS_DB: int = 2
+    CELERY_BEATS_REGISTRATION_INTERVAL: int = 600  # 10 minutes
+    REFRESH_MACHINE_TOKEN_INTERVAL: int = 86_400  # 24 hours
+
+    CELERY_BROKER_TYPE: CeleryBrokerType = CeleryBrokerType.REDIS
+    WEBAPP_FRAMEWORK: WebAppFramework = WebAppFramework.FASTAPI
+
+    @property
+    def celery_broker_uri(self):
+        return (
+            f"redis://{self.REDIS_HOST}:{self.REDIS_PORT}/{self.CELERY_BROKER_REDIS_DB}"
+        )
+
+    @property
+    def celery_result_backend_uri(self):
+        return f"redis://{self.REDIS_HOST}:{self.REDIS_PORT}/{self.CELERY_BACKEND_REDIS_DB}"
+
+    class Config:
+        env_file = ".env"
+        case_sensitive = True

licenseware/constants/__init__.py

@@ -0,0 +1,19 @@
+from .allowed_filters import AllowedFilters
+from .attributes_type import AttributesType
+from .base_enum import BaseEnum
+from .column_types import ColumnTypes
+from .file_types import DownloadFileType, FileTypes
+from .filter_item_type import FilterItemType, FilterUI
+from .flags import Flags
+from .http import HTTP_METHODS, HTTP_STATUS_CODES
+from .icons import Icons
+from .report_component_type import ReportComponentType
+from .report_component_types import RCTypes
+from .report_filter_type import ReportFilterType
+from .report_type import ReportType
+from .states import States
+from .style_attributes_type import StyleAttrsType
+from .tenant_types import *
+from .uploader_types import *
+from .web_response import WebResponse
+from .worker_event_type import WorkerEvent