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
102 lines
3.1 KiB
Python
102 lines
3.1 KiB
Python
"""Configuration utility functions and global settings management."""
|
|
|
|
from typing import Any
|
|
|
|
import hvac
|
|
import redis.asyncio as redis
|
|
from minio import Minio
|
|
from qdrant_client import QdrantClient
|
|
|
|
from libs.events.base import EventBus
|
|
|
|
from .factories import (
|
|
MinIOClientFactory,
|
|
Neo4jDriverFactory,
|
|
QdrantClientFactory,
|
|
RedisClientFactory,
|
|
VaultClientFactory,
|
|
)
|
|
from .settings import BaseAppSettings
|
|
|
|
# Global settings instance
|
|
_settings: BaseAppSettings | None = None
|
|
|
|
|
|
def get_settings() -> BaseAppSettings:
|
|
"""Get global settings instance"""
|
|
global _settings # pylint: disable=global-variable-not-assigned
|
|
if _settings is None:
|
|
raise RuntimeError("Settings not initialized. Call init_settings() first.")
|
|
return _settings
|
|
|
|
|
|
def init_settings(
|
|
settings_class: type[BaseAppSettings] = BaseAppSettings, **kwargs: Any
|
|
) -> BaseAppSettings:
|
|
"""Initialize settings with custom class"""
|
|
global _settings # pylint: disable=global-statement
|
|
_settings = settings_class(**kwargs)
|
|
return _settings
|
|
|
|
|
|
# Convenience functions for backward compatibility
|
|
def create_vault_client(settings: BaseAppSettings) -> hvac.Client:
|
|
"""Create Vault client"""
|
|
return VaultClientFactory.create_client(settings)
|
|
|
|
|
|
def create_minio_client(settings: BaseAppSettings) -> Minio:
|
|
"""Create MinIO client"""
|
|
return MinIOClientFactory.create_client(settings)
|
|
|
|
|
|
def create_qdrant_client(settings: BaseAppSettings) -> QdrantClient:
|
|
"""Create Qdrant client"""
|
|
return QdrantClientFactory.create_client(settings)
|
|
|
|
|
|
def create_neo4j_client(settings: BaseAppSettings) -> Any:
|
|
"""Create Neo4j driver"""
|
|
return Neo4jDriverFactory.create_driver(settings)
|
|
|
|
|
|
async def create_redis_client(settings: BaseAppSettings) -> "redis.Redis[str]":
|
|
"""Create Redis client"""
|
|
return await RedisClientFactory.create_client(settings)
|
|
|
|
|
|
def create_event_bus(settings: BaseAppSettings) -> EventBus:
|
|
"""Create event bus"""
|
|
# pylint: disable=import-outside-toplevel
|
|
from libs.events import create_event_bus as _create_event_bus
|
|
|
|
# Extract NATS servers as a list
|
|
nats_servers = [s.strip() for s in settings.nats_servers.split(",")]
|
|
|
|
return _create_event_bus(
|
|
settings.event_bus_type,
|
|
servers=nats_servers,
|
|
stream_name=settings.nats_stream_name,
|
|
consumer_group=settings.nats_consumer_group,
|
|
bootstrap_servers=settings.kafka_bootstrap_servers,
|
|
region_name=settings.aws_region,
|
|
)
|
|
|
|
|
|
def get_default_settings(**overrides: Any) -> BaseAppSettings:
|
|
"""Get default settings with optional overrides"""
|
|
defaults = {
|
|
"service_name": "default-service",
|
|
"vault_addr": "http://vault:8200",
|
|
"postgres_url": "postgresql://user:pass@postgres:5432/taxagent",
|
|
"neo4j_uri": "bolt://neo4j:7687",
|
|
"neo4j_password": "password",
|
|
"redis_url": "redis://redis:6379",
|
|
"minio_endpoint": "minio:9000",
|
|
"minio_access_key": "minioadmin",
|
|
"minio_secret_key": "minioadmin",
|
|
"qdrant_url": "http://qdrant:6333",
|
|
}
|
|
defaults.update(overrides)
|
|
return BaseAppSettings(**defaults) # type: ignore
|