Robert Helewka
37bb38ee43
Update `_check_s3` to read S3 settings from the `STORAGES` dict instead of deprecated top-level `AWS_*` settings. Skip the check when local storage is enabled and return an error early if no bucket is configured.
103 lines
2.9 KiB
Python
103 lines
2.9 KiB
Python
"""
|
|
Mnemosyne project-level views — landing page and dashboard.
|
|
"""
|
|
|
|
from django.contrib.auth.decorators import login_required
|
|
from django.core.cache import cache
|
|
from django.db import connection
|
|
from django.http import JsonResponse
|
|
from django.shortcuts import render
|
|
|
|
from llm_manager.models import LLMApi, LLMModel
|
|
|
|
|
|
def landing(request):
|
|
"""
|
|
Public landing page. Redirects authenticated users to dashboard.
|
|
"""
|
|
if request.user.is_authenticated:
|
|
return dashboard(request)
|
|
return render(request, "mnemosyne/landing.html")
|
|
|
|
|
|
@login_required
|
|
def dashboard(request):
|
|
"""
|
|
Authenticated user dashboard — overview of all Mnemosyne features.
|
|
"""
|
|
context = {
|
|
"api_count": LLMApi.objects.filter(is_active=True).count(),
|
|
"model_count": LLMModel.objects.filter(is_active=True).count(),
|
|
"system_embedding": LLMModel.get_system_embedding_model(),
|
|
"system_chat": LLMModel.get_system_chat_model(),
|
|
}
|
|
|
|
# Try to get library counts (Neo4j may be unavailable)
|
|
try:
|
|
from library.utils import neo4j_available
|
|
|
|
if neo4j_available():
|
|
from library.models import Library
|
|
|
|
context["library_count"] = len(Library.nodes.all())
|
|
else:
|
|
context["library_count"] = None
|
|
except Exception:
|
|
context["library_count"] = None
|
|
|
|
return render(request, "mnemosyne/dashboard.html", context)
|
|
|
|
|
|
def live(request):
|
|
return JsonResponse({"status": "ok"})
|
|
|
|
|
|
def _check_s3() -> str | None:
|
|
"""Return an error string if the Mnemosyne S3 bucket is unreachable, else None."""
|
|
import boto3
|
|
import botocore.exceptions
|
|
from django.conf import settings
|
|
|
|
if getattr(settings, "USE_LOCAL_STORAGE", False):
|
|
return None
|
|
|
|
opts = settings.STORAGES.get("default", {}).get("OPTIONS", {})
|
|
bucket = opts.get("bucket_name", "")
|
|
if not bucket:
|
|
return "No S3 bucket configured"
|
|
|
|
try:
|
|
client = boto3.client(
|
|
"s3",
|
|
endpoint_url=opts.get("endpoint_url") or None,
|
|
aws_access_key_id=opts.get("access_key") or None,
|
|
aws_secret_access_key=opts.get("secret_key") or None,
|
|
region_name=opts.get("region_name") or None,
|
|
use_ssl=opts.get("use_ssl", True),
|
|
verify=opts.get("verify", True),
|
|
)
|
|
client.head_bucket(Bucket=bucket)
|
|
return None
|
|
except botocore.exceptions.ClientError as e:
|
|
return f"HTTP {e.response['Error']['Code']}"
|
|
except Exception as e:
|
|
return str(e)
|
|
|
|
|
|
def ready(request):
|
|
errors = {}
|
|
try:
|
|
connection.ensure_connection()
|
|
except Exception as e:
|
|
errors["db"] = str(e)
|
|
try:
|
|
cache.get("__readiness_probe__")
|
|
except Exception as e:
|
|
errors["cache"] = str(e)
|
|
s3_error = _check_s3()
|
|
if s3_error:
|
|
errors["s3"] = s3_error
|
|
if errors:
|
|
return JsonResponse({"status": "error", "errors": errors}, status=503)
|
|
return JsonResponse({"status": "ok"})
|