49 lines
1.4 KiB
Python
49 lines
1.4 KiB
Python
# app/deps.py
|
|
import os
|
|
import pymysql
|
|
from pymysql.cursors import DictCursor
|
|
from fastapi import Depends, Request, HTTPException, status
|
|
from qdrant_client import QdrantClient
|
|
from config import get_settings
|
|
|
|
def get_db():
|
|
# Einfache, robuste DB-Verbindung pro Aufruf
|
|
conn = pymysql.connect(
|
|
host=os.getenv("MARIADB_HOST", "db"),
|
|
user=os.getenv("MARIADB_USER", "crumb"),
|
|
password=os.getenv("MARIADB_PASSWORD", "secret"),
|
|
database=os.getenv("MARIADB_DATABASE", "crumbcrm"),
|
|
autocommit=True,
|
|
charset="utf8mb4",
|
|
cursorclass=DictCursor,
|
|
)
|
|
return conn
|
|
|
|
# Singleton Qdrant client
|
|
_qdrant_client = None
|
|
|
|
def get_qdrant_client() -> QdrantClient:
|
|
"""
|
|
Get or create Qdrant client instance (singleton pattern).
|
|
"""
|
|
global _qdrant_client
|
|
if _qdrant_client is None:
|
|
settings = get_settings()
|
|
_qdrant_client = QdrantClient(
|
|
host=settings.qdrant_host,
|
|
port=settings.qdrant_port
|
|
)
|
|
return _qdrant_client
|
|
|
|
def current_user(req: Request):
|
|
return req.session.get("user")
|
|
|
|
def admin_required(user = Depends(current_user)):
|
|
if not user:
|
|
# 401 -> Login
|
|
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="login required")
|
|
if user.get("role") != "admin":
|
|
# 403 -> Nicht genug Rechte
|
|
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="admin only")
|
|
return user
|