flare / config_provider.py
ciyidogan's picture
Update config_provider.py
f5a3621 verified
raw
history blame
24.8 kB
"""
Thread-Safe Configuration Provider for Flare Platform
"""
import threading
import os
import json
import commentjson
from typing import Optional, Dict, List, Any
from datetime import datetime
from pathlib import Path
import tempfile
import shutil
from utils import get_current_timestamp, normalize_timestamp, timestamps_equal
from config_models import (
ServiceConfig, GlobalConfig, ProjectConfig, VersionConfig,
IntentConfig, APIConfig, ActivityLogEntry, ParameterConfig
)
from logger import log_info, log_error, log_warning, log_debug, LogTimer
from exceptions import (
RaceConditionError, ConfigurationError, ResourceNotFoundError,
DuplicateResourceError, ValidationError
)
from encryption_utils import encrypt, decrypt
class ConfigProvider:
"""Thread-safe singleton configuration provider"""
_instance: Optional[ServiceConfig] = None
_lock = threading.RLock() # Reentrant lock for nested calls
_file_lock = threading.Lock() # Separate lock for file operations
_CONFIG_PATH = Path(__file__).parent / "service_config.jsonc"
@staticmethod
def _normalize_date(date_str: Optional[str]) -> str:
"""Normalize date string for comparison"""
if not date_str:
return ""
return date_str.replace(' ', 'T').replace('+00:00', 'Z').replace('.000Z', 'Z')
@classmethod
def get(cls) -> ServiceConfig:
"""Get cached configuration - thread-safe"""
if cls._instance is None:
with cls._lock:
# Double-checked locking pattern
if cls._instance is None:
with LogTimer("config_load"):
cls._instance = cls._load()
cls._instance.build_index()
log_info("Configuration loaded successfully")
return cls._instance
@classmethod
def reload(cls) -> ServiceConfig:
"""Force reload configuration from file"""
with cls._lock:
log_info("Reloading configuration...")
cls._instance = None
return cls.get()
@classmethod
def _load(cls) -> ServiceConfig:
"""Load configuration from file"""
try:
if not cls._CONFIG_PATH.exists():
raise ConfigurationError(
f"Config file not found: {cls._CONFIG_PATH}",
config_key="service_config.jsonc"
)
with open(cls._CONFIG_PATH, 'r', encoding='utf-8') as f:
config_data = commentjson.load(f)
# Debug: İlk project'in tarihini kontrol et
if 'projects' in config_data and len(config_data['projects']) > 0:
first_project = config_data['projects'][0]
log_debug(f"🔍 Raw project data - last_update_date: {first_project.get('last_update_date')}")
# Ensure required fields
if 'config' not in config_data:
config_data['config'] = {}
# Parse API configs (handle JSON strings)
if 'apis' in config_data:
cls._parse_api_configs(config_data['apis'])
# Validate and create model
cfg = ServiceConfig.model_validate(config_data)
# Debug: Model'e dönüştükten sonra kontrol et
if cfg.projects and len(cfg.projects) > 0:
log_debug(f"🔍 Parsed project - last_update_date: {cfg.projects[0].last_update_date}")
log_debug(f"🔍 Type: {type(cfg.projects[0].last_update_date)}")
log_debug(
"Configuration loaded",
projects=len(cfg.projects),
apis=len(cfg.apis),
users=len(cfg.global_config.users)
)
return cfg
except Exception as e:
log_error(f"Error loading config", error=str(e), path=str(cls._CONFIG_PATH))
raise ConfigurationError(f"Failed to load configuration: {e}")
@classmethod
def _parse_api_configs(cls, apis: List[Dict[str, Any]]) -> None:
"""Parse JSON string fields in API configs"""
for api in apis:
# Parse headers
if 'headers' in api and isinstance(api['headers'], str):
try:
api['headers'] = json.loads(api['headers'])
except json.JSONDecodeError:
api['headers'] = {}
# Parse body_template
if 'body_template' in api and isinstance(api['body_template'], str):
try:
api['body_template'] = json.loads(api['body_template'])
except json.JSONDecodeError:
api['body_template'] = {}
# Parse auth configs
if 'auth' in api and api['auth']:
cls._parse_auth_config(api['auth'])
@classmethod
def _parse_auth_config(cls, auth: Dict[str, Any]) -> None:
"""Parse auth configuration"""
# Parse token_request_body
if 'token_request_body' in auth and isinstance(auth['token_request_body'], str):
try:
auth['token_request_body'] = json.loads(auth['token_request_body'])
except json.JSONDecodeError:
auth['token_request_body'] = {}
# Parse token_refresh_body
if 'token_refresh_body' in auth and isinstance(auth['token_refresh_body'], str):
try:
auth['token_refresh_body'] = json.loads(auth['token_refresh_body'])
except json.JSONDecodeError:
auth['token_refresh_body'] = {}
@classmethod
def save(cls, config: ServiceConfig, username: str) -> None:
"""Thread-safe configuration save with optimistic locking"""
with cls._file_lock:
try:
# Load current config for race condition check
current_config = cls._load()
# Check for race condition
if config.last_update_date and current_config.last_update_date:
if not timestamps_equal(config.last_update_date, current_config.last_update_date):
raise RaceConditionError(
"Configuration was modified by another user",
current_user=username,
last_update_user=current_config.last_update_user,
last_update_date=current_config.last_update_date,
entity_type="configuration"
)
# Update metadata
config.last_update_date = get_current_timestamp()
config.last_update_user = username
# Convert to JSON - Pydantic v2 kullanımı
data = config.model_dump(mode='json') # to_jsonc_dict yerine model_dump
json_str = json.dumps(data, ensure_ascii=False, indent=2)
# Önce /tmp'de oluştur
with tempfile.NamedTemporaryFile(
mode='w',
encoding='utf-8',
dir='/tmp', # /app yerine /tmp kullan
delete=False,
prefix='flare_config_',
suffix='.jsonc'
) as tmp_file:
tmp_file.write(json_str)
temp_path = Path(tmp_file.name)
# Sonra hedef konuma taşı
import shutil
shutil.move(str(temp_path), str(cls._CONFIG_PATH))
# Update cache
with cls._lock:
cls._instance = config
cls._instance.build_index()
log_info(
"Configuration saved",
user=username,
size=len(json_str)
)
except Exception as e:
log_error("Failed to save configuration", error=str(e), user=username)
raise
# ===================== Environment Methods =====================
@classmethod
def update_environment(cls, update_data: dict, username: str) -> None:
"""Update environment configuration"""
with cls._lock:
config = cls.get()
# Update providers
if 'llm_provider' in update_data:
config.global_config.llm_provider = update_data['llm_provider']
if 'tts_provider' in update_data:
config.global_config.tts_provider = update_data['tts_provider']
if 'stt_provider' in update_data:
config.global_config.stt_provider = update_data['stt_provider']
# Log activity
cls._add_activity(
config, username, "UPDATE_ENVIRONMENT",
"environment", None, None,
f"Updated providers"
)
# Save
cls.save(config, username)
# ===================== Project Methods =====================
@classmethod
def get_project(cls, project_id: int) -> Optional[ProjectConfig]:
"""Get project by ID"""
config = cls.get()
return next((p for p in config.projects if p.id == project_id), None)
@classmethod
def create_project(cls, project_data: dict, username: str) -> ProjectConfig:
"""Create new project with thread safety"""
with cls._lock:
config = cls.get()
# Check for duplicate name
if any(p.name == project_data['name'] for p in config.projects):
raise DuplicateResourceError("project", project_data['name'])
# Create project
project = ProjectConfig(
id=config.project_id_counter,
created_date=get_current_timestamp(),
created_by=username,
**project_data
)
# Update config
config.projects.append(project)
config.project_id_counter += 1
# Log activity
cls._add_activity(
config, username, "CREATE_PROJECT",
"project", project.id, project.name
)
# Save
cls.save(config, username)
log_info(
"Project created",
project_id=project.id,
name=project.name,
user=username
)
return project
@classmethod
def update_project(cls, project_id: int, update_data: dict, username: str, expected_last_update: Optional[str] = None) -> ProjectConfig:
"""Update project with optimistic locking"""
with cls._lock:
config = cls.get()
project = cls.get_project(project_id)
if not project:
raise ResourceNotFoundError("project", project_id)
# Check race condition
if expected_last_update is not None and expected_last_update != '':
if project.last_update_date and not timestamps_equal(expected_last_update, project.last_update_date):
raise RaceConditionError(
f"Project '{project.name}' was modified by another user",
current_user=username,
last_update_user=project.last_update_user,
last_update_date=project.last_update_date,
entity_type="project",
entity_id=project_id
)
# Update fields
for key, value in update_data.items():
if hasattr(project, key) and key not in ['id', 'created_date', 'created_by', 'last_update_date', 'last_update_user']:
setattr(project, key, value)
project.last_update_date = get_current_timestamp()
project.last_update_user = username
# Log activity
cls._add_activity(
config, username, "UPDATE_PROJECT",
"project", project.id, project.name
)
# Save
cls.save(config, username)
log_info(
"Project updated",
project_id=project.id,
user=username
)
return project
@classmethod
def delete_project(cls, project_id: int, username: str) -> None:
"""Soft delete project"""
with cls._lock:
config = cls.get()
project = cls.get_project(project_id)
if not project:
raise ResourceNotFoundError("project", project_id)
project.deleted = True
project.last_update_date = get_current_timestamp()
project.last_update_user = username
# Log activity
cls._add_activity(
config, username, "DELETE_PROJECT",
"project", project.id, project.name
)
# Save
cls.save(config, username)
log_info(
"Project deleted",
project_id=project.id,
user=username
)
# ===================== Version Methods =====================
@classmethod
def create_version(cls, project_id: int, version_data: dict, username: str) -> VersionConfig:
"""Create new version"""
with cls._lock:
config = cls.get()
project = cls.get_project(project_id)
if not project:
raise ResourceNotFoundError("project", project_id)
# Create version
version = VersionConfig(
no=project.version_id_counter,
created_date=get_current_timestamp(),
created_by=username,
**version_data
)
# Update project
project.versions.append(version)
project.version_id_counter += 1
project.last_update_date = get_current_timestamp()
project.last_update_user = username
# Log activity
cls._add_activity(
config, username, "CREATE_VERSION",
"version", version.no, f"{project.name} v{version.no}",
f"Project: {project.name}"
)
# Save
cls.save(config, username)
log_info(
"Version created",
project_id=project.id,
version_no=version.no,
user=username
)
return version
@classmethod
def publish_version(cls, project_id: int, version_no: int, username: str) -> VersionConfig:
"""Publish a version"""
with cls._lock:
config = cls.get()
project = cls.get_project(project_id)
if not project:
raise ResourceNotFoundError("project", project_id)
version = next((v for v in project.versions if v.no == version_no), None)
if not version:
raise ResourceNotFoundError("version", version_no)
# Unpublish other versions
for v in project.versions:
if v.published and v.no != version_no:
v.published = False
# Publish this version
version.published = True
version.publish_date = get_current_timestamp()
version.published_by = username
# Update project
project.last_update_date = get_current_timestamp()
project.last_update_user = username
# Log activity
cls._add_activity(
config, username, "PUBLISH_VERSION",
"version", version.no, f"{project.name} v{version.no}",
f"Published version {version.no}"
)
# Save
cls.save(config, username)
log_info(
"Version published",
project_id=project.id,
version_no=version.no,
user=username
)
return version
@classmethod
def update_version(cls, project_id: int, version_no: int, update_data: dict, username: str, expected_last_update: Optional[str] = None) -> VersionConfig:
"""Update version with optimistic locking"""
with cls._lock:
config = cls.get()
project = cls.get_project(project_id)
if not project:
raise ResourceNotFoundError("project", project_id)
version = next((v for v in project.versions if v.no == version_no), None)
if not version:
raise ResourceNotFoundError("version", version_no)
# Published versions cannot be edited
if version.published:
raise ValidationError("Published versions cannot be modified")
# Check race condition
if expected_last_update is not None and expected_last_update != '':
if version.last_update_date and not timestamps_equal(expected_last_update, version.last_update_date):
raise RaceConditionError(
f"Version '{version.no}' was modified by another user",
current_user=username,
last_update_user=version.last_update_user,
last_update_date=version.last_update_date,
entity_type="version",
entity_id=f"{project_id}:{version_no}"
)
# Update fields
for key, value in update_data.items():
if hasattr(version, key) and key not in ['no', 'created_date', 'created_by', 'published', 'last_update_date']:
setattr(version, key, value)
version.last_update_date = get_current_timestamp()
version.last_update_user = username
# Update project last update
project.last_update_date = get_current_timestamp()
project.last_update_user = username
# Log activity
cls._add_activity(
config, username, "UPDATE_VERSION",
"version", f"{project.id}:{version.no}", f"{project.name} v{version.no}"
)
# Save
cls.save(config, username)
log_info(
"Version updated",
project_id=project.id,
version_no=version.no,
user=username
)
return version
# ===================== API Methods =====================
@classmethod
def create_api(cls, api_data: dict, username: str) -> APIConfig:
"""Create new API"""
with cls._lock:
config = cls.get()
# Check for duplicate name
if any(a.name == api_data['name'] for a in config.apis):
raise DuplicateResourceError("api", api_data['name'])
# Create API
api = APIConfig(
created_date=get_current_timestamp(),
created_by=username,
**api_data
)
# Add to config
config.apis.append(api)
# Rebuild index
config.build_index()
# Log activity
cls._add_activity(
config, username, "CREATE_API",
"api", api.name, api.name
)
# Save
cls.save(config, username)
log_info(
"API created",
api_name=api.name,
user=username
)
return api
@classmethod
def update_api(cls, api_name: str, update_data: dict, username: str, expected_last_update: Optional[str] = None) -> APIConfig:
"""Update API with optimistic locking"""
with cls._lock:
config = cls.get()
api = config.get_api(api_name)
if not api:
raise ResourceNotFoundError("api", api_name)
# Check race condition
if expected_last_update is not None and expected_last_update != '':
if api.last_update_date and not timestamps_equal(expected_last_update, api.last_update_date):
raise RaceConditionError(
f"API '{api.name}' was modified by another user",
current_user=username,
last_update_user=api.last_update_user,
last_update_date=api.last_update_date,
entity_type="api",
entity_id=api.name
)
# Update fields
for key, value in update_data.items():
if hasattr(api, key) and key not in ['name', 'created_date', 'created_by', 'last_update_date']:
setattr(api, key, value)
api.last_update_date = get_current_timestamp()
api.last_update_user = username
# Rebuild index
config.build_index()
# Log activity
cls._add_activity(
config, username, "UPDATE_API",
"api", api.name, api.name
)
# Save
cls.save(config, username)
log_info(
"API updated",
api_name=api.name,
user=username
)
return api
@classmethod
def delete_api(cls, api_name: str, username: str) -> None:
"""Soft delete API"""
with cls._lock:
config = cls.get()
api = config.get_api(api_name)
if not api:
raise ResourceNotFoundError("api", api_name)
api.deleted = True
api.last_update_date = get_current_timestamp()
api.last_update_user = username
# Rebuild index
config.build_index()
# Log activity
cls._add_activity(
config, username, "DELETE_API",
"api", api.name, api.name
)
# Save
cls.save(config, username)
log_info(
"API deleted",
api_name=api.name,
user=username
)
# ===================== Activity Methods =====================
@classmethod
def _add_activity(
cls,
config: ServiceConfig,
username: str,
action: str,
entity_type: str,
entity_id: Any,
entity_name: Optional[str] = None,
details: Optional[str] = None
) -> None:
"""Add activity log entry"""
# Activity ID'sini oluştur - mevcut en yüksek ID'yi bul
max_id = 0
if config.activity_log:
max_id = max((entry.id for entry in config.activity_log if entry.id), default=0)
activity_id = max_id + 1
activity = ActivityLogEntry(
id=activity_id,
timestamp=get_current_timestamp(), # utils'ten import etmeyi unutma
username=username,
action=action,
entity_type=entity_type,
entity_id=str(entity_id) if entity_id else None,
entity_name=entity_name,
details=details
)
config.activity_log.append(activity)
# Keep only last 1000 entries
if len(config.activity_log) > 1000:
config.activity_log = config.activity_log[-1000:]