bd33fbad01
Frontend: - DetectionsList: Simplify columns, improve truncation and display for IPs, hosts, bot info - IncidentsView: Replace metric cards with compact stat cards (unique IPs, known bots, ML anomalies, threat levels) - InvestigationView: Add section navigation anchors, reorganize layout with proper IDs - ThreatIntelView: Add navigation links to investigation pages, add comment column, improve table layout Backend: - Various route and model adjustments - Configuration updates Co-authored-by: Qwen-Coder <qwen-coder@alibabacloud.com>
52 lines
1.5 KiB
Python
52 lines
1.5 KiB
Python
"""
|
|
Connexion à ClickHouse
|
|
"""
|
|
import clickhouse_connect
|
|
from typing import Optional
|
|
from .config import settings
|
|
|
|
|
|
class ClickHouseClient:
|
|
"""Gestionnaire de connexion ClickHouse"""
|
|
|
|
def __init__(self):
|
|
self._client: Optional[clickhouse_connect.driver.client.Client] = None
|
|
|
|
def connect(self) -> clickhouse_connect.driver.client.Client:
|
|
"""Établit la connexion à ClickHouse"""
|
|
if self._client is None or not self._ping():
|
|
self._client = clickhouse_connect.get_client(
|
|
host=settings.CLICKHOUSE_HOST,
|
|
port=settings.CLICKHOUSE_PORT,
|
|
database=settings.CLICKHOUSE_DB,
|
|
user=settings.CLICKHOUSE_USER,
|
|
password=settings.CLICKHOUSE_PASSWORD,
|
|
connect_timeout=10
|
|
)
|
|
return self._client
|
|
|
|
def _ping(self) -> bool:
|
|
"""Vérifie si la connexion est active"""
|
|
try:
|
|
if self._client:
|
|
self._client.ping()
|
|
return True
|
|
except Exception:
|
|
pass
|
|
return False
|
|
|
|
def query(self, query: str, params: Optional[dict] = None):
|
|
"""Exécute une requête SELECT"""
|
|
client = self.connect()
|
|
return client.query(query, params)
|
|
|
|
def close(self):
|
|
"""Ferme la connexion"""
|
|
if self._client:
|
|
self._client.close()
|
|
self._client = None
|
|
|
|
|
|
# Instance globale
|
|
db = ClickHouseClient()
|