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

Refactor authentication for DICOM endpoints to use TokenAuth and enhance error handling in upload process

Browse Source
This commit is contained in:
Ross
2026-03-02 10:02:17 +00:00
parent b7cf022622
commit b43d441da1
2 changed files with 160 additions and 77 deletions
Download Patch File Download Diff File Expand all files Collapse all files
atlas/api.py
+46 -42
View File
@@ -129,7 +129,7 @@ class CaseSchema(ModelSchema):
fields = ["id", "title"]
@router.post("/upload_dicom", auth=BearerAuth())
@router.post("/upload_dicom", auth=TokenAuth())
def upload_dicom(request, files: List[UploadedFile] = File(...)):
uploaded = []
duplicate = []
@@ -162,7 +162,7 @@ def upload_dicom(request, files: List[UploadedFile] = File(...)):
}
@router.post("/generate_image_hash", auth=BearerAuth())
@router.post("/generate_image_hash", auth=TokenAuth())
def generate_image_hash(request, id: int):
s = SeriesImage.objects.get(pk=id)
s.generate_hashes()
@@ -174,7 +174,7 @@ def generate_image_hash(request, id: int):
}
@router.post("/clear_dicoms", auth=BearerAuth())
@router.post("/clear_dicoms", auth=TokenAuth())
def clear_dicoms(request):
if "selection" in request.POST:
dicoms = UncategorisedDicom.objects.filter(
@@ -189,7 +189,7 @@ def clear_dicoms(request):
return True
@router.get("/uncategorised_dicoms", auth=BearerAuth())
@router.get("/uncategorised_dicoms", auth=TokenAuth())
def uncategorised_dicoms(request):
dicoms = UncategorisedDicom.objects.filter(user=request.user)
@@ -291,7 +291,7 @@ def import_dicoms_helper(request, case_id: int | None = None, dicoms=None):
@router.post(
"/import_dicoms", auth=BearerAuth(), response=List[Tuple[SeriesSchema, str]]
"/import_dicoms", auth=TokenAuth(), response=List[Tuple[SeriesSchema, str]]
)
def import_dicoms(request):
return import_dicoms_helper(request)
@@ -299,14 +299,14 @@ def import_dicoms(request):
@router.post(
"/import_dicoms/{case_id}",
auth=BearerAuth(),
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=BearerAuth())
@router.post("/upload_dicom_case/{case_id}", auth=TokenAuth())
def upload_dicom_case(request, case_id: int, files: List[UploadedFile] = File(...)):
"""Upload DICOM files and immediately import them into the given case.
@@ -360,12 +360,12 @@ def upload_dicom_case(request, case_id: int, files: List[UploadedFile] = File(..
}
@router.get("/orphan_series", auth=BearerAuth(), 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=BearerAuth())
@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)
@@ -382,12 +382,12 @@ def series_remove_duplicate_images(request, series_id: int):
return len(dupes)
@router.get("/get_cases_user", auth=BearerAuth(), 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=BearerAuth(), 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)."""
logger.error(f"Getting cases available to user {request.user}")
@@ -405,7 +405,7 @@ class APITokenOut(Schema):
last_used: str | None = None
@router.get("/api_tokens", auth=BearerAuth(), response=List[APITokenOut])
@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 = []
@@ -552,7 +552,7 @@ def token_check(request):
# return {"token": token, "id": obj.id}
@router.post("/api_tokens/{token_id}/revoke", auth=BearerAuth())
@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)
@@ -563,7 +563,7 @@ def revoke_api_token(request, token_id: int):
return {"status": "revoked"}
@router.get("/check_image_hash/{hash}", auth=BearerAuth())
@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)
@@ -578,7 +578,7 @@ def check_image_hash(request, hash: str):
return data
@router.post("/check_image_hashes/", auth=BearerAuth())
@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
@@ -586,33 +586,37 @@ def check_images_hashes(request, hashes: List[str]):
{ "hash_id": {"id": "series_id|false", "url": "series_url|false"}, ...}
"""
hash_status = {}
try:
hash_status = {}
for hash in hashes:
try:
# TOOD also check against uncategorised dicoms?
series_image = SeriesImage.objects.get(image_blake3_hash=hash)
data = {
"id": series_image.pk,
"url": series_image.series.get_absolute_url(),
"type": "series",
}
except SeriesImage.DoesNotExist:
try:
uncategorised_dicom = UncategorisedDicom.objects.get(
image_blake3_hash=hash
)
data = {
"id": uncategorised_dicom.pk,
"url": reverse("atlas:user_uploads"),
"type": "uncategorised",
}
except UncategorisedDicom.DoesNotExist:
data = {"id": False, "url": False}
for hash in hashes:
# Prefer filter().first() to avoid MultipleObjectsReturned
series_image = SeriesImage.objects.filter(image_blake3_hash=hash).select_related("series").first()
if series_image:
try:
url = series_image.series.get_absolute_url()
except Exception:
url = None
data = {"id": series_image.pk, "url": url, "type": "series"}
hash_status[hash] = data
continue
hash_status[hash] = data
uncategorised = UncategorisedDicom.objects.filter(image_blake3_hash=hash).first()
if uncategorised:
try:
uploads_url = reverse("atlas:user_uploads")
except Exception:
uploads_url = None
data = {"id": uncategorised.pk, "url": uploads_url, "type": "uncategorised"}
hash_status[hash] = data
continue
return hash_status
hash_status[hash] = {"id": False, "url": False}
return hash_status
except Exception as e:
logger.exception("check_images_hashes failed")
return JsonResponse({"detail": "Internal server error"}, status=500)
# @router.get("/generate_image_hash/{id}", auth=django_auth)
@@ -626,13 +630,13 @@ def check_images_hashes(request, hashes: List[str]):
# print(series_image)
@router.get("/view_dicom_tags/{hash}", auth=BearerAuth())
@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=BearerAuth())
@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)
if not series.check_user_can_edit(request.user):
@@ -679,7 +683,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=BearerAuth())
@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)