Files
service-finder/backend/app/services/config_service.py
Kincses 451900ae1a feat: implement hybrid address system and premium search logic
- Added centralized, self-learning GeoService (ZIP, City, Street)
- Implemented Hybrid Address Management (Centralized table + Denormalized fields)
- Fixed Gamification logic (PointsLedger field names & filtering)
- Added address autocomplete and two-tier (Free/Premium) search API
- Synchronized UserStats and PointsLedger schemas
2026-02-08 16:26:39 +00:00

63 lines
2.0 KiB
Python
Executable File

from typing import Any, Optional, Dict
import logging
from sqlalchemy import text
from app.db.session import SessionLocal
logger = logging.getLogger(__name__)
class ConfigService:
def __init__(self):
self._cache: Dict[str, Any] = {}
async def get_setting(
self,
key: str,
org_id: Optional[int] = None,
region_code: Optional[str] = None,
tier_id: Optional[int] = None,
default: Any = None
) -> Any:
# 1. Cache kulcs generálása (hierarchiát is figyelembe véve)
cache_key = f"{key}_{org_id}_{tier_id}_{region_code}"
if cache_key in self._cache:
return self._cache[cache_key]
query = text("""
SELECT value_json
FROM data.system_settings
WHERE key_name = :key
AND (
(org_id = :org_id) OR
(org_id IS NULL AND tier_id = :tier_id) OR
(org_id IS NULL AND tier_id IS NULL AND region_code = :region_code) OR
(org_id IS NULL AND tier_id IS NULL AND region_code IS NULL)
)
ORDER BY
(org_id IS NOT NULL) DESC,
(tier_id IS NOT NULL) DESC,
(region_code IS NOT NULL) DESC
LIMIT 1
""")
try:
async with SessionLocal() as db:
result = await db.execute(query, {
"key": key,
"org_id": org_id,
"tier_id": tier_id,
"region_code": region_code
})
row = result.fetchone()
val = row[0] if row else default
# 2. Mentés cache-be
self._cache[cache_key] = val
return val
except Exception as e:
logger.error(f"ConfigService Error: {e}")
return default
def clear_cache(self):
self._cache = {}
config = ConfigService()