fdba81809f
Some checks failed
CI/CD Pipeline / Generate SBOM (push) Has been cancelled
CI/CD Pipeline / Deploy to Staging (push) Has been cancelled
CI/CD Pipeline / Deploy to Production (push) Has been cancelled
CI/CD Pipeline / Code Quality & Linting (push) Has been cancelled
CI/CD Pipeline / Policy Validation (push) Has been cancelled
CI/CD Pipeline / Test Suite (push) Has been cancelled
CI/CD Pipeline / Build Docker Images (svc-coverage) (push) Has been cancelled
CI/CD Pipeline / Build Docker Images (svc-extract) (push) Has been cancelled
CI/CD Pipeline / Build Docker Images (svc-firm-connectors) (push) Has been cancelled
CI/CD Pipeline / Build Docker Images (svc-forms) (push) Has been cancelled
CI/CD Pipeline / Build Docker Images (svc-hmrc) (push) Has been cancelled
CI/CD Pipeline / Build Docker Images (svc-ingestion) (push) Has been cancelled
CI/CD Pipeline / Build Docker Images (svc-kg) (push) Has been cancelled
CI/CD Pipeline / Build Docker Images (svc-normalize-map) (push) Has been cancelled
CI/CD Pipeline / Build Docker Images (svc-ocr) (push) Has been cancelled
CI/CD Pipeline / Build Docker Images (svc-rag-indexer) (push) Has been cancelled
CI/CD Pipeline / Build Docker Images (svc-rag-retriever) (push) Has been cancelled
CI/CD Pipeline / Build Docker Images (svc-reason) (push) Has been cancelled
CI/CD Pipeline / Build Docker Images (svc-rpa) (push) Has been cancelled
CI/CD Pipeline / Build Docker Images (ui-review) (push) Has been cancelled
CI/CD Pipeline / Security Scanning (svc-coverage) (push) Has been cancelled
CI/CD Pipeline / Security Scanning (svc-extract) (push) Has been cancelled
CI/CD Pipeline / Security Scanning (svc-kg) (push) Has been cancelled
CI/CD Pipeline / Security Scanning (svc-rag-retriever) (push) Has been cancelled
CI/CD Pipeline / Security Scanning (ui-review) (push) Has been cancelled
CI/CD Pipeline / Notifications (push) Has been cancelled
88 lines
2.8 KiB
Python
88 lines
2.8 KiB
Python
"""Client factories for various services."""
|
|
|
|
from typing import Any
|
|
|
|
import hvac
|
|
import redis.asyncio as redis
|
|
from minio import Minio
|
|
from neo4j import GraphDatabase
|
|
from qdrant_client import QdrantClient
|
|
|
|
from .settings import BaseAppSettings
|
|
|
|
|
|
class VaultClientFactory: # pylint: disable=too-few-public-methods
|
|
"""Factory for creating Vault clients"""
|
|
|
|
@staticmethod
|
|
def create_client(settings: BaseAppSettings) -> hvac.Client:
|
|
"""Create authenticated Vault client"""
|
|
client = hvac.Client(url=settings.vault_addr)
|
|
|
|
if settings.vault_token:
|
|
# Development mode with token
|
|
client.token = settings.vault_token
|
|
elif settings.vault_role_id and settings.vault_secret_id:
|
|
# Production mode with AppRole
|
|
try:
|
|
auth_response = client.auth.approle.login(
|
|
role_id=settings.vault_role_id, secret_id=settings.vault_secret_id
|
|
)
|
|
client.token = auth_response["auth"]["client_token"]
|
|
except Exception as e:
|
|
raise ValueError("Failed to authenticate with Vault") from e
|
|
else:
|
|
raise ValueError(
|
|
"Either vault_token or vault_role_id/vault_secret_id must be provided"
|
|
)
|
|
|
|
if not client.is_authenticated():
|
|
raise ValueError("Failed to authenticate with Vault")
|
|
|
|
return client
|
|
|
|
|
|
class MinIOClientFactory: # pylint: disable=too-few-public-methods
|
|
"""Factory for creating MinIO clients"""
|
|
|
|
@staticmethod
|
|
def create_client(settings: BaseAppSettings) -> Minio:
|
|
"""Create MinIO client"""
|
|
return Minio(
|
|
endpoint=settings.minio_endpoint,
|
|
access_key=settings.minio_access_key,
|
|
secret_key=settings.minio_secret_key,
|
|
secure=settings.minio_secure,
|
|
)
|
|
|
|
|
|
class QdrantClientFactory: # pylint: disable=too-few-public-methods
|
|
"""Factory for creating Qdrant clients"""
|
|
|
|
@staticmethod
|
|
def create_client(settings: BaseAppSettings) -> QdrantClient:
|
|
"""Create Qdrant client"""
|
|
return QdrantClient(url=settings.qdrant_url, api_key=settings.qdrant_api_key)
|
|
|
|
|
|
class Neo4jDriverFactory: # pylint: disable=too-few-public-methods
|
|
"""Factory for creating Neo4j drivers"""
|
|
|
|
@staticmethod
|
|
def create_driver(settings: BaseAppSettings) -> Any:
|
|
"""Create Neo4j driver"""
|
|
return GraphDatabase.driver(
|
|
settings.neo4j_uri, auth=(settings.neo4j_user, settings.neo4j_password)
|
|
)
|
|
|
|
|
|
class RedisClientFactory: # pylint: disable=too-few-public-methods
|
|
"""Factory for creating Redis clients"""
|
|
|
|
@staticmethod
|
|
async def create_client(settings: BaseAppSettings) -> "redis.Redis[str]":
|
|
"""Create Redis client"""
|
|
return redis.from_url(
|
|
settings.redis_url, encoding="utf-8", decode_responses=True
|
|
)
|