import sys
import time
import random
from decimal import Decimal
from django.conf import settings
from django.db import transaction, connections
from django.db.models import Prefetch, Q, Max, Count, OuterRef, Subquery, IntegerField
from django.db.models.functions import Coalesce
from django.forms.models import model_to_dict
from django.shortcuts import get_object_or_404, render
from rest_framework import status
from rest_framework.generics import (
ListCreateAPIView, CreateAPIView, ListAPIView,
UpdateAPIView, RetrieveAPIView, DestroyAPIView
)
from rest_framework.response import Response
from rest_framework.views import APIView
from cdb_rest.models import (
GlobalTag, GlobalTagStatus, PayloadList, PayloadType,
PayloadIOV, PayloadListIdSequence
)
from cdb_rest.serializers import (
GlobalTagCreateSerializer, GlobalTagReadSerializer,
GlobalTagStatusSerializer, GlobalTagListSerializer, GlobalTagDetailedSerializer,
PayloadListCreateSerializer, PayloadListReadSerializer,
PayloadTypeSerializer, PayloadIOVSerializer,
PayloadListSerializer, PayloadListReadShortSerializer,
GlobalTagBrowseSerializer, PayloadListBrowseSerializer,
PayloadIOVBrowseSerializer
)
import cdb_rest.queries
from .iov_comparisons import get_iov_config, compute_comb_iov
from .utils import load_permission_plugin, load_auth_class
[docs]
class WriteAuthMixin:
"""Require JWT authentication for write methods (POST/PUT/PATCH/DELETE), allow anonymous reads."""
[docs]
def get_authenticators(self):
if self.request and self.request.method in ('POST', 'PUT', 'PATCH', 'DELETE'):
auth_class = load_auth_class()
if auth_class:
return [auth_class()]
return []
[docs]
def paginate_browse(request, queryset, serializer_class, search_fields, sort_map, default_sort):
"""Server-side search/sort/pagination for browse endpoints (opt-in via ?page=).
Returns {count, page, page_size, total_pages, results}. Sort fields are
whitelisted through sort_map to prevent ordering by arbitrary columns.
"""
params = request.query_params
search = params.get('search', '').strip()
if search:
q = Q()
for field in search_fields:
q |= Q(**{field + '__icontains': search})
queryset = queryset.filter(q)
sort = params.get('sort', default_sort)
sort_field = sort_map.get(sort, sort_map[default_sort])
if params.get('order', 'asc') == 'desc':
sort_field = '-' + sort_field
queryset = queryset.order_by(sort_field, 'pk')
try:
page = max(1, int(params.get('page', 1)))
except (TypeError, ValueError):
page = 1
try:
page_size = min(1000, max(1, int(params.get('page_size', 25))))
except (TypeError, ValueError):
page_size = 25
total = queryset.count()
total_pages = max(1, (total + page_size - 1) // page_size)
start = (page - 1) * page_size
serializer = serializer_class(queryset[start:start + page_size], many=True)
return Response({
"count": total,
"page": page,
"page_size": page_size,
"total_pages": total_pages,
"results": serializer.data,
})
# ── GlobalTag views ──────────────────────────────────────────────────────────
[docs]
class GlobalTagDetailAPIView(WriteAuthMixin, RetrieveAPIView):
serializer_class = GlobalTagReadSerializer
queryset = GlobalTag.objects.all()
[docs]
class GlobalTagByNameDetailAPIView(WriteAuthMixin, RetrieveAPIView):
"""Retrieve a GlobalTag by name with all nested PayloadLists and IOVs.
Pass ?light=1 to get metadata only (no nested payload lists),
with payload list / IOV counts computed in the database.
"""
serializer_class = GlobalTagReadSerializer
queryset = GlobalTag.objects.all()
lookup_url_kwarg = 'globalTagName'
[docs]
def get_object(self):
gt_name = self.kwargs.get('globalTagName')
queryset = GlobalTag.objects.all()
obj = get_object_or_404(queryset, name=gt_name)
return obj
[docs]
def retrieve(self, request, *args, **kwargs):
if 'light' in request.query_params:
obj = get_object_or_404(
GlobalTag.objects.select_related('status').annotate(
payload_lists_count=Count('payload_lists', distinct=True),
payload_iov_count=Count('payload_lists__payload_iov', distinct=True),
),
name=self.kwargs.get('globalTagName'))
return Response(GlobalTagBrowseSerializer(obj).data)
return super().retrieve(request, *args, **kwargs)
[docs]
class TimeoutListAPIView(WriteAuthMixin, ListAPIView):
"""Test endpoint that simulates a long-running request (30 min timeout)."""
[docs]
def list(self, request):
time.sleep(1800)
return Response()
[docs]
class GlobalTagListCreationAPIView(WriteAuthMixin, ListCreateAPIView):
"""List all GlobalTags (GET) or create a new one (POST). Requires admin permission to create."""
serializer_class = GlobalTagCreateSerializer
[docs]
def get_queryset(self):
return GlobalTag.objects.all()
[docs]
def list(self, request):
queryset = self.get_queryset()
serializer = GlobalTagReadSerializer(queryset, many=True)
return Response(serializer.data)
[docs]
def create(self, request, *args, **kwargs):
data = request.data
plugin = load_permission_plugin()
target_object = {"object": "GlobalTag", "role": "admin", "name": data['name']}
if not plugin.has_permission(request, target_object):
return Response({"detail": "Permission denied."}, status=status.HTTP_403_FORBIDDEN)
try:
gt_status = GlobalTagStatus.objects.get(name=data['status'])
data['status'] = gt_status.pk
except KeyError:
return Response({"detail": "GlobalTagStatus not found."}, status=status.HTTP_500_INTERNAL_SERVER_ERROR)
serializer = self.get_serializer(data=request.data)
serializer.is_valid(raise_exception=True)
try:
instance = serializer.save()
except Exception as e:
return Response({"detail": "GlobalTag creation failed."}, status=status.HTTP_500_INTERNAL_SERVER_ERROR)
if instance.pk is None:
return Response({"detail": "GlobalTag was not saved to DB."}, status=500)
ret = serializer.data
ret['status'] = gt_status.name
return Response(ret)
[docs]
class GlobalTagDeleteAPIView(WriteAuthMixin, DestroyAPIView):
"""Delete a GlobalTag by name. Locked and frozen GTs are immutable."""
serializer_class = GlobalTagReadSerializer
lookup_url_kwarg = 'globalTagName'
lookup_field = 'name'
[docs]
def get_gtag(self):
try:
return GlobalTag.objects.get(name=self.kwargs['globalTagName'])
except GlobalTag.DoesNotExist:
return None
[docs]
def destroy(self, request, *args, **kwargs):
plugin = load_permission_plugin()
target_object = {"object": "GlobalTag", "role": "admin", "name": self.kwargs['globalTagName']}
if not plugin.has_permission(request, target_object):
return Response({"detail": "Permission denied."}, status=status.HTTP_403_FORBIDDEN)
gt = self.get_gtag()
if not gt:
return Response({"detail": "GlobalTag %s doesn't exist" % self.kwargs['globalTagName']}, status=status.HTTP_500_INTERNAL_SERVER_ERROR)
gt_status = GlobalTagStatus.objects.get(id=gt.status_id)
if gt_status.name in ['locked', 'frozen']:
return Response({"detail": "Global Tag is %s." % gt_status.name}, status=status.HTTP_500_INTERNAL_SERVER_ERROR)
ret = self.perform_destroy(gt)
if not ret:
ret = {"detail": "Global tag %s deleted." % gt.name}
return Response(ret)
# ── PayloadIOV views ─────────────────────────────────────────────────────────
[docs]
class PayloadIOVDeleteAPIView(WriteAuthMixin, DestroyAPIView):
"""Delete a PayloadIOV. Frozen GTs are immutable."""
serializer_class = PayloadIOVSerializer
[docs]
def get_object(self):
if 'major_iov_end' not in self.kwargs:
self.kwargs['major_iov_end'] = sys.maxsize
if 'minor_iov_end' not in self.kwargs:
self.kwargs['minor_iov_end'] = sys.maxsize
try:
return PayloadIOV.objects.get(
payload_list__global_tag__name=self.kwargs['globalTagName'],
payload_list__payload_type__name=self.kwargs['payloadType'],
major_iov=self.kwargs['major_iov'],
minor_iov=self.kwargs['minor_iov'],
major_iov_end=self.kwargs['major_iov_end'],
minor_iov_end=self.kwargs['minor_iov_end']
)
except:
return None
[docs]
def destroy(self, request, *args, **kwargs):
plugin = load_permission_plugin()
target_object = {"object": "GlobalTag", "role": "admin", "name": self.kwargs['globalTagName']}
if not plugin.has_permission(request, target_object):
return Response({"detail": "Permission denied."}, status=status.HTTP_403_FORBIDDEN)
piov = self.get_object()
if not piov:
return Response({"detail": "PayloadIOV with given parameters doesn't exist"}, status=status.HTTP_500_INTERNAL_SERVER_ERROR)
gt = GlobalTag.objects.get(name=self.kwargs['globalTagName'])
gt_status = GlobalTagStatus.objects.get(id=gt.status_id)
if gt_status.name == 'frozen':
return Response({"detail": "Global Tag is %s." % gt_status.name}, status=status.HTTP_500_INTERNAL_SERVER_ERROR)
ret = self.perform_destroy(piov)
if not ret:
ret = {"detail": "PayloadIOV %s deleted." % piov.payload_url}
return Response(ret)
# ── PayloadType views ────────────────────────────────────────────────────────
[docs]
class PayloadTypeDeleteAPIView(WriteAuthMixin, DestroyAPIView):
"""Delete a PayloadType. Fails if any PayloadLists reference it."""
serializer_class = PayloadTypeSerializer
[docs]
def get_ptype(self):
try:
return PayloadType.objects.get(name=self.kwargs['payloadTypeName'])
except PayloadType.DoesNotExist:
return None
[docs]
def get_plists(self, ptype):
try:
return PayloadList.objects.filter(payload_type=ptype)
except PayloadList.DoesNotExist:
return None
[docs]
def destroy(self, request, *args, **kwargs):
ret = {}
ptype = self.get_ptype()
if not ptype:
return Response({"detail": "PayloadType %s doesn't exist" % self.kwargs['payloadTypeName']}, status=status.HTTP_500_INTERNAL_SERVER_ERROR)
plists = list(self.get_plists(ptype))
if plists:
return Response({"detail": "PayloadType is used by %d PayloadLists" % len(plists)}, status=status.HTTP_500_INTERNAL_SERVER_ERROR)
ret = self.perform_destroy(ptype)
if not ret:
ret = {"detail": "Payload Type %s deleted." % ptype.name}
return Response(ret)
# ── PayloadList views ────────────────────────────────────────────────────────
[docs]
class PayloadListDeleteAPIView(WriteAuthMixin, DestroyAPIView):
"""Delete a PayloadList. Fails if it contains any PayloadIOVs."""
serializer_class = PayloadListSerializer
[docs]
def get_plist(self):
try:
return PayloadList.objects.get(name=self.kwargs['payloadListName'])
except PayloadList.DoesNotExist:
return None
[docs]
def get_piovs(self, plist):
try:
return PayloadIOV.objects.filter(payload_list=plist)
except PayloadIOV.DoesNotExist:
return None
[docs]
def destroy(self, request, *args, **kwargs):
ret = {}
plist = self.get_plist()
if not plist:
return Response({"detail": "PayloadList %s doesn't exist" % self.kwargs['payloadListName']}, status=status.HTTP_500_INTERNAL_SERVER_ERROR)
piovs = list(self.get_piovs(plist))
if piovs:
return Response({"detail": "PayloadList contains %d PayloadIOVs" % len(piovs)}, status=status.HTTP_500_INTERNAL_SERVER_ERROR)
ret = self.perform_destroy(plist)
if not ret:
ret = {"detail": "Payload Type %s deleted." % plist.name}
return Response(ret)
# ── List views ───────────────────────────────────────────────────────────────
# ── Status views ─────────────────────────────────────────────────────────────
# ── Creation views ───────────────────────────────────────────────────────────
[docs]
class PayloadListListCreationAPIView(WriteAuthMixin, ListCreateAPIView):
"""List all PayloadLists (GET) or create a new one (POST). Auto-generates name from PayloadType + sequence ID.
Pass ?page= to get a paginated response without nested IOVs, with
server-side search (?search=), filters (?global_tag=, ?payload_type=)
and sorting (?sort=&order=).
"""
serializer_class = PayloadListCreateSerializer
[docs]
@staticmethod
def get_next_id():
return PayloadListIdSequence.objects.create()
[docs]
def get_queryset(self):
return PayloadList.objects.all()
[docs]
def list(self, request):
if 'page' in request.query_params:
queryset = PayloadList.objects.select_related('global_tag', 'payload_type').annotate(
iov_count=Count('payload_iov'),
)
gt_name = request.query_params.get('global_tag')
if gt_name:
queryset = queryset.filter(global_tag__name=gt_name)
pt_name = request.query_params.get('payload_type')
if pt_name:
queryset = queryset.filter(payload_type__name=pt_name)
return paginate_browse(
request, queryset, PayloadListBrowseSerializer,
search_fields=('name', 'global_tag__name', 'payload_type__name'),
sort_map={
'id': 'id', 'name': 'name',
'global_tag': 'global_tag__name',
'payload_type': 'payload_type__name',
'iov_count': 'iov_count', 'created': 'created',
},
default_sort='name',
)
queryset = self.get_queryset()
serializer = PayloadListReadSerializer(queryset, many=True)
return Response(serializer.data)
[docs]
def create(self, request, *args, **kwargs):
data = request.data
next_id = self.get_next_id()
data['id'] = int(next_id)
data['name'] = data['payload_type'] + '_' + str(next_id)
try:
payload_type = PayloadType.objects.get(name=data['payload_type'])
data['payload_type'] = payload_type.pk
except KeyError:
return Response({"detail": "PayloadType not found."}, status=status.HTTP_500_INTERNAL_SERVER_ERROR)
data['global_tag'] = None
serializer = self.get_serializer(data=data)
serializer.is_valid(raise_exception=True)
try:
instance = serializer.save()
except Exception as e:
return Response({"detail": "PayloadList creation failed."}, status=status.HTTP_500_INTERNAL_SERVER_ERROR)
if instance.pk is None:
return Response({"detail": "PayloadList was not saved to DB."}, status=500)
ret = serializer.data
ret['payload_type'] = payload_type.name
return Response(ret)
[docs]
class PayloadListDetailAPIView(WriteAuthMixin, RetrieveAPIView):
serializer_class = PayloadListCreateSerializer
queryset = PayloadList.objects.all()
[docs]
class PayloadListByNameAPIView(WriteAuthMixin, RetrieveAPIView):
"""Retrieve a PayloadList by name (metadata + IOV count, no nested IOVs)."""
serializer_class = PayloadListBrowseSerializer
[docs]
def get_object(self):
return get_object_or_404(
PayloadList.objects.select_related('global_tag', 'payload_type').annotate(
iov_count=Count('payload_iov'),
),
name=self.kwargs.get('payloadListName'))
[docs]
class PayloadTypeListCreationAPIView(WriteAuthMixin, ListCreateAPIView):
"""List all PayloadTypes (GET) or create a new one (POST).
Pass ?page= to get a paginated response with server-side search
(?search=) and sorting (?sort=&order=).
"""
serializer_class = PayloadTypeSerializer
[docs]
def get_queryset(self):
return PayloadType.objects.all()
[docs]
def list(self, request):
if 'page' in request.query_params:
return paginate_browse(
request, PayloadType.objects.all(), PayloadTypeSerializer,
search_fields=('name',),
sort_map={'id': 'id', 'name': 'name', 'created': 'created'},
default_sort='name',
)
queryset = self.get_queryset()
serializer = PayloadTypeSerializer(queryset, many=True)
return Response(serializer.data)
[docs]
def create(self, request, *args, **kwargs):
serializer = self.get_serializer(data=request.data)
serializer.is_valid(raise_exception=True)
try:
instance = serializer.save()
except Exception as e:
return Response({"detail": "PayloadType creation failed."}, status=status.HTTP_500_INTERNAL_SERVER_ERROR)
if instance.pk is None:
return Response({"detail": "PayloadType was not saved to DB."}, status=500)
return Response(serializer.data)
[docs]
class PayloadIOVListCreationAPIView(WriteAuthMixin, ListCreateAPIView):
"""List all PayloadIOVs (GET) or create a new one (POST). Validates IOV ranges based on CDB_IOV_MODE.
Pass ?page= to get a paginated response of flat rows including payload
list / global tag / payload type names, with server-side search
(?search=), filters (?payload_list=, ?global_tag=, ?payload_type=)
and sorting (?sort=&order=).
"""
serializer_class = PayloadIOVSerializer
[docs]
def get_queryset(self):
return PayloadIOV.objects.all()
[docs]
def list(self, request):
if 'page' in request.query_params:
queryset = PayloadIOV.objects.select_related(
'payload_list__global_tag', 'payload_list__payload_type')
pl_name = request.query_params.get('payload_list')
if pl_name:
queryset = queryset.filter(payload_list__name=pl_name)
gt_name = request.query_params.get('global_tag')
if gt_name:
queryset = queryset.filter(payload_list__global_tag__name=gt_name)
pt_name = request.query_params.get('payload_type')
if pt_name:
queryset = queryset.filter(payload_list__payload_type__name=pt_name)
return paginate_browse(
request, queryset, PayloadIOVBrowseSerializer,
search_fields=('payload_url', 'payload_list__name'),
sort_map={
'id': 'id', 'payload_url': 'payload_url', 'checksum': 'checksum',
'major_iov': 'major_iov', 'minor_iov': 'minor_iov',
'major_iov_end': 'major_iov_end', 'minor_iov_end': 'minor_iov_end',
'payload_list': 'payload_list__name',
'global_tag': 'payload_list__global_tag__name',
'payload_type': 'payload_list__payload_type__name',
'inserted': 'inserted',
},
default_sort='id',
)
queryset = self.get_queryset()
serializer = PayloadIOVSerializer(queryset, many=True)
return Response(serializer.data)
[docs]
def create(self, request, *args, **kwargs):
iov_config = get_iov_config(settings.CDB_IOV_MODE)
data = request.data
plugin = load_permission_plugin()
target_object = {"object": "GlobalTag", "role": "createpayload", "name": data.get('payload_url', '')}
if not plugin.has_permission(request, target_object):
return Response({"detail": "Permission denied."}, status=status.HTTP_403_FORBIDDEN)
if 'major_iov_end' not in data:
data['major_iov_end'] = sys.maxsize
if 'minor_iov_end' not in data:
data['minor_iov_end'] = sys.maxsize
data['comb_iov'] = compute_comb_iov(data['major_iov'], data['minor_iov'])
if iov_config['is_invalid_iov_range'](data):
err_msg = "%s PayloadIOV ending IOVs should be greater or equal than starting." \
" Provided end IOVs: major_iov: %d major_iov_end: %d minor_iov: %d minor_iov_end: %d" % \
(data['payload_url'], data['major_iov'], data['major_iov_end'], data['minor_iov'],
data['minor_iov_end'])
return Response({"detail": err_msg}, status=status.HTTP_500_INTERNAL_SERVER_ERROR)
serializer = self.get_serializer(data=data)
serializer.is_valid(raise_exception=True)
try:
instance = serializer.save()
except Exception as e:
return Response({"detail": "PayloadIOV creation failed."}, status=status.HTTP_500_INTERNAL_SERVER_ERROR)
if instance.pk is None:
return Response({"detail": "PayloadIOV was not saved to DB."}, status=500)
ret = serializer.data
return Response(ret)
[docs]
class PayloadIOVDetailAPIView(WriteAuthMixin, RetrieveAPIView):
serializer_class = PayloadIOVSerializer
queryset = PayloadIOV.objects.all()
[docs]
class PayloadIOVBulkCreationAPIView(WriteAuthMixin, CreateAPIView):
"""Bulk-create PayloadIOVs from a JSON array. Skips individual validation for performance."""
serializer_class = PayloadIOVSerializer
[docs]
def get_queryset(self):
return PayloadIOV.objects.all()
[docs]
def create(self, request, *args, **kwargs):
data = request.data
batch = [PayloadIOV(id=None, payload_url=obj["payload_url"],
major_iov=obj["major_iov"], minor_iov=obj["minor_iov"],
major_iov_end=sys.maxsize, minor_iov_end=sys.maxsize,
payload_list=PayloadList.objects.get(name=obj['payload_list']),
inserted=None,
comb_iov=Decimal(Decimal(obj["major_iov"]) + Decimal(obj["minor_iov"]) / 10 ** 19))
for obj in data]
PayloadIOV.objects.bulk_create(batch)
return Response()
# ── Clone / Attach views ─────────────────────────────────────────────────────
[docs]
class GlobalTagCloneAPIView(WriteAuthMixin, CreateAPIView):
"""Deep-copy a GlobalTag: duplicates the GT, all its PayloadLists, and their PayloadIOVs."""
serializer_class = GlobalTagReadSerializer
[docs]
def get_global_tag(self):
source_name = self.kwargs.get('globalTagName')
return GlobalTag.objects.get(name=source_name)
[docs]
def get_clone_name(self):
return self.kwargs.get('cloneName')
[docs]
@staticmethod
def get_payload_lists(global_tag):
return PayloadList.objects.filter(global_tag=global_tag)
[docs]
@staticmethod
def get_payload_iovs(payload_list):
return PayloadIOV.objects.filter(payload_list=payload_list)
[docs]
@staticmethod
def get_next_id():
return PayloadListIdSequence.objects.create()
[docs]
@transaction.atomic
def create(self, request, globalTagName, cloneName):
plugin = load_permission_plugin()
target_object = {"object": "GlobalTag", "role": "admin", "name": self.get_clone_name()}
if not plugin.has_permission(request, target_object):
return Response({"detail": "Permission denied."}, status=status.HTTP_403_FORBIDDEN)
global_tag = self.get_global_tag()
payload_lists = self.get_payload_lists(global_tag)
global_tag.id = None
global_tag.name = self.get_clone_name()
global_tag.status = GlobalTagStatus.objects.get(name='unlocked')
serializer = GlobalTagCreateSerializer(instance=global_tag, data=model_to_dict(global_tag))
serializer.is_valid(raise_exception=True)
try:
instance = serializer.save()
except Exception as e:
return Response({"detail": "GlobalTag creation failed."}, status=status.HTTP_500_INTERNAL_SERVER_ERROR)
if instance.pk is None:
return Response({"detail": "GlobalTag was not saved to DB."}, status=500)
for p_list in payload_lists:
payload_iovs = self.get_payload_iovs(p_list)
p_list_id = self.get_next_id()
p_list.id = p_list_id
p_list.name = str(p_list.payload_type) + '_' + str(p_list_id)
p_list.global_tag = global_tag
serializer = PayloadListCreateSerializer(instance=p_list, data=model_to_dict(p_list))
serializer.is_valid(raise_exception=True)
try:
instance = serializer.save()
except Exception as e:
return Response({"detail": "PayloadList creation failed."}, status=status.HTTP_500_INTERNAL_SERVER_ERROR)
if instance.pk is None:
return Response({"detail": "PayloadList was not saved to DB."}, status=500)
rp = []
for payload in payload_iovs:
payload.id = None
payload.payload_list = p_list
rp.append(payload)
try:
PayloadIOV.objects.bulk_create(rp)
except Exception as e:
return Response({"detail": "PayloadIOV bulk creation failed."}, status=500)
serializer = GlobalTagListSerializer(global_tag)
return Response(serializer.data)
# ── Query views ──────────────────────────────────────────────────────────────
[docs]
class PayloadIOVsORMMaxListAPIView(WriteAuthMixin, ListAPIView):
"""Get latest PayloadIOVs per PayloadList for a given GT and IOV point, using ORM MAX aggregation."""
[docs]
def get_queryset(self):
gt_name = self.request.GET.get('gtName')
major_iov = self.request.GET.get('majorIOV')
minor_iov = self.request.GET.get('minorIOV')
tmp1 = PayloadIOV.objects.filter(payload_list__global_tag__name=gt_name) \
.filter(comb_iov__lte=Decimal(Decimal(major_iov) + Decimal(minor_iov) / 10 ** 19)) \
.values('payload_list_id').annotate(max_comb_iov=Max('comb_iov'))
q_statement = Q()
for pair in tmp1:
q_statement |= (Q(payload_list_id=pair['payload_list_id']) & Q(comb_iov=pair['max_comb_iov']))
return PayloadIOV.objects.filter(q_statement)
[docs]
def list(self, request):
queryset = self.get_queryset()
serializer = PayloadIOVSerializer(queryset, many=True)
return Response(serializer.data)
[docs]
class PayloadIOVsORMOrderByListAPIView(WriteAuthMixin, ListAPIView):
"""Get latest PayloadIOVs per PayloadList using ORM ORDER BY + DISTINCT."""
[docs]
def get_queryset(self):
gt_name = self.request.GET.get('gtName')
major_iov = self.request.GET.get('majorIOV')
minor_iov = self.request.GET.get('minorIOV')
queryset = PayloadIOV.objects.filter(payload_list__global_tag__name=gt_name) \
.filter(comb_iov__lte=Decimal(Decimal(major_iov) + Decimal(minor_iov) / 10 ** 19)) \
.order_by('payload_list_id', '-comb_iov').distinct('payload_list_id')
return queryset
[docs]
def list(self, request):
queryset = self.get_queryset()
serializer = PayloadIOVSerializer(queryset, many=True)
return Response(serializer.data)
def _resolve_query(setting_name, default):
"""Resolve a SQL query function by name from settings, falling back to default."""
query_name = getattr(settings, setting_name, None)
if query_name:
return getattr(cdb_rest.queries, query_name)
return default
[docs]
class PayloadIOVsSQLListAPIView(WriteAuthMixin, ListAPIView):
"""Get PayloadIOVs using raw SQL for performance. Distributes reads across read replicas."""
[docs]
def list(self, request):
read_dbs = [db for db in settings.DATABASES.keys() if db.startswith("read_db_")]
read_db = random.choice(read_dbs) if read_dbs else "default"
query = _resolve_query('CDB_PAYLOAD_IOVS_QUERY', cdb_rest.queries.get_payload_iovs)
with connections[read_db].cursor() as cursor:
cursor.execute(query,
{'my_major_iov': self.request.GET.get('majorIOV'),
'my_minor_iov': self.request.GET.get('minorIOV'),
'my_gt': self.request.GET.get('gtName')})
if self.request.GET.get('shape') == 'dict':
columns = [col[0] for col in cursor.description]
result = [dict(zip(columns, row)) for row in cursor.fetchall()]
else:
result = cursor.fetchall()
return Response(result)
[docs]
class PayloadIOVsRangesListAPIView(WriteAuthMixin, ListAPIView):
"""Get PayloadIOVs within a given IOV range, grouped by PayloadList."""
[docs]
def get_queryset(self):
gt_name = self.request.GET.get('gtName')
start_major_iov = self.request.GET.get('startMajorIOV')
start_minor_iov = self.request.GET.get('startMinorIOV')
end_major_iov = self.request.GET.get('endMajorIOV')
end_minor_iov = self.request.GET.get('endMinorIOV')
q = {'major_iov__gte': start_major_iov, 'minor_iov__gte': start_minor_iov}
if end_major_iov != '-1':
q.update({'major_iov__lte': end_major_iov})
if end_minor_iov != '-1':
q.update({'minor_iov__lte': end_minor_iov})
p_lists = PayloadList.objects.filter(global_tag__name=gt_name)
piov_ids = []
for pl in p_lists:
q.update({'payload_list': pl})
piovs = PayloadIOV.objects.filter(**q).values_list('id', flat=True)
if piovs:
piov_ids.extend(piovs)
queryset = PayloadIOV.objects.filter(id__in=piov_ids)
return PayloadList.objects.filter(global_tag__name=gt_name) \
.prefetch_related(Prefetch('payload_iov', queryset=queryset)) \
.filter(payload_iov__in=queryset).distinct()
[docs]
def list(self, request):
queryset = self.get_queryset()
serializer = PayloadListReadSerializer(queryset, many=True)
return Response(serializer.data)
[docs]
class PayloadListAttachAPIView(WriteAuthMixin, UpdateAPIView):
"""Attach a PayloadList to a GlobalTag. Detaches any existing list of the same PayloadType first."""
serializer_class = PayloadListCreateSerializer
[docs]
@transaction.atomic
def put(self, request, *args, **kwargs):
data = request.data
plugin = load_permission_plugin()
target_object = {"object": "GlobalTag", "role": "admin", "name": data['global_tag']}
if not plugin.has_permission(request, target_object):
return Response({"detail": "Permission denied."}, status=status.HTTP_403_FORBIDDEN)
try:
p_list = PayloadList.objects.get(name=data['payload_list'])
except KeyError:
return Response({"detail": "PayloadList not found."}, status=status.HTTP_500_INTERNAL_SERVER_ERROR)
try:
global_tag = GlobalTag.objects.get(name=data['global_tag'])
except KeyError:
return Response({"detail": "GlobalTag not found."}, status=status.HTTP_500_INTERNAL_SERVER_ERROR)
pl_type = p_list.payload_type
gt_status = GlobalTagStatus.objects.get(id=global_tag.status_id)
if gt_status.name == 'frozen':
return Response({"detail": "Global Tag is %s." % gt_status.name}, status=status.HTTP_500_INTERNAL_SERVER_ERROR)
if (PayloadList.objects.filter(global_tag=global_tag, payload_type=pl_type) and gt_status.name == 'locked'):
return Response({"detail": "Payload List of type %s already attached and Global Tag is locked." % pl_type}, status=status.HTTP_500_INTERNAL_SERVER_ERROR)
PayloadList.objects.filter(global_tag=global_tag, payload_type=pl_type).update(global_tag=None)
p_list.global_tag = global_tag
serializer = PayloadListCreateSerializer(instance=p_list, data=model_to_dict(p_list))
serializer.is_valid(raise_exception=True)
try:
instance = serializer.save()
except Exception as e:
return Response({"detail": "PayloadList update failed."}, status=status.HTTP_500_INTERNAL_SERVER_ERROR)
if instance.pk is None:
return Response({"detail": "PayloadList was not saved to DB."}, status=500)
serializer = GlobalTagCreateSerializer(instance=global_tag, data=model_to_dict(global_tag))
serializer.is_valid(raise_exception=True)
try:
instance = serializer.save()
except Exception as e:
return Response({"detail": "GlobalTag update failed."}, status=status.HTTP_500_INTERNAL_SERVER_ERROR)
if instance.pk is None:
return Response({"detail": "GlobalTag was not updated in the DB."}, status=500)
serializer = PayloadListSerializer(p_list)
ret = serializer.data
ret['global_tag'] = global_tag.name
ret['payload_type'] = pl_type.name
return Response(ret)
[docs]
class PayloadIOVAttachAPIView(WriteAuthMixin, UpdateAPIView):
"""
Attach a PayloadIOV to a PayloadList. Handles overlap resolution:
- Locked GT: rejects conflicting IOVs (append-only), with special case for open-ended Online GT IOVs.
- Unlocked GT: splits/trims existing IOVs to accommodate the new one.
"""
serializer_class = PayloadIOVSerializer
[docs]
@transaction.atomic
def put(self, request, *args, **kwargs):
iov_config = get_iov_config(settings.CDB_IOV_MODE)
offset = iov_config['next_iov_offset']
data = request.data
try:
p_list = PayloadList.objects.get(name=data['payload_list'])
except KeyError:
return Response({"detail": "PayloadList not found."}, status=status.HTTP_500_INTERNAL_SERVER_ERROR)
try:
piov = PayloadIOV.objects.get(id=data['piov_id'])
except KeyError:
return Response({"detail": "PayloadIOV not found."}, status=status.HTTP_500_INTERNAL_SERVER_ERROR)
plugin = load_permission_plugin()
gt_name = p_list.global_tag.name if p_list.global_tag else ''
target_object = {"object": "GlobalTag", "role": "createiov", "name": gt_name}
if not plugin.has_permission(request, target_object):
return Response({"detail": "Permission denied."}, status=status.HTTP_403_FORBIDDEN)
is_gt_locked = False
if p_list.global_tag:
gt_status = GlobalTagStatus.objects.get(id=p_list.global_tag.status_id)
if gt_status.name == 'locked':
is_gt_locked = True
elif gt_status.name == 'frozen':
return Response({"detail": "Global Tag is %s." % gt_status.name}, status=status.HTTP_500_INTERNAL_SERVER_ERROR)
list_piovs = PayloadIOV.objects.filter(payload_list=p_list)
if is_gt_locked:
piovs = list_piovs.filter(major_iov=piov.major_iov, minor_iov=piov.minor_iov)
if piovs:
payload_url, major_iov, minor_iov, major_iov_end, minor_iov_end = \
piovs.values_list('payload_url', 'major_iov', 'minor_iov', 'major_iov_end', 'minor_iov_end')[0]
err_msg = "GT is LOCKED. You are attempting to insert IOV (major_iov,minor_iov,major_iov_end, " \
"minor_iov_end) (%d,%d,%d,%d). Conflicts with existing IOV %s (%d,%d,%d,%d)" % \
(piov.major_iov, piov.minor_iov, piov.major_iov_end, piov.minor_iov_end, payload_url,
major_iov, minor_iov, major_iov_end, minor_iov_end)
return Response({"detail": err_msg}, status=status.HTTP_500_INTERNAL_SERVER_ERROR)
# Special case for Online GT - allow open IOV recover last open IOV
special_case = False
if (piov.major_iov_end == 0 or piov.major_iov_end == sys.maxsize) and piov.minor_iov_end == sys.maxsize:
piovs = list_piovs.all().order_by('-comb_iov')
if piovs:
comb_iov, major_iov_end, minor_iov_end = piovs.values_list('comb_iov', 'major_iov_end', 'minor_iov_end')[0]
if (major_iov_end == 0 or major_iov_end == sys.maxsize) and minor_iov_end == sys.maxsize:
if comb_iov < piov.comb_iov:
special_case = True
if not special_case:
piovs = list_piovs.filter(Q(major_iov__lt=piov.major_iov) |
Q(major_iov=piov.major_iov, minor_iov__lt=piov.minor_iov)) \
.order_by('-major_iov', '-minor_iov')
if piovs:
payload_url, major_iov, minor_iov, major_iov_end, minor_iov_end = \
piovs.values_list('payload_url', 'major_iov', 'minor_iov', 'major_iov_end', 'minor_iov_end')[0]
if iov_config['is_conflicting_iov_tail'](piov, major_iov_end, minor_iov_end):
err_msg = "GT is LOCKED. You are attempting to insert IOV (major_iov,minor_iov,major_iov_end, " \
"minor_iov_end) (%d,%d,%d,%d). Conflicts with existing IOV %s (%d,%d,%d,%d)" % \
(piov.major_iov, piov.minor_iov, piov.major_iov_end, piov.minor_iov_end, payload_url,
major_iov, minor_iov, major_iov_end, minor_iov_end)
return Response({"detail": err_msg}, status=status.HTTP_500_INTERNAL_SERVER_ERROR)
piovs = list_piovs.filter(Q(major_iov__gt=piov.major_iov) |
Q(major_iov=piov.major_iov, minor_iov__gt=piov.minor_iov)) \
.order_by('major_iov', 'minor_iov')
if piovs:
payload_url, major_iov, minor_iov, major_iov_end, minor_iov_end = \
piovs.values_list('payload_url', 'major_iov', 'minor_iov', 'major_iov_end', 'minor_iov_end')[0]
if iov_config['is_conflicting_iov_end'](piov, major_iov_end, minor_iov_end):
err_msg = "GT is LOCKED. You are attempting to insert IOV (major_iov,minor_iov,major_iov_end, " \
"minor_iov_end) (%d,%d,%d,%d). Conflicts with existing IOV %s (%d,%d,%d,%d)" % \
(piov.major_iov, piov.minor_iov, piov.major_iov_end, piov.minor_iov_end, payload_url,
major_iov, minor_iov, major_iov_end, minor_iov_end)
return Response({"detail": err_msg}, status=status.HTTP_500_INTERNAL_SERVER_ERROR)
else:
list_piovs.filter(
Q(major_iov__gt=piov.major_iov) | Q(major_iov=piov.major_iov, minor_iov__gte=piov.minor_iov)) \
.filter(Q(major_iov_end__lt=piov.major_iov_end) | Q(major_iov_end=piov.major_iov_end,
minor_iov_end__lte=piov.minor_iov_end)) \
.update(payload_list=None)
piovs = list_piovs.filter(
Q(major_iov__lt=piov.major_iov) | Q(major_iov=piov.major_iov, minor_iov__lte=piov.minor_iov)).order_by(
'-major_iov', '-minor_iov')
if piovs:
major_iov_end, minor_iov_end = piovs.values_list('major_iov_end', 'minor_iov_end')[0]
if iov_config['is_conflicting_iov'](piov, major_iov_end, minor_iov_end):
piovs[0].major_iov_end = piov.major_iov + offset
piovs[0].minor_iov_end = piov.minor_iov + offset
piovs[0].save(update_fields=['major_iov_end', 'minor_iov_end'])
if iov_config['is_iov_end_inside'](piov, major_iov_end, minor_iov_end):
third_piov = piovs[0]
third_piov.major_iov = piov.major_iov_end
third_piov.minor_iov = piov.minor_iov_end
third_piov.comb_iov = Decimal(Decimal(third_piov.major_iov) + Decimal(third_piov.minor_iov) / 10 ** 19)
third_piov.major_iov_end = major_iov_end + offset
third_piov.minor_iov_end = minor_iov_end + offset
third_piov.id = None
third_piov.save()
piovs = list_piovs.filter(
Q(major_iov__gt=piov.major_iov) | Q(major_iov=piov.major_iov, minor_iov__gt=piov.minor_iov)).order_by(
'major_iov', 'minor_iov')
if piovs:
major_iov, minor_iov = piovs.values_list('major_iov', 'minor_iov')[0]
if iov_config['is_conflicting_iov_end'](piov, major_iov_end, minor_iov_end):
piovs[0].major_iov = piov.major_iov_end
piovs[0].minor_iov = piov.minor_iov_end
piovs[0].comb_iov = Decimal(Decimal(piovs[0].major_iov) + Decimal(piovs[0].minor_iov) / 10 ** 19)
piovs[0].save(update_fields=['major_iov', 'minor_iov', 'comb_iov'])
piov.payload_list = p_list
piov.comb_iov = Decimal(Decimal(piov.major_iov) + Decimal(piov.minor_iov) / 10 ** 19)
serializer = PayloadIOVSerializer(instance=piov, data=model_to_dict(piov))
serializer.is_valid(raise_exception=True)
try:
instance = serializer.save()
except Exception as e:
return Response({"detail": "PayloadIOV update failed."}, status=status.HTTP_500_INTERNAL_SERVER_ERROR)
if instance.pk is None:
return Response({"detail": "PayloadIOV was not updated in the DB."}, status=500)
serializer = PayloadListCreateSerializer(instance=p_list, data=model_to_dict(p_list))
serializer.is_valid(raise_exception=True)
try:
instance = serializer.save()
except Exception as e:
return Response({"detail": "PayloadList update failed."}, status=status.HTTP_500_INTERNAL_SERVER_ERROR)
if instance.pk is None:
return Response({"detail": "PayloadList was not updated in the DB."}, status=500)
serializer = PayloadIOVSerializer(piov)
ret = serializer.data
return Response(ret)
# ── Status change views ──────────────────────────────────────────────────────
[docs]
class GlobalTagChangeStatusAPIView(WriteAuthMixin, UpdateAPIView):
"""Change a GlobalTag's status (e.g. unlocked -> locked -> frozen)."""
serializer_class = GlobalTagCreateSerializer
[docs]
def get_global_tag(self):
global_tag_name = self.kwargs.get('globalTagName')
return GlobalTag.objects.get(name=global_tag_name)
[docs]
def get_gt_status(self):
gt_status = self.kwargs.get('newStatus')
return GlobalTagStatus.objects.get(name=gt_status)
[docs]
def put(self, request, *args, **kwargs):
plugin = load_permission_plugin()
target_object = {"object": "GlobalTag", "role": "admin", "name": self.kwargs.get('globalTagName')}
if not plugin.has_permission(request, target_object):
return Response({"detail": "Permission denied."}, status=status.HTTP_403_FORBIDDEN)
try:
gt = self.get_global_tag()
except KeyError:
return Response({"detail": "GlobalTag not found."}, status=status.HTTP_500_INTERNAL_SERVER_ERROR)
try:
gt_status = self.get_gt_status()
except KeyError:
return Response({"detail": "GlobalTag Status not found."}, status=status.HTTP_500_INTERNAL_SERVER_ERROR)
gt.status = gt_status
serializer = GlobalTagCreateSerializer(instance=gt, data=model_to_dict(gt))
serializer.is_valid(raise_exception=True)
try:
instance = serializer.save()
except Exception as e:
return Response({"detail": "GlobalTag update failed."}, status=status.HTTP_500_INTERNAL_SERVER_ERROR)
if instance.pk is None:
return Response({"detail": "GlobalTag was not updated in the DB."}, status=500)
return Response(serializer.data)
# ── Settings views ───────────────────────────────────────────────────────────
[docs]
class CDBSettingAPIView(WriteAuthMixin, APIView):
"""Expose CDB_* environment variables as read-only settings."""
[docs]
def get(self, request, name):
if name not in settings.CDB_USER_SETTINGS:
return Response(
{"detail": f"Setting '{name}' not found."},
status=status.HTTP_404_NOT_FOUND
)
value = settings.CDB_USER_SETTINGS.get(name)
return Response({name: value})
[docs]
class AuthDecisionAPIView(APIView):
"""Authorization decision endpoint for nginx auth_request subrequests.
nginx passes the original request's method and URI in the
X-Original-Method and X-Original-URI headers. Reads are always allowed,
file uploads (PUT) require authentication and the permission plugin's
approval, and any other method is denied. Returns 200 to allow,
401/403 to deny; no response body is needed.
"""
[docs]
def get_authenticators(self):
if self.request and self.request.headers.get('X-Original-Method', 'GET') == 'PUT':
auth_class = load_auth_class()
if auth_class:
return [auth_class()]
return []
[docs]
def get(self, request):
method = request.headers.get('X-Original-Method', 'GET')
if method in ('GET', 'HEAD'):
return Response(status=status.HTTP_200_OK)
if method != 'PUT':
# only file uploads are allowed; deletion or anything else is denied
return Response(status=status.HTTP_403_FORBIDDEN)
plugin = load_permission_plugin()
uri = request.headers.get('X-Original-URI', '')
file_name = uri.split('?')[0].rstrip('/').rsplit('/', 1)[-1]
target_object = {"object": "PayloadStorage", "role": "admin", "name": file_name}
if not plugin.has_permission(request, target_object):
return Response(status=status.HTTP_403_FORBIDDEN)
return Response(status=status.HTTP_200_OK)
# ── Web views ────────────────────────────────────────────────────────────────
[docs]
def cdb_web_view(request):
"""Render the Conditions Database web interface."""
return render(request, 'cdb_rest/app.html')
