ross/penracourses
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Add API token management: implement APIToken model, views, and templates for creating and revoking tokens

Browse Source
This commit is contained in:
Ross
2026-02-23 11:57:57 +00:00
parent 83d1b48213
commit d7f3ebf950
8 changed files with 309 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files
atlas/api.py
+112 -17
View File
@@ -18,10 +18,13 @@ from ninja.security import django_auth
from ninja import NinjaAPI, File
from ninja.files import UploadedFile
import hashlib
import secrets
from django.utils import timezone
from generic.models import Examination, Modality
from .models import Case, DuplicateDicom, Series, SeriesImage, UncategorisedDicom
from .models import Case, DuplicateDicom, Series, SeriesImage, UncategorisedDicom, APIToken
from atlas.helpers import get_cases_available_to_user
from loguru import logger
@@ -30,6 +33,41 @@ from loguru import logger
router = Router()
class TokenAuth:
"""Simple bearer token auth for Ninja endpoints.
Integrates with `APIToken` model; use as `auth=TokenAuth()` on router
endpoints that should accept tokens in `Authorization: Bearer <token>`.
"""
def __call__(self, request):
# Extract header
auth = request.META.get("HTTP_AUTHORIZATION")
if not auth:
return None
parts = auth.split()
if len(parts) != 2 or parts[0].lower() != "bearer":
return None
token = parts[1]
token_hash = hashlib.sha256(token.encode()).hexdigest()
try:
api_token = APIToken.objects.get(token_hash=token_hash, revoked=False)
except APIToken.DoesNotExist:
return None
if api_token.expires and api_token.expires < timezone.now():
return None
# mark last used (best-effort)
try:
api_token.mark_used()
except Exception:
pass
# Attach token object to request for handlers
request.api_token = api_token
return api_token.user
class SeriesSchema(ModelSchema):
case_id: List[int] = []
@@ -54,7 +92,7 @@ class CaseSchema(ModelSchema):
fields = ["id", "title"]
@router.post("/upload_dicom", auth=django_auth)
@router.post("/upload_dicom", auth=TokenAuth())
def upload_dicom(request, files: List[UploadedFile] = File(...)):
uploaded = []
duplicate = []
@@ -87,7 +125,7 @@ def upload_dicom(request, files: List[UploadedFile] = File(...)):
}
@router.post("/generate_image_hash", auth=django_auth)
@router.post("/generate_image_hash", auth=TokenAuth())
def generate_image_hash(request, id: int):
s = SeriesImage.objects.get(pk=id)
s.generate_hashes()
@@ -99,7 +137,7 @@ def generate_image_hash(request, id: int):
}
@router.post("/clear_dicoms", auth=django_auth)
@router.post("/clear_dicoms", auth=TokenAuth())
def clear_dicoms(request):
if "selection" in request.POST:
dicoms = UncategorisedDicom.objects.filter(
@@ -114,7 +152,7 @@ def clear_dicoms(request):
return True
@router.get("/uncategorised_dicoms", auth=django_auth)
@router.get("/uncategorised_dicoms", auth=TokenAuth())
def uncategorised_dicoms(request):
dicoms = UncategorisedDicom.objects.filter(user=request.user)
@@ -213,7 +251,7 @@ def import_dicoms_helper(request, case_id: int | None = None):
@router.post(
"/import_dicoms", auth=django_auth, response=List[Tuple[SeriesSchema, str]]
"/import_dicoms", auth=TokenAuth(), response=List[Tuple[SeriesSchema, str]]
)
def import_dicoms(request):
return import_dicoms_helper(request)
@@ -221,14 +259,14 @@ def import_dicoms(request):
@router.post(
"/import_dicoms/{case_id}",
auth=django_auth,
auth=TokenAuth(),
response=List[Tuple[SeriesSchema, str]],
)
def import_dicoms_case(request, case_id: int):
return import_dicoms_helper(request, case_id=case_id)
@router.post("/upload_dicom_case/{case_id}", auth=django_auth, response=List[Tuple[SeriesSchema, str]])
@router.post("/upload_dicom_case/{case_id}", auth=TokenAuth(), response=List[Tuple[SeriesSchema, str]])
def upload_dicom_case(request, case_id: int, files: List[UploadedFile] = File(...)):
"""Upload DICOM files and immediately import them into the given case.
@@ -248,12 +286,12 @@ def upload_dicom_case(request, case_id: int, files: List[UploadedFile] = File(..
return import_dicoms_helper(request, case_id=case_id)
@router.get("/orphan_series", auth=django_auth, response=List[SeriesSchema])
@router.get("/orphan_series", auth=TokenAuth(), response=List[SeriesSchema])
def orphan_series(request):
return request.user.series.filter(case=None)
@router.get("/series_remove_duplicate_images", auth=django_auth)
@router.get("/series_remove_duplicate_images", auth=TokenAuth())
def series_remove_duplicate_images(request, series_id: int):
series = get_object_or_404(Series, pk=series_id)
@@ -270,19 +308,76 @@ def series_remove_duplicate_images(request, series_id: int):
return len(dupes)
@router.get("/get_cases_user", auth=django_auth, response=List[CaseSchema])
@router.get("/get_cases_user", auth=TokenAuth(), response=List[CaseSchema])
def get_cases_user(request):
return Case.objects.filter(author=request.user)
@router.get("/get_cases_available", auth=django_auth, response=List[CaseSchema])
@router.get("/get_cases_available", auth=TokenAuth(), response=List[CaseSchema])
def get_cases_available(request):
"""Return cases available to the authenticated user (via get_cases_available_to_user)."""
qs = get_cases_available_to_user(request.user)
return qs.order_by('-created_date')[:200]
@router.get("/check_image_hash/{hash}", auth=django_auth)
class APITokenOut(Schema):
id: int
name: str
scopes: str = ""
created: str
expires: str | None = None
revoked: bool = False
last_used: str | None = None
@router.get("/api_tokens", auth=TokenAuth(), response=List[APITokenOut])
def list_api_tokens(request):
tokens = APIToken.objects.filter(user=request.user).order_by("-created")
out = []
for t in tokens:
out.append(
{
"id": t.id,
"name": t.name,
"scopes": t.scopes,
"created": t.created.isoformat(),
"expires": t.expires.isoformat() if t.expires else None,
"revoked": t.revoked,
"last_used": t.last_used.isoformat() if t.last_used else None,
}
)
return out
class CreateTokenIn(Schema):
name: str = Field("", description="Friendly name for token")
scopes: str = Field("", description="Space-separated scopes")
expires_days: int | None = Field(None, description="Expire after N days")
@router.post("/api_tokens", auth=TokenAuth())
def create_api_token(request, payload: CreateTokenIn):
expires = None
if payload.expires_days:
expires = timezone.now() + timezone.timedelta(days=payload.expires_days)
token, obj = APIToken.create_token(request.user, name=payload.name, scopes=payload.scopes, expires=expires)
# Return raw token once — callers must store it
return {"token": token, "id": obj.id}
@router.post("/api_tokens/{token_id}/revoke", auth=TokenAuth())
def revoke_api_token(request, token_id: int):
try:
t = APIToken.objects.get(pk=token_id, user=request.user)
except APIToken.DoesNotExist:
return {"status": "not found"}
t.revoked = True
t.save(update_fields=["revoked"])
return {"status": "revoked"}
@router.get("/check_image_hash/{hash}", auth=TokenAuth())
def check_image_hash(request, hash: str):
try:
series_image = SeriesImage.objects.get(image_blake3_hash=hash)
@@ -297,7 +392,7 @@ def check_image_hash(request, hash: str):
return data
@router.post("/check_image_hashes/", auth=django_auth)
@router.post("/check_image_hashes/", auth=TokenAuth())
def check_images_hashes(request, hashes: List[str]):
"""Checks a list of image hashes and returns the series id / url if found
@@ -345,13 +440,13 @@ def check_images_hashes(request, hashes: List[str]):
# print(series_image)
@router.get("/view_dicom_tags/{hash}", auth=django_auth)
@router.get("/view_dicom_tags/{hash}", auth=TokenAuth())
def view_dicom_tags(request, hash: str):
item = SeriesImage.objects.get(image_blake3_hash=hash)
return item.get_dicom_json()
@router.get("/series_split_by_dicom_tag/{series_id}/{dicom_tag}", auth=django_auth)
@router.get("/series_split_by_dicom_tag/{series_id}/{dicom_tag}", auth=TokenAuth())
def series_split_by_tag(request, series_id: int, dicom_tag: str):
series = get_object_or_404(Series, pk=series_id)
@@ -399,7 +494,7 @@ def series_split_by_tag(request, series_id: int, dicom_tag: str):
return new_series
@router.get("/split_order_by_dicom_tag/{series_id}/{dicom_tag}", auth=django_auth)
@router.get("/split_order_by_dicom_tag/{series_id}/{dicom_tag}", auth=TokenAuth())
def series_order_by_tag(request, series_id: int, dicom_tag: str):
series = get_object_or_404(Series, pk=series_id)