[pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

Author: pre-commit-ci[bot]

CogniwareIms/.pre-commit-config.yaml

@@ -1,3 +1,6 @@
+# Copyright (C) 2026 Intel Corporation
+# SPDX-License-Identifier: Apache-2.0
+
 # Pre-commit hooks configuration for OPEA compliance
 # See https://pre-commit.com for more information
 

CogniwareIms/README.md

@@ -91,4 +91,3 @@ Apache 2.0 - See [LICENSE](LICENSE) file for details.
 ## Support
 
 For issues and questions, please open an issue in the OPEA GenAIExamples repository.
-

CogniwareIms/backend/Dockerfile

@@ -1,3 +1,6 @@
+# Copyright (C) 2026 Intel Corporation
+# SPDX-License-Identifier: Apache-2.0
+
 # Multi-stage build for production optimization
 FROM python:3.11-slim as builder
 

CogniwareIms/backend/app/core/config.py

@@ -1,6 +1,5 @@
 # Copyright (C) 2024 Intel Corporation
 # SPDX-License-Identifier: Apache-2.0
-
 """Application configuration Centralized settings management following 12-factor app principles."""
 
 import os
@@ -22,14 +21,10 @@ class Settings(BaseSettings):
 
     # API Configuration
     API_V1_PREFIX: str = "/api"
-    ALLOWED_ORIGINS: List[str] = os.getenv(
-        "ALLOWED_ORIGINS", "http://localhost:3000,http://frontend:3000"
-    ).split(",")
+    ALLOWED_ORIGINS: List[str] = os.getenv("ALLOWED_ORIGINS", "http://localhost:3000,http://frontend:3000").split(",")
 
     # Security
-    JWT_SECRET_KEY: str = os.getenv(
-        "JWT_SECRET_KEY", "CHANGE_THIS_IN_PRODUCTION_USE_openssl_rand_hex_32"
-    )
+    JWT_SECRET_KEY: str = os.getenv("JWT_SECRET_KEY", "CHANGE_THIS_IN_PRODUCTION_USE_openssl_rand_hex_32")
     JWT_ALGORITHM: str = "HS256"
     ACCESS_TOKEN_EXPIRE_MINUTES: int = 30
     REFRESH_TOKEN_EXPIRE_DAYS: int = 7
@@ -39,9 +34,7 @@ class Settings(BaseSettings):
     RATE_LIMIT_PER_MINUTE: int = int(os.getenv("RATE_LIMIT_PER_MINUTE", "60"))
 
     # Database
-    DATABASE_URL: str = os.getenv(
-        "DATABASE_URL", "postgresql://postgres:postgres@postgres:5432/opea_ims"
-    )
+    DATABASE_URL: str = os.getenv("DATABASE_URL", "postgresql://postgres:postgres@postgres:5432/opea_ims")
     DB_POOL_SIZE: int = 10
     DB_MAX_OVERFLOW: int = 20
 
@@ -50,13 +43,9 @@ class Settings(BaseSettings):
     REDIS_MAX_CONNECTIONS: int = 50
 
     # OPEA Services
-    OPEA_EMBEDDING_URL: str = os.getenv(
-        "OPEA_EMBEDDING_URL", "http://embedding-service:6000"
-    )
+    OPEA_EMBEDDING_URL: str = os.getenv("OPEA_EMBEDDING_URL", "http://embedding-service:6000")
     OPEA_LLM_URL: str = os.getenv("OPEA_LLM_URL", "http://llm-service:9000")
-    OPEA_RETRIEVAL_URL: str = os.getenv(
-        "OPEA_RETRIEVAL_URL", "http://retrieval-service:7000"
-    )
+    OPEA_RETRIEVAL_URL: str = os.getenv("OPEA_RETRIEVAL_URL", "http://retrieval-service:7000")
     OPEA_GATEWAY_URL: str = os.getenv("OPEA_GATEWAY_URL", "http://opea-gateway:8888")
 
     # Models

CogniwareIms/backend/app/core/security.py

@@ -1,6 +1,5 @@
 # Copyright (C) 2024 Intel Corporation
 # SPDX-License-Identifier: Apache-2.0
-
 """
 Security utilities - JWT, password hashing, authentication
 Industry-standard security implementation
@@ -53,9 +52,7 @@ def get_password_hash(password: str) -> str:
         return pwd_context.hash(password)
 
     @staticmethod
-    def create_access_token(
-        data: Dict[str, Any], expires_delta: Optional[timedelta] = None
-    ) -> str:
+    def create_access_token(data: Dict[str, Any], expires_delta: Optional[timedelta] = None) -> str:
         """Create a JWT access token.
 
         Args:
@@ -141,9 +138,7 @@ def protected_route(user: Dict = Depends(get_current_user)):
     # Extract user info from payload
     email = payload.get("sub")
     if email is None:
-        raise HTTPException(
-            status_code=401, detail="Invalid authentication credentials"
-        )
+        raise HTTPException(status_code=401, detail="Invalid authentication credentials")
 
     return payload
 
@@ -160,9 +155,7 @@ def admin_route(user: Dict = Depends(require_role("Super Admin"))):
     def role_checker(current_user: Dict = Depends(get_current_user)):
         user_role = current_user.get("role")
         if user_role != required_role:
-            raise HTTPException(
-                status_code=403, detail=f"Access denied. Required role: {required_role}"
-            )
+            raise HTTPException(status_code=403, detail=f"Access denied. Required role: {required_role}")
         return current_user
 
     return role_checker
@@ -241,9 +234,7 @@ class RateLimiter:
     def __init__(self):
         self.requests = {}
 
-    def is_allowed(
-        self, identifier: str, max_requests: int = 60, window_seconds: int = 60
-    ) -> bool:
+    def is_allowed(self, identifier: str, max_requests: int = 60, window_seconds: int = 60) -> bool:
         """Check if request is allowed under rate limit.
 
         Args:
@@ -261,9 +252,7 @@ def is_allowed(
 
         # Clean old requests
         cutoff = now - timedelta(seconds=window_seconds)
-        self.requests[identifier] = [
-            req_time for req_time in self.requests[identifier] if req_time > cutoff
-        ]
+        self.requests[identifier] = [req_time for req_time in self.requests[identifier] if req_time > cutoff]
 
         # Check limit
         if len(self.requests[identifier]) >= max_requests:

CogniwareIms/backend/app/init_knowledge_base.py

@@ -1,6 +1,5 @@
 # Copyright (C) 2024 Intel Corporation
 # SPDX-License-Identifier: Apache-2.0
-
 """Knowledge Base Initialization Script Processes all CSV files and creates initial embeddings Run this after first
 deployment to populate the knowledge base."""
 
@@ -19,9 +18,7 @@
 from app.services.knowledge_manager import knowledge_manager
 from app.services.retrieval_service import retrieval_service
 
-logging.basicConfig(
-    level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
-)
+logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s")
 logger = logging.getLogger(__name__)
 
 
@@ -49,17 +46,13 @@ async def initialize_knowledge_base():
 
         for i in range(0, len(documents), batch_size):
             batch = documents[i : i + batch_size]
-            logger.info(
-                f"   Processing batch {i//batch_size + 1}/{(len(documents)-1)//batch_size + 1}..."
-            )
+            logger.info(f"   Processing batch {i//batch_size + 1}/{(len(documents)-1)//batch_size + 1}...")
 
             # Extract texts
             texts = [doc["text"] for doc in batch]
 
             # Generate embeddings in batch
-            embeddings = await embedding_service.embed_batch(
-                texts, batch_size=batch_size
-            )
+            embeddings = await embedding_service.embed_batch(texts, batch_size=batch_size)
 
             # Index each document
             for doc, embedding in zip(batch, embeddings):
@@ -97,11 +90,11 @@ async def initialize_knowledge_base():
         logger.info("\n" + "=" * 60)
         logger.info("🎉 Knowledge Base Initialization Complete!")
         logger.info("=" * 60)
-        logger.info(f"\n📊 Summary:")
+        logger.info("\n📊 Summary:")
         logger.info(f"   CSV Files Processed: {len(dataframes)}")
         logger.info(f"   Documents Indexed: {total_indexed}")
         logger.info(f"   Vector Store Count: {doc_count}")
-        logger.info(f"\n✅ System is ready for AI-powered queries!")
+        logger.info("\n✅ System is ready for AI-powered queries!")
 
         return {
             "success": True,
@@ -134,9 +127,7 @@ async def quick_test():
 
         # Test knowledge manager
         stats = await knowledge_manager.get_knowledge_stats()
-        logger.info(
-            f"✅ Knowledge manager: OK ({stats.get('total_documents', 0)} documents)"
-        )
+        logger.info(f"✅ Knowledge manager: OK ({stats.get('total_documents', 0)} documents)")
 
         logger.info("\n🎉 All systems operational!")
 

CogniwareIms/backend/app/main.py

@@ -1,6 +1,5 @@
 # Copyright (C) 2024 Intel Corporation
 # SPDX-License-Identifier: Apache-2.0
-
 """
 OPEA Inventory Management System - Complete Backend API
 Full integration with all OPEA GenAIComps microservices
@@ -160,9 +159,7 @@ async def health_check():
     db_health = await dbqna_service.health_check()
 
     return {
-        "status": (
-            "healthy" if all([embedding_health, llm_health, db_health]) else "degraded"
-        ),
+        "status": ("healthy" if all([embedding_health, llm_health, db_health]) else "degraded"),
         "timestamp": datetime.now().isoformat(),
         "services": {
             "api": "up",
@@ -325,9 +322,7 @@ async def upload_csv_knowledge(file: UploadFile = File(...)):
         content = await file.read()
 
         # Process using file upload service
-        result = await file_upload_service.upload_and_process(
-            filename=file.filename, content=content
-        )
+        result = await file_upload_service.upload_and_process(filename=file.filename, content=content)
 
         return result
 
@@ -345,9 +340,7 @@ async def upload_knowledge_file(file: UploadFile = File(...)):
         content = await file.read()
 
         # Process using file upload service
-        result = await file_upload_service.upload_and_process(
-            filename=file.filename, content=content
-        )
+        result = await file_upload_service.upload_and_process(filename=file.filename, content=content)
 
         return result
 
@@ -658,9 +651,7 @@ async def startup_event():
 
     # Load knowledge base stats
     stats = await knowledge_manager.get_knowledge_stats()
-    logger.info(
-        f"  Knowledge Base: {stats.get('total_documents', 0)} documents indexed"
-    )
+    logger.info(f"  Knowledge Base: {stats.get('total_documents', 0)} documents indexed")
 
     logger.info("✅ OPEA IMS Platform started successfully!")
 

CogniwareIms/backend/app/services/csv_processor.py

@@ -1,6 +1,5 @@
 # Copyright (C) 2024 Intel Corporation
 # SPDX-License-Identifier: Apache-2.0
-
 """CSV Data Processor Ingests CSV files and prepares them for OPEA knowledge base."""
 
 import json
@@ -37,9 +36,7 @@ def load_all_csv_files(self) -> Dict[str, pd.DataFrame]:
 
         return dataframes
 
-    def prepare_for_embedding(
-        self, dataframes: Dict[str, pd.DataFrame]
-    ) -> List[Dict[str, Any]]:
+    def prepare_for_embedding(self, dataframes: Dict[str, pd.DataFrame]) -> List[Dict[str, Any]]:
         """Prepare data for OPEA embedding service."""
         documents = []
 
@@ -66,9 +63,7 @@ def prepare_for_embedding(
         logger.info(f"Prepared {len(documents)} documents for embedding")
         return documents
 
-    def extract_inventory_data(
-        self, dataframes: Dict[str, pd.DataFrame]
-    ) -> Dict[str, Any]:
+    def extract_inventory_data(self, dataframes: Dict[str, pd.DataFrame]) -> Dict[str, Any]:
         """Extract structured inventory data."""
         inventory_data = {
             "products": [],
@@ -117,9 +112,7 @@ def create_knowledge_base(self) -> Dict[str, Any]:
         with open(output_dir / "knowledge_base.json", "w") as f:
             json.dump(knowledge_base, f, indent=2, default=str)
 
-        logger.info(
-            f"Knowledge base created with {len(knowledge_base['documents'])} documents"
-        )
+        logger.info(f"Knowledge base created with {len(knowledge_base['documents'])} documents")
 
         return knowledge_base
 

CogniwareIms/backend/app/services/dbqna_service.py

@@ -1,6 +1,5 @@
 # Copyright (C) 2024 Intel Corporation
 # SPDX-License-Identifier: Apache-2.0
-
 """
 OPEA DBQnA Service - Database Query & Answer
 Converts natural language to SQL and executes against inventory database
@@ -22,9 +21,7 @@ class DBQnAService:
     """Database Query & Answer service using OPEA LLM for SQL generation."""
 
     def __init__(self):
-        self.database_url = os.getenv(
-            "DATABASE_URL", "postgresql://postgres:postgres@postgres:5432/opea_ims"
-        )
+        self.database_url = os.getenv("DATABASE_URL", "postgresql://postgres:postgres@postgres:5432/opea_ims")
         self.engine = None
         self.schema_cache = None
 
@@ -66,15 +63,9 @@ async def get_schema(self) -> Dict[str, Any]:
                         ORDER BY ordinal_position
                     """
                     )
-                    columns = conn.execute(
-                        columns_query, {"table_name": table_name}
-                    ).fetchall()
-
-                    schema["tables"][table_name] = {
-                        "columns": [
-                            {"name": col, "type": dtype} for col, dtype in columns
-                        ]
-                    }
+                    columns = conn.execute(columns_query, {"table_name": table_name}).fetchall()
+
+                    schema["tables"][table_name] = {"columns": [{"name": col, "type": dtype} for col, dtype in columns]}
 
             self.schema_cache = schema
             return schema
@@ -83,9 +74,7 @@ async def get_schema(self) -> Dict[str, Any]:
             logger.error(f"Error getting schema: {e}")
             return {"tables": {}, "relationships": []}
 
-    async def natural_language_query(
-        self, question: str, include_explanation: bool = True
-    ) -> Dict[str, Any]:
+    async def natural_language_query(self, question: str, include_explanation: bool = True) -> Dict[str, Any]:
         """Convert natural language question to SQL and execute.
 
         Args:
@@ -112,9 +101,7 @@ async def natural_language_query(
                 columns = result.keys()
 
                 # Convert to dict format
-                data = [
-                    {col: value for col, value in zip(columns, row)} for row in rows
-                ]
+                data = [{col: value for col, value in zip(columns, row)} for row in rows]
 
             response = {
                 "success": True,